diff options
Diffstat (limited to 'lib/sqlalchemy/engine')
| -rw-r--r-- | lib/sqlalchemy/engine/base.py | 71 | ||||
| -rw-r--r-- | lib/sqlalchemy/engine/default.py | 104 |
2 files changed, 110 insertions, 65 deletions
diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py index 880362938..660d54604 100644 --- a/lib/sqlalchemy/engine/base.py +++ b/lib/sqlalchemy/engine/base.py @@ -528,7 +528,7 @@ class Connection(Connectable): ``contextual_connect()`` methods of Engine. """ - self.__engine = engine + self.engine = engine self.__connection = connection or engine.raw_connection() self.__transaction = None self.__close_with_result = close_with_result @@ -549,10 +549,9 @@ class Connection(Connectable): This is used to execute "sub" statements within a single execution, usually an INSERT statement. """ - return Connection(self.__engine, self.__connection, _branch=True) + return Connection(self.engine, self.__connection, _branch=True) - engine = property(lambda s:s.__engine, doc="The Engine with which this Connection is associated.") - dialect = property(lambda s:s.__engine.dialect, doc="Dialect used by this Connection.") + dialect = property(lambda s:s.engine.dialect, doc="Dialect used by this Connection.") connection = property(_get_connection, doc="The underlying DB-API connection managed by this Connection.") should_close_with_result = property(lambda s:s.__close_with_result, doc="Indicates if this Connection should be closed when a corresponding ResultProxy is closed; this is essentially an auto-release mode.") properties = property(lambda s: s._get_connection().properties, @@ -652,18 +651,18 @@ class Connection(Connectable): "Cannot start a two phase transaction when a transaction " "is already in progress.") if xid is None: - xid = self.__engine.dialect.create_xid(); + xid = self.engine.dialect.create_xid(); self.__transaction = TwoPhaseTransaction(self, xid) return self.__transaction def recover_twophase(self): - return self.__engine.dialect.do_recover_twophase(self) + return self.engine.dialect.do_recover_twophase(self) def rollback_prepared(self, xid, recover=False): - self.__engine.dialect.do_rollback_twophase(self, xid, recover=recover) + self.engine.dialect.do_rollback_twophase(self, xid, recover=recover) def commit_prepared(self, xid, recover=False): - self.__engine.dialect.do_commit_twophase(self, xid, recover=recover) + self.engine.dialect.do_commit_twophase(self, xid, recover=recover) def in_transaction(self): """Return True if a transaction is in progress.""" @@ -671,28 +670,28 @@ class Connection(Connectable): return self.__transaction is not None def _begin_impl(self): - if self.__engine._should_log_info: - self.__engine.logger.info("BEGIN") + if self.engine._should_log_info: + self.engine.logger.info("BEGIN") try: - self.__engine.dialect.do_begin(self.__connection) + self.engine.dialect.do_begin(self.__connection) except Exception, e: raise exceptions.DBAPIError.instance(None, None, e) def _rollback_impl(self): if self.__connection.is_valid: - if self.__engine._should_log_info: - self.__engine.logger.info("ROLLBACK") + if self.engine._should_log_info: + self.engine.logger.info("ROLLBACK") try: - self.__engine.dialect.do_rollback(self.__connection) + self.engine.dialect.do_rollback(self.__connection) except Exception, e: raise exceptions.DBAPIError.instance(None, None, e) self.__transaction = None def _commit_impl(self): - if self.__engine._should_log_info: - self.__engine.logger.info("COMMIT") + if self.engine._should_log_info: + self.engine.logger.info("COMMIT") try: - self.__engine.dialect.do_commit(self.__connection) + self.engine.dialect.do_commit(self.__connection) except Exception, e: raise exceptions.DBAPIError.instance(None, None, e) self.__transaction = None @@ -702,38 +701,38 @@ class Connection(Connectable): self.__savepoint_seq += 1 name = 'sa_savepoint_%s' % self.__savepoint_seq if self.__connection.is_valid: - self.__engine.dialect.do_savepoint(self, name) + self.engine.dialect.do_savepoint(self, name) return name def _rollback_to_savepoint_impl(self, name, context): if self.__connection.is_valid: - self.__engine.dialect.do_rollback_to_savepoint(self, name) + self.engine.dialect.do_rollback_to_savepoint(self, name) self.__transaction = context def _release_savepoint_impl(self, name, context): if self.__connection.is_valid: - self.__engine.dialect.do_release_savepoint(self, name) + self.engine.dialect.do_release_savepoint(self, name) self.__transaction = context def _begin_twophase_impl(self, xid): if self.__connection.is_valid: - self.__engine.dialect.do_begin_twophase(self, xid) + self.engine.dialect.do_begin_twophase(self, xid) def _prepare_twophase_impl(self, xid): if self.__connection.is_valid: assert isinstance(self.__transaction, TwoPhaseTransaction) - self.__engine.dialect.do_prepare_twophase(self, xid) + self.engine.dialect.do_prepare_twophase(self, xid) def _rollback_twophase_impl(self, xid, is_prepared): if self.__connection.is_valid: assert isinstance(self.__transaction, TwoPhaseTransaction) - self.__engine.dialect.do_rollback_twophase(self, xid, is_prepared) + self.engine.dialect.do_rollback_twophase(self, xid, is_prepared) self.__transaction = None def _commit_twophase_impl(self, xid, is_prepared): if self.__connection.is_valid: assert isinstance(self.__transaction, TwoPhaseTransaction) - self.__engine.dialect.do_commit_twophase(self, xid, is_prepared) + self.engine.dialect.do_commit_twophase(self, xid, is_prepared) self.__transaction = None def _autocommit(self, context): @@ -784,7 +783,7 @@ class Connection(Connectable): raise exceptions.InvalidRequestError("Unexecutable object type: " + str(type(object))) def _execute_default(self, default, multiparams=None, params=None): - return self.__engine.dialect.defaultrunner(self.__create_execution_context()).traverse_single(default) + return self.engine.dialect.defaultrunner(self.__create_execution_context()).traverse_single(default) def _execute_text(self, statement, multiparams, params): parameters = self.__distill_params(multiparams, params) @@ -848,7 +847,7 @@ class Connection(Connectable): return context.result() def __create_execution_context(self, **kwargs): - return self.__engine.dialect.create_execution_context(connection=self, **kwargs) + return self.engine.dialect.create_execution_context(connection=self, **kwargs) def __execute_raw(self, context): if context.executemany: @@ -857,9 +856,9 @@ class Connection(Connectable): self._cursor_execute(context.cursor, context.statement, context.parameters[0], context=context) def _cursor_execute(self, cursor, statement, parameters, context=None): - if self.__engine._should_log_info: - self.__engine.logger.info(statement) - self.__engine.logger.info(repr(parameters)) + if self.engine._should_log_info: + self.engine.logger.info(statement) + self.engine.logger.info(repr(parameters)) try: self.dialect.do_execute(cursor, statement, parameters, context=context) except Exception, e: @@ -873,9 +872,9 @@ class Connection(Connectable): raise exceptions.DBAPIError.instance(statement, parameters, e) def _cursor_executemany(self, cursor, statement, parameters, context=None): - if self.__engine._should_log_info: - self.__engine.logger.info(statement) - self.__engine.logger.info(repr(parameters)) + if self.engine._should_log_info: + self.engine.logger.info(statement) + self.engine.logger.info(repr(parameters)) try: self.dialect.do_executemany(cursor, statement, parameters, context=context) except Exception, e: @@ -900,20 +899,20 @@ class Connection(Connectable): def create(self, entity, **kwargs): """Create a Table or Index given an appropriate Schema object.""" - return self.__engine.create(entity, connection=self, **kwargs) + return self.engine.create(entity, connection=self, **kwargs) def drop(self, entity, **kwargs): """Drop a Table or Index given an appropriate Schema object.""" - return self.__engine.drop(entity, connection=self, **kwargs) + return self.engine.drop(entity, connection=self, **kwargs) def reflecttable(self, table, include_columns=None): """Reflect the columns in the given string table name from the database.""" - return self.__engine.reflecttable(table, self, include_columns) + return self.engine.reflecttable(table, self, include_columns) def default_schema_name(self): - return self.__engine.dialect.get_default_schema_name(self) + return self.engine.dialect.get_default_schema_name(self) def run_callable(self, callable_): return callable_(self) diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py index c98519ffe..771ca06f9 100644 --- a/lib/sqlalchemy/engine/default.py +++ b/lib/sqlalchemy/engine/default.py @@ -136,15 +136,32 @@ class DefaultExecutionContext(base.ExecutionContext): self.dialect = dialect self._connection = self.root_connection = connection self.compiled = compiled - self._postfetch_cols = util.Set() self.engine = connection.engine if compiled is not None: + # compiled clauseelement. process bind params, process table defaults, + # track collections used by ResultProxy to target and process results + + self.processors = dict([ + (key, value) for key, value in + [( + compiled.bind_names[bindparam], + bindparam.bind_processor(self.dialect) + ) for bindparam in compiled.bind_names] + if value is not None + ]) + self.typemap = compiled.typemap self.column_labels = compiled.column_labels - self.statement = unicode(compiled) + + if not dialect.supports_unicode_statements: + self.statement = unicode(compiled).encode(self.dialect.encoding) + else: + self.statement = unicode(compiled) + self.isinsert = compiled.isinsert self.isupdate = compiled.isupdate + if not parameters: self.compiled_parameters = [compiled.construct_params()] self.executemany = False @@ -152,20 +169,26 @@ class DefaultExecutionContext(base.ExecutionContext): self.compiled_parameters = [compiled.construct_params(m) for m in parameters] self.executemany = len(parameters) > 1 + self.cursor = self.create_cursor() + self.__process_defaults() + self.parameters = self.__convert_compiled_params(self.compiled_parameters) + elif statement is not None: + # plain text statement. self.typemap = self.column_labels = None self.parameters = self.__encode_param_keys(parameters) self.executemany = len(parameters) > 1 - self.statement = statement + if not dialect.supports_unicode_statements: + self.statement = statement.encode(self.dialect.encoding) + else: + self.statement = statement self.isinsert = self.isupdate = False + self.cursor = self.create_cursor() else: + # no statement. used for standalone ColumnDefault execution. self.statement = None self.isinsert = self.isupdate = self.executemany = False - - if self.statement is not None and not dialect.supports_unicode_statements: - self.statement = self.statement.encode(self.dialect.encoding) - - self.cursor = self.create_cursor() + self.cursor = self.create_cursor() connection = property(lambda s:s._connection._branch()) @@ -190,13 +213,40 @@ class DefaultExecutionContext(base.ExecutionContext): 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, parameters): - processors = parameters[0].get_processors() + 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: - parameters = [p.get_raw_list(processors) for p in parameters] + 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 - parameters = [p.get_raw_dict(processors, encode_keys=encode) for p in parameters] + 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 is_select(self): @@ -220,8 +270,7 @@ class DefaultExecutionContext(base.ExecutionContext): return AUTOCOMMIT_REGEXP.match(self.statement) def pre_exec(self): - self._process_defaults() - self.parameters = self.__convert_compiled_params(self.compiled_parameters) + pass def post_exec(self): pass @@ -251,7 +300,7 @@ class DefaultExecutionContext(base.ExecutionContext): return self._last_updated_params def lastrow_has_defaults(self): - return len(self._postfetch_cols) + return hasattr(self, '_postfetch_cols') and len(self._postfetch_cols) def postfetch_cols(self): return self._postfetch_cols @@ -282,7 +331,7 @@ class DefaultExecutionContext(base.ExecutionContext): inputsizes[key.encode(self.dialect.encoding)] = dbtype self.cursor.setinputsizes(**inputsizes) - def _process_defaults(self): + def __process_defaults(self): """generate default values for compiled insert/update statements, and generate last_inserted_ids() collection.""" @@ -292,6 +341,9 @@ class DefaultExecutionContext(base.ExecutionContext): 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: @@ -299,32 +351,26 @@ class DefaultExecutionContext(base.ExecutionContext): else: val = drunner.get_column_onupdate(c) if val is not None: - param.set_value(c.key, val) + param[c.key] = val self.compiled_parameters = params else: compiled_parameters = self.compiled_parameters[0] drunner = self.dialect.defaultrunner(self) - if self.isinsert: - self._last_inserted_ids = [] + 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.set_value(c.key, val) + compiled_parameters[c.key] = val if self.isinsert: - processors = compiled_parameters.get_processors() - for c in self.compiled.statement.table.primary_key: - if c.key in compiled_parameters: - self._last_inserted_ids.append(compiled_parameters.get_processed(c.key, processors)) - else: - self._last_inserted_ids.append(None) - - self._postfetch_cols = self.compiled.postfetch - 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 |