default.py

SQLAlchemy. 经典的Python ORM框架。学习必看。

    connection = property(lambda s:s._connection._branch())

    def __encode_param_keys(self, params):
        """apply string encoding to the keys of dictionary-based bind parameters.

        This is only used executing textual, non-compiled SQL expressions."""

        if self.dialect.positional or self.dialect.supports_unicode_statements:
            if params:
                return params
            elif self.dialect.positional:
                return [()]
            else:
                return [{}]
        else:
            def proc(d):
                # sigh, sometimes we get positional arguments with a dialect
                # that doesnt specify positional (because of execute_text())
                if not isinstance(d, dict):
                    return d
                return dict([(k.encode(self.dialect.encoding), d[k]) for k in d])
            return [proc(d) for d in params] or [{}]

    def __convert_compiled_params(self, compiled_parameters):
        """convert the dictionary of bind parameter values into a dict or list
        to be sent to the DBAPI's execute() or executemany() method.
        """

        processors = self.processors
        parameters = []
        if self.dialect.positional:
            for compiled_params in compiled_parameters:
                param = []
                for key in self.compiled.positiontup:
                    if key in processors:
                        param.append(processors[key](compiled_params[key]))
                    else:
                        param.append(compiled_params[key])
                parameters.append(param)
        else:
            encode = not self.dialect.supports_unicode_statements
            for compiled_params in compiled_parameters:
                param = {}
                if encode:
                    encoding = self.dialect.encoding
                    for key in compiled_params:
                        if key in processors:
                            param[key.encode(encoding)] = processors[key](compiled_params[key])
                        else:
                            param[key.encode(encoding)] = compiled_params[key]
                else:
                    for key in compiled_params:
                        if key in processors:
                            param[key] = processors[key](compiled_params[key])
                        else:
                            param[key] = compiled_params[key]
                parameters.append(param)
        return parameters

    def returns_rows_compiled(self, compiled):
        return isinstance(compiled.statement, expression.Selectable)

    def returns_rows_text(self, statement):
        return SELECT_REGEXP.match(statement)

    def should_autocommit_compiled(self, compiled):
        return isinstance(compiled.statement, expression._UpdateBase)

    def should_autocommit_text(self, statement):
        return AUTOCOMMIT_REGEXP.match(statement)


    def create_cursor(self):
        return self._connection.connection.cursor()

    def pre_execution(self):
        self.pre_exec()

    def post_execution(self):
        self.post_exec()

    def result(self):
        return self.get_result_proxy()

    def pre_exec(self):
        pass

    def post_exec(self):
        pass

    def get_result_proxy(self):
        return base.ResultProxy(self)

    def get_rowcount(self):
        if hasattr(self, '_rowcount'):
            return self._rowcount
        else:
            return self.cursor.rowcount

    def supports_sane_rowcount(self):
        return self.dialect.supports_sane_rowcount

    def supports_sane_multi_rowcount(self):
        return self.dialect.supports_sane_multi_rowcount

    def last_inserted_ids(self):
        return self._last_inserted_ids

    def last_inserted_params(self):
        return self._last_inserted_params

    def last_updated_params(self):
        return self._last_updated_params

    def lastrow_has_defaults(self):
        return hasattr(self, 'postfetch_cols') and len(self.postfetch_cols)

    def set_input_sizes(self):
        """Given a cursor and ClauseParameters, call the appropriate
        style of ``setinputsizes()`` on the cursor, using DB-API types
        from the bind parameter's ``TypeEngine`` objects.
        """

        types = dict([
                (self.compiled.bind_names[bindparam], bindparam.type)
                 for bindparam in self.compiled.bind_names
            ])

        if self.dialect.positional:
            inputsizes = []
            for key in self.compiled.positiontup:
               typeengine = types[key]
               dbtype = typeengine.dialect_impl(self.dialect).get_dbapi_type(self.dialect.dbapi)
               if dbtype is not None:
                    inputsizes.append(dbtype)
            try:
                self.cursor.setinputsizes(*inputsizes)
            except Exception, e:
                self._connection._handle_dbapi_exception(e, None, None, None)
                raise
        else:
            inputsizes = {}
            for key in self.compiled.bind_names.values():
                typeengine = types[key]
                dbtype = typeengine.dialect_impl(self.dialect).get_dbapi_type(self.dialect.dbapi)
                if dbtype is not None:
                    inputsizes[key.encode(self.dialect.encoding)] = dbtype
            try:
                self.cursor.setinputsizes(**inputsizes)
            except Exception, e:
                self._connection._handle_dbapi_exception(e, None, None, None)
                raise

    def __process_defaults(self):
        """generate default values for compiled insert/update statements,
        and generate last_inserted_ids() collection."""

        if self.isinsert or self.isupdate:
            if self.executemany:
                if len(self.compiled.prefetch):
                    drunner = self.dialect.defaultrunner(self)
                    params = self.compiled_parameters
                    for param in params:
                        # assign each dict of params to self.compiled_parameters;
                        # this allows user-defined default generators to access the full
                        # set of bind params for the row
                        self.compiled_parameters = param
                        for c in self.compiled.prefetch:
                            if self.isinsert:
                                val = drunner.get_column_default(c)
                            else:
                                val = drunner.get_column_onupdate(c)
                            if val is not None:
                                param[c.key] = val
                    self.compiled_parameters = params

            else:
                compiled_parameters = self.compiled_parameters[0]
                drunner = self.dialect.defaultrunner(self)

                for c in self.compiled.prefetch:
                    if self.isinsert:
                        val = drunner.get_column_default(c)
                    else:
                        val = drunner.get_column_onupdate(c)

                    if val is not None:
                        compiled_parameters[c.key] = val

                if self.isinsert:
                    self._last_inserted_ids = [compiled_parameters.get(c.key, None) for c in self.compiled.statement.table.primary_key]
                    self._last_inserted_params = compiled_parameters
                else:
                    self._last_updated_params = compiled_parameters

                self.postfetch_cols = self.compiled.postfetch