1. 模块列表
  2. 函数列表
  3. sqlalchemy.schema._DDLCompiles()

Python sqlalchemy.schema 模块，_DDLCompiles() 实例源码

我们从Python开源项目中，提取了以下26个代码示例，用于说明如何使用sqlalchemy.schema._DDLCompiles()。

项目：Flask_Blog    作者：sugarguo    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(dialect=compare_dialect)
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline)
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：QXSConsolas    作者：qxsch    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(dialect=compare_dialect)
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline)
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：chihu    作者：yelongyu    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(dialect=compare_dialect)
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline)
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：ShelbySearch    作者：Agentscreech    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    schema_translate_map=context.
                    execution_options.get('schema_translate_map'))
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline,
                    schema_translate_map=context.
                    execution_options.get('schema_translate_map'))
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：Price-Comparator    作者：Thejas-1    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    schema_translate_map=context.
                    execution_options.get('schema_translate_map'))
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline,
                    schema_translate_map=context.
                    execution_options.get('schema_translate_map'))
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：Flask-NvRay-Blog    作者：rui7157    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(dialect=compare_dialect)
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline)
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：Flask-NvRay-Blog    作者：rui7157    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(dialect=compare_dialect)
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline)
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：Callandtext    作者：iaora    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(dialect=compare_dialect)
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline)
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：python_ddd_flask    作者：igorvinnicius    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(dialect=compare_dialect)
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline)
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：webapp    作者：superchilli    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    schema_translate_map=context.
                    execution_options.get('schema_translate_map'))
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline,
                    schema_translate_map=context.
                    execution_options.get('schema_translate_map'))
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：QualquerMerdaAPI    作者：tiagovizoto    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    schema_translate_map=context.
                    execution_options.get('schema_translate_map'))
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline,
                    schema_translate_map=context.
                    execution_options.get('schema_translate_map'))
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：gardenbot    作者：GoestaO    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    schema_translate_map=context.
                    execution_options.get('schema_translate_map'))
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline,
                    schema_translate_map=context.
                    execution_options.get('schema_translate_map'))
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：flask-zhenai-mongo-echarts    作者：Fretice    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    schema_translate_map=context.
                    execution_options.get('schema_translate_map'))
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline,
                    schema_translate_map=context.
                    execution_options.get('schema_translate_map'))
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：Data-visualization    作者：insta-code1    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(dialect=compare_dialect)
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline)
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：micro-blog    作者：nickChenyx    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(dialect=compare_dialect)
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline)
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：python-flask-security    作者：weinbergdavid    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(dialect=compare_dialect)
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline)
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：watcher    作者：nosmokingbandit    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    schema_translate_map=context.
                    execution_options.get('schema_translate_map'))
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline,
                    schema_translate_map=context.
                    execution_options.get('schema_translate_map'))
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：Lixiang_zhaoxin    作者：hejaxian    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(dialect=compare_dialect)
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline)
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：flask    作者：bobohope    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    schema_translate_map=context.
                    execution_options.get('schema_translate_map'))
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline,
                    schema_translate_map=context.
                    execution_options.get('schema_translate_map'))
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：Chorus    作者：DonaldBough    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    schema_translate_map=context.
                    execution_options.get('schema_translate_map'))
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline,
                    schema_translate_map=context.
                    execution_options.get('schema_translate_map'))
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：Hawkeye    作者：tozhengxq    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(dialect=compare_dialect)
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline)
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：Alfred    作者：jkachhadia    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(dialect=compare_dialect)
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline)
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：ngx_status    作者：YoYoAdorkable    | 项目源码 | 文件源码 
def _received_statement(self, execute_observed):
        """reconstruct the statement and params in terms
        of a target dialect, which for CompiledSQL is just DefaultDialect."""

        context = execute_observed.context
        compare_dialect = self._compile_dialect(execute_observed)
        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(dialect=compare_dialect)
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=compare_dialect,
                    column_keys=context.compiled.column_keys,
                    inline=context.compiled.inline)
            )
        _received_statement = re.sub(r'[\n\t]', '', util.text_type(compiled))
        parameters = execute_observed.parameters

        if not parameters:
            _received_parameters = [compiled.construct_params()]
        else:
            _received_parameters = [
                compiled.construct_params(m) for m in parameters]

        return _received_statement, _received_parameters
项目：flasky    作者：RoseOu    | 项目源码 | 文件源码 
def process_cursor_execute(self, statement, parameters, context,
                               executemany):
        if not context:
            return
        from sqlalchemy.schema import _DDLCompiles
        _received_parameters = list(context.compiled_parameters)

        # recompile from the context, using the default dialect

        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(dialect=DefaultDialect())
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=DefaultDialect(),
                    column_keys=context.compiled.column_keys)
            )
        _received_statement = re.sub(r'[\n\t]', '', str(compiled))
        equivalent = self.statement == _received_statement
        if self.params:
            if util.callable(self.params):
                params = self.params(context)
            else:
                params = self.params
            if not isinstance(params, list):
                params = [params]
            else:
                params = list(params)
            all_params = list(params)
            all_received = list(_received_parameters)
            while params:
                param = dict(params.pop(0))
                for k, v in context.compiled.params.items():
                    param.setdefault(k, v)
                if param not in _received_parameters:
                    equivalent = False
                    break
                else:
                    _received_parameters.remove(param)
            if _received_parameters:
                equivalent = False
        else:
            params = {}
            all_params = {}
            all_received = []
        self._result = equivalent
        if not self._result:
            print('Testing for compiled statement %r partial params '
                  '%r, received %r with params %r' %
                  (self.statement, all_params,
                   _received_statement, all_received))
            self._errmsg = (
                'Testing for compiled statement %r partial params %r, '
                'received %r with params %r' %
                (self.statement, all_params,
                 _received_statement, all_received))

            # print self._errmsg
项目：oa_qian    作者：sunqb    | 项目源码 | 文件源码 
def process_cursor_execute(self, statement, parameters, context,
                               executemany):
        if not context:
            return
        from sqlalchemy.schema import _DDLCompiles
        _received_parameters = list(context.compiled_parameters)

        # recompile from the context, using the default dialect

        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(dialect=DefaultDialect())
        else:
            compiled = (
                context.compiled.statement.compile(
                    dialect=DefaultDialect(),
                    column_keys=context.compiled.column_keys)
            )
        _received_statement = re.sub(r'[\n\t]', '', str(compiled))
        equivalent = self.statement == _received_statement
        if self.params:
            if util.callable(self.params):
                params = self.params(context)
            else:
                params = self.params
            if not isinstance(params, list):
                params = [params]
            else:
                params = list(params)
            all_params = list(params)
            all_received = list(_received_parameters)
            while params:
                param = dict(params.pop(0))
                for k, v in context.compiled.params.items():
                    param.setdefault(k, v)
                if param not in _received_parameters:
                    equivalent = False
                    break
                else:
                    _received_parameters.remove(param)
            if _received_parameters:
                equivalent = False
        else:
            params = {}
            all_params = {}
            all_received = []
        self._result = equivalent
        if not self._result:
            print('Testing for compiled statement %r partial params '
                  '%r, received %r with params %r' %
                  (self.statement, all_params,
                   _received_statement, all_received))
            self._errmsg = (
                'Testing for compiled statement %r partial params %r, '
                'received %r with params %r' %
                (self.statement, all_params,
                 _received_statement, all_received))

            # print self._errmsg
项目：pyetje    作者：rorlika    | 项目源码 | 文件源码 
def process_cursor_execute(self, statement, parameters, context,
                               executemany):
        if not context:
            return
        from sqlalchemy.schema import _DDLCompiles
        _received_parameters = list(context.compiled_parameters)

        # recompile from the context, using the default dialect

        if isinstance(context.compiled.statement, _DDLCompiles):
            compiled = \
                context.compiled.statement.compile(dialect=DefaultDialect())
        else:
            compiled = \
                context.compiled.statement.compile(dialect=DefaultDialect(),
                column_keys=context.compiled.column_keys)
        _received_statement = re.sub(r'[\n\t]', '', str(compiled))
        equivalent = self.statement == _received_statement
        if self.params:
            if util.callable(self.params):
                params = self.params(context)
            else:
                params = self.params
            if not isinstance(params, list):
                params = [params]
            else:
                params = list(params)
            all_params = list(params)
            all_received = list(_received_parameters)
            while params:
                param = dict(params.pop(0))
                for k, v in context.compiled.params.iteritems():
                    param.setdefault(k, v)
                if param not in _received_parameters:
                    equivalent = False
                    break
                else:
                    _received_parameters.remove(param)
            if _received_parameters:
                equivalent = False
        else:
            params = {}
            all_params = {}
            all_received = []
        self._result = equivalent
        if not self._result:
            print 'Testing for compiled statement %r partial params '\
                '%r, received %r with params %r' % (self.statement,
                    all_params, _received_statement, all_received)
            self._errmsg = \
                'Testing for compiled statement %r partial params %r, '\
                'received %r with params %r' % (self.statement,
                    all_params, _received_statement, all_received)


            # print self._errmsg