diff options
Diffstat (limited to 'lib/sqlalchemy/engine/base.py')
| -rw-r--r-- | lib/sqlalchemy/engine/base.py | 3450 |
1 files changed, 3450 insertions, 0 deletions
diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py new file mode 100644 index 0000000..f126eb0 --- /dev/null +++ b/lib/sqlalchemy/engine/base.py @@ -0,0 +1,3450 @@ +# engine/base.py +# Copyright (C) 2005-2022 the SQLAlchemy authors and contributors +# <see AUTHORS file> +# +# This module is part of SQLAlchemy and is released under +# the MIT License: https://www.opensource.org/licenses/mit-license.php +from __future__ import with_statement + +import contextlib +import sys + +from .interfaces import Connectable +from .interfaces import ExceptionContext +from .util import _distill_params +from .util import _distill_params_20 +from .util import TransactionalContext +from .. import exc +from .. import inspection +from .. import log +from .. import util +from ..sql import compiler +from ..sql import util as sql_util + + +"""Defines :class:`_engine.Connection` and :class:`_engine.Engine`. + +""" + +_EMPTY_EXECUTION_OPTS = util.immutabledict() + + +class Connection(Connectable): + """Provides high-level functionality for a wrapped DB-API connection. + + **This is the SQLAlchemy 1.x.x version** of the :class:`_engine.Connection` + class. For the :term:`2.0 style` version, which features some API + differences, see :class:`_future.Connection`. + + The :class:`_engine.Connection` object is procured by calling + the :meth:`_engine.Engine.connect` method of the :class:`_engine.Engine` + object, and provides services for execution of SQL statements as well + as transaction control. + + The Connection object is **not** thread-safe. While a Connection can be + shared among threads using properly synchronized access, it is still + possible that the underlying DBAPI connection may not support shared + access between threads. Check the DBAPI documentation for details. + + The Connection object represents a single DBAPI connection checked out + from the connection pool. In this state, the connection pool has no affect + upon the connection, including its expiration or timeout state. For the + connection pool to properly manage connections, connections should be + returned to the connection pool (i.e. ``connection.close()``) whenever the + connection is not in use. + + .. index:: + single: thread safety; Connection + + """ + + _is_future = False + _sqla_logger_namespace = "sqlalchemy.engine.Connection" + + # used by sqlalchemy.engine.util.TransactionalContext + _trans_context_manager = None + + def __init__( + self, + engine, + connection=None, + close_with_result=False, + _branch_from=None, + _execution_options=None, + _dispatch=None, + _has_events=None, + _allow_revalidate=True, + ): + """Construct a new Connection.""" + self.engine = engine + self.dialect = engine.dialect + self.__branch_from = _branch_from + + if _branch_from: + # branching is always "from" the root connection + assert _branch_from.__branch_from is None + self._dbapi_connection = connection + self._execution_options = _execution_options + self._echo = _branch_from._echo + self.should_close_with_result = False + self.dispatch = _dispatch + self._has_events = _branch_from._has_events + else: + self._dbapi_connection = ( + connection + if connection is not None + else engine.raw_connection() + ) + + self._transaction = self._nested_transaction = None + self.__savepoint_seq = 0 + self.__in_begin = False + self.should_close_with_result = close_with_result + + self.__can_reconnect = _allow_revalidate + self._echo = self.engine._should_log_info() + + if _has_events is None: + # if _has_events is sent explicitly as False, + # then don't join the dispatch of the engine; we don't + # want to handle any of the engine's events in that case. + self.dispatch = self.dispatch._join(engine.dispatch) + self._has_events = _has_events or ( + _has_events is None and engine._has_events + ) + + assert not _execution_options + self._execution_options = engine._execution_options + + if self._has_events or self.engine._has_events: + self.dispatch.engine_connect(self, _branch_from is not None) + + @util.memoized_property + def _message_formatter(self): + if "logging_token" in self._execution_options: + token = self._execution_options["logging_token"] + return lambda msg: "[%s] %s" % (token, msg) + else: + return None + + def _log_info(self, message, *arg, **kw): + fmt = self._message_formatter + + if fmt: + message = fmt(message) + + if log.STACKLEVEL: + kw["stacklevel"] = 1 + log.STACKLEVEL_OFFSET + + self.engine.logger.info(message, *arg, **kw) + + def _log_debug(self, message, *arg, **kw): + fmt = self._message_formatter + + if fmt: + message = fmt(message) + + if log.STACKLEVEL: + kw["stacklevel"] = 1 + log.STACKLEVEL_OFFSET + + self.engine.logger.debug(message, *arg, **kw) + + @property + def _schema_translate_map(self): + return self._execution_options.get("schema_translate_map", None) + + def schema_for_object(self, obj): + """Return the schema name for the given schema item taking into + account current schema translate map. + + """ + + name = obj.schema + schema_translate_map = self._execution_options.get( + "schema_translate_map", None + ) + + if ( + schema_translate_map + and name in schema_translate_map + and obj._use_schema_map + ): + return schema_translate_map[name] + else: + return name + + def _branch(self): + """Return a new Connection which references this Connection's + engine and connection; but does not have close_with_result enabled, + and also whose close() method does nothing. + + .. deprecated:: 1.4 the "branching" concept will be removed in + SQLAlchemy 2.0 as well as the "Connection.connect()" method which + is the only consumer for this. + + The Core uses this very sparingly, only in the case of + custom SQL default functions that are to be INSERTed as the + primary key of a row where we need to get the value back, so we have + to invoke it distinctly - this is a very uncommon case. + + Userland code accesses _branch() when the connect() + method is called. The branched connection + acts as much as possible like the parent, except that it stays + connected when a close() event occurs. + + """ + return self.engine._connection_cls( + self.engine, + self._dbapi_connection, + _branch_from=self.__branch_from if self.__branch_from else self, + _execution_options=self._execution_options, + _has_events=self._has_events, + _dispatch=self.dispatch, + ) + + def _generate_for_options(self): + """define connection method chaining behavior for execution_options""" + + if self._is_future: + return self + else: + c = self.__class__.__new__(self.__class__) + c.__dict__ = self.__dict__.copy() + return c + + def __enter__(self): + return self + + def __exit__(self, type_, value, traceback): + self.close() + + def execution_options(self, **opt): + r""" Set non-SQL options for the connection which take effect + during execution. + + For a "future" style connection, this method returns this same + :class:`_future.Connection` object with the new options added. + + For a legacy connection, this method returns a copy of this + :class:`_engine.Connection` which references the same underlying DBAPI + connection, but also defines the given execution options which will + take effect for a call to + :meth:`execute`. As the new :class:`_engine.Connection` references the + same underlying resource, it's usually a good idea to ensure that + the copies will be discarded immediately, which is implicit if used + as in:: + + result = connection.execution_options(stream_results=True).\ + execute(stmt) + + Note that any key/value can be passed to + :meth:`_engine.Connection.execution_options`, + and it will be stored in the + ``_execution_options`` dictionary of the :class:`_engine.Connection`. + It + is suitable for usage by end-user schemes to communicate with + event listeners, for example. + + The keywords that are currently recognized by SQLAlchemy itself + include all those listed under :meth:`.Executable.execution_options`, + as well as others that are specific to :class:`_engine.Connection`. + + :param autocommit: Available on: Connection, statement. + When True, a COMMIT will be invoked after execution + when executed in 'autocommit' mode, i.e. when an explicit + transaction is not begun on the connection. Note that this + is **library level, not DBAPI level autocommit**. The DBAPI + connection will remain in a real transaction unless the + "AUTOCOMMIT" isolation level is used. + + .. deprecated:: 1.4 The "autocommit" execution option is deprecated + and will be removed in SQLAlchemy 2.0. See + :ref:`migration_20_autocommit` for discussion. + + :param compiled_cache: Available on: Connection. + A dictionary where :class:`.Compiled` objects + will be cached when the :class:`_engine.Connection` + compiles a clause + expression into a :class:`.Compiled` object. This dictionary will + supersede the statement cache that may be configured on the + :class:`_engine.Engine` itself. If set to None, caching + is disabled, even if the engine has a configured cache size. + + Note that the ORM makes use of its own "compiled" caches for + some operations, including flush operations. The caching + used by the ORM internally supersedes a cache dictionary + specified here. + + :param logging_token: Available on: :class:`_engine.Connection`, + :class:`_engine.Engine`. + + Adds the specified string token surrounded by brackets in log + messages logged by the connection, i.e. the logging that's enabled + either via the :paramref:`_sa.create_engine.echo` flag or via the + ``logging.getLogger("sqlalchemy.engine")`` logger. This allows a + per-connection or per-sub-engine token to be available which is + useful for debugging concurrent connection scenarios. + + .. versionadded:: 1.4.0b2 + + .. seealso:: + + :ref:`dbengine_logging_tokens` - usage example + + :paramref:`_sa.create_engine.logging_name` - adds a name to the + name used by the Python logger object itself. + + :param isolation_level: Available on: :class:`_engine.Connection`. + + Set the transaction isolation level for the lifespan of this + :class:`_engine.Connection` object. + Valid values include those string + values accepted by the :paramref:`_sa.create_engine.isolation_level` + parameter passed to :func:`_sa.create_engine`. These levels are + semi-database specific; see individual dialect documentation for + valid levels. + + The isolation level option applies the isolation level by emitting + statements on the DBAPI connection, and **necessarily affects the + original Connection object overall**, not just the copy that is + returned by the call to :meth:`_engine.Connection.execution_options` + method. The isolation level will remain at the given setting until + the DBAPI connection itself is returned to the connection pool, i.e. + the :meth:`_engine.Connection.close` method on the original + :class:`_engine.Connection` is called, + where an event handler will emit + additional statements on the DBAPI connection in order to revert the + isolation level change. + + .. warning:: The ``isolation_level`` execution option should + **not** be used when a transaction is already established, that + is, the :meth:`_engine.Connection.begin` + method or similar has been + called. A database cannot change the isolation level on a + transaction in progress, and different DBAPIs and/or + SQLAlchemy dialects may implicitly roll back or commit + the transaction, or not affect the connection at all. + + .. note:: The ``isolation_level`` execution option is implicitly + reset if the :class:`_engine.Connection` is invalidated, e.g. via + the :meth:`_engine.Connection.invalidate` method, or if a + disconnection error occurs. The new connection produced after + the invalidation will not have the isolation level re-applied + to it automatically. + + .. seealso:: + + :paramref:`_sa.create_engine.isolation_level` + - set per :class:`_engine.Engine` isolation level + + :meth:`_engine.Connection.get_isolation_level` + - view current level + + :ref:`SQLite Transaction Isolation <sqlite_isolation_level>` + + :ref:`PostgreSQL Transaction Isolation <postgresql_isolation_level>` + + :ref:`MySQL Transaction Isolation <mysql_isolation_level>` + + :ref:`SQL Server Transaction Isolation <mssql_isolation_level>` + + :ref:`session_transaction_isolation` - for the ORM + + :param no_parameters: When ``True``, if the final parameter + list or dictionary is totally empty, will invoke the + statement on the cursor as ``cursor.execute(statement)``, + not passing the parameter collection at all. + Some DBAPIs such as psycopg2 and mysql-python consider + percent signs as significant only when parameters are + present; this option allows code to generate SQL + containing percent signs (and possibly other characters) + that is neutral regarding whether it's executed by the DBAPI + or piped into a script that's later invoked by + command line tools. + + :param stream_results: Available on: Connection, statement. + Indicate to the dialect that results should be + "streamed" and not pre-buffered, if possible. For backends + such as PostgreSQL, MySQL and MariaDB, this indicates the use of + a "server side cursor" as opposed to a client side cursor. + Other backends such as that of Oracle may already use server + side cursors by default. + + The usage of + :paramref:`_engine.Connection.execution_options.stream_results` is + usually combined with setting a fixed number of rows to to be fetched + in batches, to allow for efficient iteration of database rows while + at the same time not loading all result rows into memory at once; + this can be configured on a :class:`_engine.Result` object using the + :meth:`_engine.Result.yield_per` method, after execution has + returned a new :class:`_engine.Result`. If + :meth:`_engine.Result.yield_per` is not used, + the :paramref:`_engine.Connection.execution_options.stream_results` + mode of operation will instead use a dynamically sized buffer + which buffers sets of rows at a time, growing on each batch + based on a fixed growth size up until a limit which may + be configured using the + :paramref:`_engine.Connection.execution_options.max_row_buffer` + parameter. + + When using the ORM to fetch ORM mapped objects from a result, + :meth:`_engine.Result.yield_per` should always be used with + :paramref:`_engine.Connection.execution_options.stream_results`, + so that the ORM does not fetch all rows into new ORM objects at once. + + For typical use, the + :paramref:`_engine.Connection.execution_options.yield_per` execution + option should be preferred, which sets up both + :paramref:`_engine.Connection.execution_options.stream_results` and + :meth:`_engine.Result.yield_per` at once. This option is supported + both at a core level by :class:`_engine.Connection` as well as by the + ORM :class:`_engine.Session`; the latter is described at + :ref:`orm_queryguide_yield_per`. + + .. seealso:: + + :ref:`engine_stream_results` - background on + :paramref:`_engine.Connection.execution_options.stream_results` + + :paramref:`_engine.Connection.execution_options.max_row_buffer` + + :paramref:`_engine.Connection.execution_options.yield_per` + + :ref:`orm_queryguide_yield_per` - in the :ref:`queryguide_toplevel` + describing the ORM version of ``yield_per`` + + :param max_row_buffer: Available on: :class:`_engine.Connection`, + :class:`_sql.Executable`. Sets a maximum + buffer size to use when the + :paramref:`_engine.Connection.execution_options.stream_results` + execution option is used on a backend that supports server side + cursors. The default value if not specified is 1000. + + .. seealso:: + + :paramref:`_engine.Connection.execution_options.stream_results` + + :ref:`engine_stream_results` + + + :param yield_per: Available on: :class:`_engine.Connection`, + :class:`_sql.Executable`. Integer value applied which will + set the :paramref:`_engine.Connection.execution_options.stream_results` + execution option and invoke :meth:`_engine.Result.yield_per` + automatically at once. Allows equivalent functionality as + is present when using this parameter with the ORM. + + .. versionadded:: 1.4.40 + + .. seealso:: + + :ref:`engine_stream_results` - background and examples + on using server side cursors with Core. + + :ref:`orm_queryguide_yield_per` - in the :ref:`queryguide_toplevel` + describing the ORM version of ``yield_per`` + + :param schema_translate_map: Available on: :class:`_engine.Connection`, + :class:`_engine.Engine`, :class:`_sql.Executable`. + + :param schema_translate_map: Available on: Connection, Engine. + A dictionary mapping schema names to schema names, that will be + applied to the :paramref:`_schema.Table.schema` element of each + :class:`_schema.Table` + encountered when SQL or DDL expression elements + are compiled into strings; the resulting schema name will be + converted based on presence in the map of the original name. + + .. versionadded:: 1.1 + + .. seealso:: + + :ref:`schema_translating` + + .. seealso:: + + :meth:`_engine.Engine.execution_options` + + :meth:`.Executable.execution_options` + + :meth:`_engine.Connection.get_execution_options` + + + """ # noqa + c = self._generate_for_options() + c._execution_options = c._execution_options.union(opt) + if self._has_events or self.engine._has_events: + self.dispatch.set_connection_execution_options(c, opt) + self.dialect.set_connection_execution_options(c, opt) + return c + + def get_execution_options(self): + """Get the non-SQL options which will take effect during execution. + + .. versionadded:: 1.3 + + .. seealso:: + + :meth:`_engine.Connection.execution_options` + """ + return self._execution_options + + @property + def closed(self): + """Return True if this connection is closed.""" + + # note this is independent for a "branched" connection vs. + # the base + + return self._dbapi_connection is None and not self.__can_reconnect + + @property + def invalidated(self): + """Return True if this connection was invalidated.""" + + # prior to 1.4, "invalid" was stored as a state independent of + # "closed", meaning an invalidated connection could be "closed", + # the _dbapi_connection would be None and closed=True, yet the + # "invalid" flag would stay True. This meant that there were + # three separate states (open/valid, closed/valid, closed/invalid) + # when there is really no reason for that; a connection that's + # "closed" does not need to be "invalid". So the state is now + # represented by the two facts alone. + + if self.__branch_from: + return self.__branch_from.invalidated + + return self._dbapi_connection is None and not self.closed + + @property + def connection(self): + """The underlying DB-API connection managed by this Connection. + + This is a SQLAlchemy connection-pool proxied connection + which then has the attribute + :attr:`_pool._ConnectionFairy.dbapi_connection` that refers to the + actual driver connection. + + .. seealso:: + + + :ref:`dbapi_connections` + + """ + + if self._dbapi_connection is None: + try: + return self._revalidate_connection() + except (exc.PendingRollbackError, exc.ResourceClosedError): + raise + except BaseException as e: + self._handle_dbapi_exception(e, None, None, None, None) + else: + return self._dbapi_connection + + def get_isolation_level(self): + """Return the current isolation level assigned to this + :class:`_engine.Connection`. + + This will typically be the default isolation level as determined + by the dialect, unless if the + :paramref:`.Connection.execution_options.isolation_level` + feature has been used to alter the isolation level on a + per-:class:`_engine.Connection` basis. + + This attribute will typically perform a live SQL operation in order + to procure the current isolation level, so the value returned is the + actual level on the underlying DBAPI connection regardless of how + this state was set. Compare to the + :attr:`_engine.Connection.default_isolation_level` accessor + which returns the dialect-level setting without performing a SQL + query. + + .. versionadded:: 0.9.9 + + .. seealso:: + + :attr:`_engine.Connection.default_isolation_level` + - view default level + + :paramref:`_sa.create_engine.isolation_level` + - set per :class:`_engine.Engine` isolation level + + :paramref:`.Connection.execution_options.isolation_level` + - set per :class:`_engine.Connection` isolation level + + """ + try: + return self.dialect.get_isolation_level(self.connection) + except BaseException as e: + self._handle_dbapi_exception(e, None, None, None, None) + + @property + def default_isolation_level(self): + """The default isolation level assigned to this + :class:`_engine.Connection`. + + This is the isolation level setting that the + :class:`_engine.Connection` + has when first procured via the :meth:`_engine.Engine.connect` method. + This level stays in place until the + :paramref:`.Connection.execution_options.isolation_level` is used + to change the setting on a per-:class:`_engine.Connection` basis. + + Unlike :meth:`_engine.Connection.get_isolation_level`, + this attribute is set + ahead of time from the first connection procured by the dialect, + so SQL query is not invoked when this accessor is called. + + .. versionadded:: 0.9.9 + + .. seealso:: + + :meth:`_engine.Connection.get_isolation_level` + - view current level + + :paramref:`_sa.create_engine.isolation_level` + - set per :class:`_engine.Engine` isolation level + + :paramref:`.Connection.execution_options.isolation_level` + - set per :class:`_engine.Connection` isolation level + + """ + return self.dialect.default_isolation_level + + def _invalid_transaction(self): + if self.invalidated: + raise exc.PendingRollbackError( + "Can't reconnect until invalid %stransaction is rolled " + "back." + % ( + "savepoint " + if self._nested_transaction is not None + else "" + ), + code="8s2b", + ) + else: + assert not self._is_future + raise exc.PendingRollbackError( + "This connection is on an inactive %stransaction. " + "Please rollback() fully before proceeding." + % ( + "savepoint " + if self._nested_transaction is not None + else "" + ), + code="8s2a", + ) + + def _revalidate_connection(self): + if self.__branch_from: + return self.__branch_from._revalidate_connection() + if self.__can_reconnect and self.invalidated: + if self._transaction is not None: + self._invalid_transaction() + self._dbapi_connection = self.engine.raw_connection( + _connection=self + ) + return self._dbapi_connection + raise exc.ResourceClosedError("This Connection is closed") + + @property + def _still_open_and_dbapi_connection_is_valid(self): + return self._dbapi_connection is not None and getattr( + self._dbapi_connection, "is_valid", False + ) + + @property + def info(self): + """Info dictionary associated with the underlying DBAPI connection + referred to by this :class:`_engine.Connection`, allowing user-defined + data to be associated with the connection. + + The data here will follow along with the DBAPI connection including + after it is returned to the connection pool and used again + in subsequent instances of :class:`_engine.Connection`. + + """ + + return self.connection.info + + @util.deprecated_20(":meth:`.Connection.connect`") + def connect(self, close_with_result=False): + """Returns a branched version of this :class:`_engine.Connection`. + + The :meth:`_engine.Connection.close` method on the returned + :class:`_engine.Connection` can be called and this + :class:`_engine.Connection` will remain open. + + This method provides usage symmetry with + :meth:`_engine.Engine.connect`, including for usage + with context managers. + + """ + + return self._branch() + + def invalidate(self, exception=None): + """Invalidate the underlying DBAPI connection associated with + this :class:`_engine.Connection`. + + An attempt will be made to close the underlying DBAPI connection + immediately; however if this operation fails, the error is logged + but not raised. The connection is then discarded whether or not + close() succeeded. + + Upon the next use (where "use" typically means using the + :meth:`_engine.Connection.execute` method or similar), + this :class:`_engine.Connection` will attempt to + procure a new DBAPI connection using the services of the + :class:`_pool.Pool` as a source of connectivity (e.g. + a "reconnection"). + + If a transaction was in progress (e.g. the + :meth:`_engine.Connection.begin` method has been called) when + :meth:`_engine.Connection.invalidate` method is called, at the DBAPI + level all state associated with this transaction is lost, as + the DBAPI connection is closed. The :class:`_engine.Connection` + will not allow a reconnection to proceed until the + :class:`.Transaction` object is ended, by calling the + :meth:`.Transaction.rollback` method; until that point, any attempt at + continuing to use the :class:`_engine.Connection` will raise an + :class:`~sqlalchemy.exc.InvalidRequestError`. + This is to prevent applications from accidentally + continuing an ongoing transactional operations despite the + fact that the transaction has been lost due to an + invalidation. + + The :meth:`_engine.Connection.invalidate` method, + just like auto-invalidation, + will at the connection pool level invoke the + :meth:`_events.PoolEvents.invalidate` event. + + :param exception: an optional ``Exception`` instance that's the + reason for the invalidation. is passed along to event handlers + and logging functions. + + .. seealso:: + + :ref:`pool_connection_invalidation` + + """ + + if self.__branch_from: + return self.__branch_from.invalidate(exception=exception) + + if self.invalidated: + return + + if self.closed: + raise exc.ResourceClosedError("This Connection is closed") + + if self._still_open_and_dbapi_connection_is_valid: + self._dbapi_connection.invalidate(exception) + self._dbapi_connection = None + + def detach(self): + """Detach the underlying DB-API connection from its connection pool. + + E.g.:: + + with engine.connect() as conn: + conn.detach() + conn.execute(text("SET search_path TO schema1, schema2")) + + # work with connection + + # connection is fully closed (since we used "with:", can + # also call .close()) + + This :class:`_engine.Connection` instance will remain usable. + When closed + (or exited from a context manager context as above), + the DB-API connection will be literally closed and not + returned to its originating pool. + + This method can be used to insulate the rest of an application + from a modified state on a connection (such as a transaction + isolation level or similar). + + """ + + self._dbapi_connection.detach() + + def _autobegin(self): + self.begin() + + def begin(self): + """Begin a transaction and return a transaction handle. + + The returned object is an instance of :class:`.Transaction`. + This object represents the "scope" of the transaction, + which completes when either the :meth:`.Transaction.rollback` + or :meth:`.Transaction.commit` method is called. + + .. tip:: + + The :meth:`_engine.Connection.begin` method is invoked when using + the :meth:`_engine.Engine.begin` context manager method as well. + All documentation that refers to behaviors specific to the + :meth:`_engine.Connection.begin` method also apply to use of the + :meth:`_engine.Engine.begin` method. + + Legacy use: nested calls to :meth:`.begin` on the same + :class:`_engine.Connection` will return new :class:`.Transaction` + objects that represent an emulated transaction within the scope of the + enclosing transaction, that is:: + + trans = conn.begin() # outermost transaction + trans2 = conn.begin() # "nested" + trans2.commit() # does nothing + trans.commit() # actually commits + + Calls to :meth:`.Transaction.commit` only have an effect + when invoked via the outermost :class:`.Transaction` object, though the + :meth:`.Transaction.rollback` method of any of the + :class:`.Transaction` objects will roll back the + transaction. + + .. tip:: + + The above "nesting" behavior is a legacy behavior specific to + :term:`1.x style` use and will be removed in SQLAlchemy 2.0. For + notes on :term:`2.0 style` use, see + :meth:`_future.Connection.begin`. + + + .. seealso:: + + :meth:`_engine.Connection.begin_nested` - use a SAVEPOINT + + :meth:`_engine.Connection.begin_twophase` - + use a two phase /XID transaction + + :meth:`_engine.Engine.begin` - context manager available from + :class:`_engine.Engine` + + """ + if self._is_future: + assert not self.__branch_from + elif self.__branch_from: + return self.__branch_from.begin() + + if self.__in_begin: + # for dialects that emit SQL within the process of + # dialect.do_begin() or dialect.do_begin_twophase(), this + # flag prevents "autobegin" from being emitted within that + # process, while allowing self._transaction to remain at None + # until it's complete. + return + elif self._transaction is None: + self._transaction = RootTransaction(self) + return self._transaction + else: + if self._is_future: + raise exc.InvalidRequestError( + "This connection has already initialized a SQLAlchemy " + "Transaction() object via begin() or autobegin; can't " + "call begin() here unless rollback() or commit() " + "is called first." + ) + else: + return MarkerTransaction(self) + + def begin_nested(self): + """Begin a nested transaction (i.e. SAVEPOINT) and return a + transaction handle, assuming an outer transaction is already + established. + + Nested transactions require SAVEPOINT support in the + underlying database. Any transaction in the hierarchy may + ``commit`` and ``rollback``, however the outermost transaction + still controls the overall ``commit`` or ``rollback`` of the + transaction of a whole. + + The legacy form of :meth:`_engine.Connection.begin_nested` method has + alternate behaviors based on whether or not the + :meth:`_engine.Connection.begin` method was called previously. If + :meth:`_engine.Connection.begin` was not called, then this method will + behave the same as the :meth:`_engine.Connection.begin` method and + return a :class:`.RootTransaction` object that begins and commits a + real transaction - **no savepoint is invoked**. If + :meth:`_engine.Connection.begin` **has** been called, and a + :class:`.RootTransaction` is already established, then this method + returns an instance of :class:`.NestedTransaction` which will invoke + and manage the scope of a SAVEPOINT. + + .. tip:: + + The above mentioned behavior of + :meth:`_engine.Connection.begin_nested` is a legacy behavior + specific to :term:`1.x style` use. In :term:`2.0 style` use, the + :meth:`_future.Connection.begin_nested` method instead autobegins + the outer transaction that can be committed using + "commit-as-you-go" style; see + :meth:`_future.Connection.begin_nested` for migration details. + + .. versionchanged:: 1.4.13 The behavior of + :meth:`_engine.Connection.begin_nested` + as returning a :class:`.RootTransaction` if + :meth:`_engine.Connection.begin` were not called has been restored + as was the case in 1.3.x versions; in previous 1.4.x versions, an + outer transaction would be "autobegun" but would not be committed. + + + .. seealso:: + + :meth:`_engine.Connection.begin` + + :ref:`session_begin_nested` - ORM support for SAVEPOINT + + """ + if self._is_future: + assert not self.__branch_from + elif self.__branch_from: + return self.__branch_from.begin_nested() + + if self._transaction is None: + if not self._is_future: + util.warn_deprecated_20( + "Calling Connection.begin_nested() in 2.0 style use will " + "return a NestedTransaction (SAVEPOINT) in all cases, " + "that will not commit the outer transaction. For code " + "that is cross-compatible between 1.x and 2.0 style use, " + "ensure Connection.begin() is called before calling " + "Connection.begin_nested()." + ) + return self.begin() + else: + self._autobegin() + + return NestedTransaction(self) + + def begin_twophase(self, xid=None): + """Begin a two-phase or XA transaction and return a transaction + handle. + + The returned object is an instance of :class:`.TwoPhaseTransaction`, + which in addition to the methods provided by + :class:`.Transaction`, also provides a + :meth:`~.TwoPhaseTransaction.prepare` method. + + :param xid: the two phase transaction id. If not supplied, a + random id will be generated. + + .. seealso:: + + :meth:`_engine.Connection.begin` + + :meth:`_engine.Connection.begin_twophase` + + """ + + if self.__branch_from: + return self.__branch_from.begin_twophase(xid=xid) + + if self._transaction is not None: + raise exc.InvalidRequestError( + "Cannot start a two phase transaction when a transaction " + "is already in progress." + ) + if xid is None: + xid = self.engine.dialect.create_xid() + return TwoPhaseTransaction(self, xid) + + def 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) + + def commit_prepared(self, xid, recover=False): + self.engine.dialect.do_commit_twophase(self, xid, recover=recover) + + def in_transaction(self): + """Return True if a transaction is in progress.""" + if self.__branch_from is not None: + return self.__branch_from.in_transaction() + + return self._transaction is not None and self._transaction.is_active + + def in_nested_transaction(self): + """Return True if a transaction is in progress.""" + if self.__branch_from is not None: + return self.__branch_from.in_nested_transaction() + + return ( + self._nested_transaction is not None + and self._nested_transaction.is_active + ) + + def _is_autocommit_isolation(self): + opt_iso = self._execution_options.get("isolation_level", None) + return bool( + opt_iso == "AUTOCOMMIT" + or ( + opt_iso is None + and getattr(self.engine.dialect, "isolation_level", None) + == "AUTOCOMMIT" + ) + ) + + def get_transaction(self): + """Return the current root transaction in progress, if any. + + .. versionadded:: 1.4 + + """ + + if self.__branch_from is not None: + return self.__branch_from.get_transaction() + + return self._transaction + + def get_nested_transaction(self): + """Return the current nested transaction in progress, if any. + + .. versionadded:: 1.4 + + """ + if self.__branch_from is not None: + + return self.__branch_from.get_nested_transaction() + + return self._nested_transaction + + def _begin_impl(self, transaction): + assert not self.__branch_from + + if self._echo: + if self._is_autocommit_isolation(): + self._log_info( + "BEGIN (implicit; DBAPI should not BEGIN due to " + "autocommit mode)" + ) + else: + self._log_info("BEGIN (implicit)") + + self.__in_begin = True + + if self._has_events or self.engine._has_events: + self.dispatch.begin(self) + + try: + self.engine.dialect.do_begin(self.connection) + except BaseException as e: + self._handle_dbapi_exception(e, None, None, None, None) + finally: + self.__in_begin = False + + def _rollback_impl(self): + assert not self.__branch_from + + if self._has_events or self.engine._has_events: + self.dispatch.rollback(self) + + if self._still_open_and_dbapi_connection_is_valid: + if self._echo: + if self._is_autocommit_isolation(): + self._log_info( + "ROLLBACK using DBAPI connection.rollback(), " + "DBAPI should ignore due to autocommit mode" + ) + else: + self._log_info("ROLLBACK") + try: + self.engine.dialect.do_rollback(self.connection) + except BaseException as e: + self._handle_dbapi_exception(e, None, None, None, None) + + def _commit_impl(self, autocommit=False): + assert not self.__branch_from + + # AUTOCOMMIT isolation-level is a dialect-specific concept, however + # if a connection has this set as the isolation level, we can skip + # the "autocommit" warning as the operation will do "autocommit" + # in any case + if autocommit and not self._is_autocommit_isolation(): + util.warn_deprecated_20( + "The current statement is being autocommitted using " + "implicit autocommit, which will be removed in " + "SQLAlchemy 2.0. " + "Use the .begin() method of Engine or Connection in order to " + "use an explicit transaction for DML and DDL statements." + ) + + if self._has_events or self.engine._has_events: + self.dispatch.commit(self) + + if self._echo: + if self._is_autocommit_isolation(): + self._log_info( + "COMMIT using DBAPI connection.commit(), " + "DBAPI should ignore due to autocommit mode" + ) + else: + self._log_info("COMMIT") + try: + self.engine.dialect.do_commit(self.connection) + except BaseException as e: + self._handle_dbapi_exception(e, None, None, None, None) + + def _savepoint_impl(self, name=None): + assert not self.__branch_from + + if self._has_events or self.engine._has_events: + self.dispatch.savepoint(self, name) + + if name is None: + self.__savepoint_seq += 1 + name = "sa_savepoint_%s" % self.__savepoint_seq + if self._still_open_and_dbapi_connection_is_valid: + self.engine.dialect.do_savepoint(self, name) + return name + + def _rollback_to_savepoint_impl(self, name): + assert not self.__branch_from + + if self._has_events or self.engine._has_events: + self.dispatch.rollback_savepoint(self, name, None) + + if self._still_open_and_dbapi_connection_is_valid: + self.engine.dialect.do_rollback_to_savepoint(self, name) + + def _release_savepoint_impl(self, name): + assert not self.__branch_from + + if self._has_events or self.engine._has_events: + self.dispatch.release_savepoint(self, name, None) + + if self._still_open_and_dbapi_connection_is_valid: + self.engine.dialect.do_release_savepoint(self, name) + + def _begin_twophase_impl(self, transaction): + assert not self.__branch_from + + if self._echo: + self._log_info("BEGIN TWOPHASE (implicit)") + if self._has_events or self.engine._has_events: + self.dispatch.begin_twophase(self, transaction.xid) + + if self._still_open_and_dbapi_connection_is_valid: + self.__in_begin = True + try: + self.engine.dialect.do_begin_twophase(self, transaction.xid) + except BaseException as e: + self._handle_dbapi_exception(e, None, None, None, None) + finally: + self.__in_begin = False + + def _prepare_twophase_impl(self, xid): + assert not self.__branch_from + + if self._has_events or self.engine._has_events: + self.dispatch.prepare_twophase(self, xid) + + if self._still_open_and_dbapi_connection_is_valid: + assert isinstance(self._transaction, TwoPhaseTransaction) + try: + self.engine.dialect.do_prepare_twophase(self, xid) + except BaseException as e: + self._handle_dbapi_exception(e, None, None, None, None) + + def _rollback_twophase_impl(self, xid, is_prepared): + assert not self.__branch_from + + if self._has_events or self.engine._has_events: + self.dispatch.rollback_twophase(self, xid, is_prepared) + + if self._still_open_and_dbapi_connection_is_valid: + assert isinstance(self._transaction, TwoPhaseTransaction) + try: + self.engine.dialect.do_rollback_twophase( + self, xid, is_prepared + ) + except BaseException as e: + self._handle_dbapi_exception(e, None, None, None, None) + + def _commit_twophase_impl(self, xid, is_prepared): + assert not self.__branch_from + + if self._has_events or self.engine._has_events: + self.dispatch.commit_twophase(self, xid, is_prepared) + + if self._still_open_and_dbapi_connection_is_valid: + assert isinstance(self._transaction, TwoPhaseTransaction) + try: + self.engine.dialect.do_commit_twophase(self, xid, is_prepared) + except BaseException as e: + self._handle_dbapi_exception(e, None, None, None, None) + + def _autorollback(self): + if self.__branch_from: + self.__branch_from._autorollback() + + if not self.in_transaction(): + self._rollback_impl() + + def _warn_for_legacy_exec_format(self): + util.warn_deprecated_20( + "The connection.execute() method in " + "SQLAlchemy 2.0 will accept parameters as a single " + "dictionary or a " + "single sequence of dictionaries only. " + "Parameters passed as keyword arguments, tuples or positionally " + "oriented dictionaries and/or tuples " + "will no longer be accepted." + ) + + def close(self): + """Close this :class:`_engine.Connection`. + + This results in a release of the underlying database + resources, that is, the DBAPI connection referenced + internally. The DBAPI connection is typically restored + back to the connection-holding :class:`_pool.Pool` referenced + by the :class:`_engine.Engine` that produced this + :class:`_engine.Connection`. Any transactional state present on + the DBAPI connection is also unconditionally released via + the DBAPI connection's ``rollback()`` method, regardless + of any :class:`.Transaction` object that may be + outstanding with regards to this :class:`_engine.Connection`. + + After :meth:`_engine.Connection.close` is called, the + :class:`_engine.Connection` is permanently in a closed state, + and will allow no further operations. + + """ + + if self.__branch_from: + assert not self._is_future + util.warn_deprecated_20( + "The .close() method on a so-called 'branched' connection is " + "deprecated as of 1.4, as are 'branched' connections overall, " + "and will be removed in a future release. If this is a " + "default-handling function, don't close the connection." + ) + self._dbapi_connection = None + self.__can_reconnect = False + return + + if self._transaction: + self._transaction.close() + skip_reset = True + else: + skip_reset = False + + if self._dbapi_connection is not None: + conn = self._dbapi_connection + + # as we just closed the transaction, close the connection + # pool connection without doing an additional reset + if skip_reset: + conn._close_no_reset() + else: + conn.close() + + # There is a slight chance that conn.close() may have + # triggered an invalidation here in which case + # _dbapi_connection would already be None, however usually + # it will be non-None here and in a "closed" state. + self._dbapi_connection = None + self.__can_reconnect = False + + def scalar(self, object_, *multiparams, **params): + """Executes and returns the first column of the first row. + + The underlying result/cursor is closed after execution. + + """ + + return self.execute(object_, *multiparams, **params).scalar() + + def scalars(self, object_, *multiparams, **params): + """Executes and returns a scalar result set, which yields scalar values + from the first column of each row. + + This method is equivalent to calling :meth:`_engine.Connection.execute` + to receive a :class:`_result.Result` object, then invoking the + :meth:`_result.Result.scalars` method to produce a + :class:`_result.ScalarResult` instance. + + :return: a :class:`_result.ScalarResult` + + .. versionadded:: 1.4.24 + + """ + + return self.execute(object_, *multiparams, **params).scalars() + + def execute(self, statement, *multiparams, **params): + r"""Executes a SQL statement construct and returns a + :class:`_engine.CursorResult`. + + :param statement: The statement to be executed. May be + one of: + + * a plain string (deprecated) + * any :class:`_expression.ClauseElement` construct that is also + a subclass of :class:`.Executable`, such as a + :func:`_expression.select` construct + * a :class:`.FunctionElement`, such as that generated + by :data:`.func`, will be automatically wrapped in + a SELECT statement, which is then executed. + * a :class:`.DDLElement` object + * a :class:`.DefaultGenerator` object + * a :class:`.Compiled` object + + .. deprecated:: 2.0 passing a string to + :meth:`_engine.Connection.execute` is + deprecated and will be removed in version 2.0. Use the + :func:`_expression.text` construct with + :meth:`_engine.Connection.execute`, or the + :meth:`_engine.Connection.exec_driver_sql` + method to invoke a driver-level + SQL string. + + :param \*multiparams/\**params: represent bound parameter + values to be used in the execution. Typically, + the format is either a collection of one or more + dictionaries passed to \*multiparams:: + + conn.execute( + table.insert(), + {"id":1, "value":"v1"}, + {"id":2, "value":"v2"} + ) + + ...or individual key/values interpreted by \**params:: + + conn.execute( + table.insert(), id=1, value="v1" + ) + + In the case that a plain SQL string is passed, and the underlying + DBAPI accepts positional bind parameters, a collection of tuples + or individual values in \*multiparams may be passed:: + + conn.execute( + "INSERT INTO table (id, value) VALUES (?, ?)", + (1, "v1"), (2, "v2") + ) + + conn.execute( + "INSERT INTO table (id, value) VALUES (?, ?)", + 1, "v1" + ) + + Note above, the usage of a question mark "?" or other + symbol is contingent upon the "paramstyle" accepted by the DBAPI + in use, which may be any of "qmark", "named", "pyformat", "format", + "numeric". See `pep-249 + <https://www.python.org/dev/peps/pep-0249/>`_ for details on + paramstyle. + + To execute a textual SQL statement which uses bound parameters in a + DBAPI-agnostic way, use the :func:`_expression.text` construct. + + .. deprecated:: 2.0 use of tuple or scalar positional parameters + is deprecated. All params should be dicts or sequences of dicts. + Use :meth:`.exec_driver_sql` to execute a plain string with + tuple or scalar positional parameters. + + """ + + if isinstance(statement, util.string_types): + util.warn_deprecated_20( + "Passing a string to Connection.execute() is " + "deprecated and will be removed in version 2.0. Use the " + "text() construct, " + "or the Connection.exec_driver_sql() method to invoke a " + "driver-level SQL string." + ) + + return self._exec_driver_sql( + statement, + multiparams, + params, + _EMPTY_EXECUTION_OPTS, + future=False, + ) + + try: + meth = statement._execute_on_connection + except AttributeError as err: + util.raise_( + exc.ObjectNotExecutableError(statement), replace_context=err + ) + else: + return meth(self, multiparams, params, _EMPTY_EXECUTION_OPTS) + + def _execute_function(self, func, multiparams, params, execution_options): + """Execute a sql.FunctionElement object.""" + + return self._execute_clauseelement( + func.select(), multiparams, params, execution_options + ) + + def _execute_default( + self, + default, + multiparams, + params, + # migrate is calling this directly :( + execution_options=_EMPTY_EXECUTION_OPTS, + ): + """Execute a schema.ColumnDefault object.""" + + execution_options = self._execution_options.merge_with( + execution_options + ) + + distilled_parameters = _distill_params(self, multiparams, params) + + if self._has_events or self.engine._has_events: + ( + default, + distilled_params, + event_multiparams, + event_params, + ) = self._invoke_before_exec_event( + default, distilled_parameters, execution_options + ) + + try: + conn = self._dbapi_connection + if conn is None: + conn = self._revalidate_connection() + + dialect = self.dialect + ctx = dialect.execution_ctx_cls._init_default( + dialect, self, conn, execution_options + ) + except (exc.PendingRollbackError, exc.ResourceClosedError): + raise + except BaseException as e: + self._handle_dbapi_exception(e, None, None, None, None) + + ret = ctx._exec_default(None, default, None) + if self.should_close_with_result: + self.close() + + if self._has_events or self.engine._has_events: + self.dispatch.after_execute( + self, + default, + event_multiparams, + event_params, + execution_options, + ret, + ) + + return ret + + def _execute_ddl(self, ddl, multiparams, params, execution_options): + """Execute a schema.DDL object.""" + + execution_options = ddl._execution_options.merge_with( + self._execution_options, execution_options + ) + + distilled_parameters = _distill_params(self, multiparams, params) + + if self._has_events or self.engine._has_events: + ( + ddl, + distilled_params, + event_multiparams, + event_params, + ) = self._invoke_before_exec_event( + ddl, distilled_parameters, execution_options + ) + + exec_opts = self._execution_options.merge_with(execution_options) + schema_translate_map = exec_opts.get("schema_translate_map", None) + + dialect = self.dialect + + compiled = ddl.compile( + dialect=dialect, schema_translate_map=schema_translate_map + ) + ret = self._execute_context( + dialect, + dialect.execution_ctx_cls._init_ddl, + compiled, + None, + execution_options, + compiled, + ) + if self._has_events or self.engine._has_events: + self.dispatch.after_execute( + self, + ddl, + event_multiparams, + event_params, + execution_options, + ret, + ) + return ret + + def _invoke_before_exec_event( + self, elem, distilled_params, execution_options + ): + + if len(distilled_params) == 1: + event_multiparams, event_params = [], distilled_params[0] + else: + event_multiparams, event_params = distilled_params, {} + + for fn in self.dispatch.before_execute: + elem, event_multiparams, event_params = fn( + self, + elem, + event_multiparams, + event_params, + execution_options, + ) + + if event_multiparams: + distilled_params = list(event_multiparams) + if event_params: + raise exc.InvalidRequestError( + "Event handler can't return non-empty multiparams " + "and params at the same time" + ) + elif event_params: + distilled_params = [event_params] + else: + distilled_params = [] + + return elem, distilled_params, event_multiparams, event_params + + def _execute_clauseelement( + self, elem, multiparams, params, execution_options + ): + """Execute a sql.ClauseElement object.""" + + execution_options = elem._execution_options.merge_with( + self._execution_options, execution_options + ) + + distilled_params = _distill_params(self, multiparams, params) + + has_events = self._has_events or self.engine._has_events + if has_events: + ( + elem, + distilled_params, + event_multiparams, + event_params, + ) = self._invoke_before_exec_event( + elem, distilled_params, execution_options + ) + + if distilled_params: + # ensure we don't retain a link to the view object for keys() + # which links to the values, which we don't want to cache + keys = sorted(distilled_params[0]) + for_executemany = len(distilled_params) > 1 + else: + keys = [] + for_executemany = False + + dialect = self.dialect + + schema_translate_map = execution_options.get( + "schema_translate_map", None + ) + + compiled_cache = execution_options.get( + "compiled_cache", self.engine._compiled_cache + ) + + compiled_sql, extracted_params, cache_hit = elem._compile_w_cache( + dialect=dialect, + compiled_cache=compiled_cache, + column_keys=keys, + for_executemany=for_executemany, + schema_translate_map=schema_translate_map, + linting=self.dialect.compiler_linting | compiler.WARN_LINTING, + ) + ret = self._execute_context( + dialect, + dialect.execution_ctx_cls._init_compiled, + compiled_sql, + distilled_params, + execution_options, + compiled_sql, + distilled_params, + elem, + extracted_params, + cache_hit=cache_hit, + ) + if has_events: + self.dispatch.after_execute( + self, + elem, + event_multiparams, + event_params, + execution_options, + ret, + ) + return ret + + def _execute_compiled( + self, + compiled, + multiparams, + params, + execution_options=_EMPTY_EXECUTION_OPTS, + ): + """Execute a sql.Compiled object. + + TODO: why do we have this? likely deprecate or remove + + """ + + execution_options = compiled.execution_options.merge_with( + self._execution_options, execution_options + ) + distilled_parameters = _distill_params(self, multiparams, params) + + if self._has_events or self.engine._has_events: + ( + compiled, + distilled_params, + event_multiparams, + event_params, + ) = self._invoke_before_exec_event( + compiled, distilled_parameters, execution_options + ) + + dialect = self.dialect + + ret = self._execute_context( + dialect, + dialect.execution_ctx_cls._init_compiled, + compiled, + distilled_parameters, + execution_options, + compiled, + distilled_parameters, + None, + None, + ) + if self._has_events or self.engine._has_events: + self.dispatch.after_execute( + self, + compiled, + event_multiparams, + event_params, + execution_options, + ret, + ) + return ret + + def _exec_driver_sql( + self, statement, multiparams, params, execution_options, future + ): + + execution_options = self._execution_options.merge_with( + execution_options + ) + + distilled_parameters = _distill_params(self, multiparams, params) + + if not future: + if self._has_events or self.engine._has_events: + ( + statement, + distilled_params, + event_multiparams, + event_params, + ) = self._invoke_before_exec_event( + statement, distilled_parameters, execution_options + ) + + dialect = self.dialect + ret = self._execute_context( + dialect, + dialect.execution_ctx_cls._init_statement, + statement, + distilled_parameters, + execution_options, + statement, + distilled_parameters, + ) + + if not future: + if self._has_events or self.engine._has_events: + self.dispatch.after_execute( + self, + statement, + event_multiparams, + event_params, + execution_options, + ret, + ) + return ret + + def _execute_20( + self, + statement, + parameters=None, + execution_options=_EMPTY_EXECUTION_OPTS, + ): + args_10style, kwargs_10style = _distill_params_20(parameters) + try: + meth = statement._execute_on_connection + except AttributeError as err: + util.raise_( + exc.ObjectNotExecutableError(statement), replace_context=err + ) + else: + return meth(self, args_10style, kwargs_10style, execution_options) + + def exec_driver_sql( + self, statement, parameters=None, execution_options=None + ): + r"""Executes a SQL statement construct and returns a + :class:`_engine.CursorResult`. + + :param statement: The statement str to be executed. Bound parameters + must use the underlying DBAPI's paramstyle, such as "qmark", + "pyformat", "format", etc. + + :param parameters: represent bound parameter values to be used in the + execution. The format is one of: a dictionary of named parameters, + a tuple of positional parameters, or a list containing either + dictionaries or tuples for multiple-execute support. + + E.g. multiple dictionaries:: + + + conn.exec_driver_sql( + "INSERT INTO table (id, value) VALUES (%(id)s, %(value)s)", + [{"id":1, "value":"v1"}, {"id":2, "value":"v2"}] + ) + + Single dictionary:: + + conn.exec_driver_sql( + "INSERT INTO table (id, value) VALUES (%(id)s, %(value)s)", + dict(id=1, value="v1") + ) + + Single tuple:: + + conn.exec_driver_sql( + "INSERT INTO table (id, value) VALUES (?, ?)", + (1, 'v1') + ) + + .. note:: The :meth:`_engine.Connection.exec_driver_sql` method does + not participate in the + :meth:`_events.ConnectionEvents.before_execute` and + :meth:`_events.ConnectionEvents.after_execute` events. To + intercept calls to :meth:`_engine.Connection.exec_driver_sql`, use + :meth:`_events.ConnectionEvents.before_cursor_execute` and + :meth:`_events.ConnectionEvents.after_cursor_execute`. + + .. seealso:: + + :pep:`249` + + """ + + args_10style, kwargs_10style = _distill_params_20(parameters) + + return self._exec_driver_sql( + statement, + args_10style, + kwargs_10style, + execution_options, + future=True, + ) + + def _execute_context( + self, + dialect, + constructor, + statement, + parameters, + execution_options, + *args, + **kw + ): + """Create an :class:`.ExecutionContext` and execute, returning + a :class:`_engine.CursorResult`.""" + + branched = self + if self.__branch_from: + # if this is a "branched" connection, do everything in terms + # of the "root" connection, *except* for .close(), which is + # the only feature that branching provides + self = self.__branch_from + + if execution_options: + yp = execution_options.get("yield_per", None) + if yp: + execution_options = execution_options.union( + {"stream_results": True, "max_row_buffer": yp} + ) + + try: + conn = self._dbapi_connection + if conn is None: + conn = self._revalidate_connection() + + context = constructor( + dialect, self, conn, execution_options, *args, **kw + ) + except (exc.PendingRollbackError, exc.ResourceClosedError): + raise + except BaseException as e: + self._handle_dbapi_exception( + e, util.text_type(statement), parameters, None, None + ) + + if ( + self._transaction + and not self._transaction.is_active + or ( + self._nested_transaction + and not self._nested_transaction.is_active + ) + ): + self._invalid_transaction() + + elif self._trans_context_manager: + TransactionalContext._trans_ctx_check(self) + + if self._is_future and self._transaction is None: + self._autobegin() + + context.pre_exec() + + if dialect.use_setinputsizes: + context._set_input_sizes() + + cursor, statement, parameters = ( + context.cursor, + context.statement, + context.parameters, + ) + + if not context.executemany: + parameters = parameters[0] + + if self._has_events or self.engine._has_events: + for fn in self.dispatch.before_cursor_execute: + statement, parameters = fn( + self, + cursor, + statement, + parameters, + context, + context.executemany, + ) + + if self._echo: + + self._log_info(statement) + + stats = context._get_cache_stats() + + if not self.engine.hide_parameters: + self._log_info( + "[%s] %r", + stats, + sql_util._repr_params( + parameters, batches=10, ismulti=context.executemany + ), + ) + else: + self._log_info( + "[%s] [SQL parameters hidden due to hide_parameters=True]" + % (stats,) + ) + + evt_handled = False + try: + if context.executemany: + if self.dialect._has_events: + for fn in self.dialect.dispatch.do_executemany: + if fn(cursor, statement, parameters, context): + evt_handled = True + break + if not evt_handled: + self.dialect.do_executemany( + cursor, statement, parameters, context + ) + elif not parameters and context.no_parameters: + if self.dialect._has_events: + for fn in self.dialect.dispatch.do_execute_no_params: + if fn(cursor, statement, context): + evt_handled = True + break + if not evt_handled: + self.dialect.do_execute_no_params( + cursor, statement, context + ) + else: + if self.dialect._has_events: + for fn in self.dialect.dispatch.do_execute: + if fn(cursor, statement, parameters, context): + evt_handled = True + break + if not evt_handled: + self.dialect.do_execute( + cursor, statement, parameters, context + ) + + if self._has_events or self.engine._has_events: + self.dispatch.after_cursor_execute( + self, + cursor, + statement, + parameters, + context, + context.executemany, + ) + + context.post_exec() + + result = context._setup_result_proxy() + + if not self._is_future: + should_close_with_result = branched.should_close_with_result + + if not result._soft_closed and should_close_with_result: + result._autoclose_connection = True + + if ( + # usually we're in a transaction so avoid relatively + # expensive / legacy should_autocommit call + self._transaction is None + and context.should_autocommit + ): + self._commit_impl(autocommit=True) + + # for "connectionless" execution, we have to close this + # Connection after the statement is complete. + # legacy stuff. + if should_close_with_result and context._soft_closed: + assert not self._is_future + + # CursorResult already exhausted rows / has no rows. + # close us now + branched.close() + + except BaseException as e: + self._handle_dbapi_exception( + e, statement, parameters, cursor, context + ) + + return result + + def _cursor_execute(self, cursor, statement, parameters, context=None): + """Execute a statement + params on the given cursor. + + Adds appropriate logging and exception handling. + + This method is used by DefaultDialect for special-case + executions, such as for sequences and column defaults. + The path of statement execution in the majority of cases + terminates at _execute_context(). + + """ + if self._has_events or self.engine._has_events: + for fn in self.dispatch.before_cursor_execute: + statement, parameters = fn( + self, cursor, statement, parameters, context, False + ) + + if self._echo: + self._log_info(statement) + self._log_info("[raw sql] %r", parameters) + try: + for fn in ( + () + if not self.dialect._has_events + else self.dialect.dispatch.do_execute + ): + if fn(cursor, statement, parameters, context): + break + else: + self.dialect.do_execute(cursor, statement, parameters, context) + except BaseException as e: + self._handle_dbapi_exception( + e, statement, parameters, cursor, context + ) + + if self._has_events or self.engine._has_events: + self.dispatch.after_cursor_execute( + self, cursor, statement, parameters, context, False + ) + + def _safe_close_cursor(self, cursor): + """Close the given cursor, catching exceptions + and turning into log warnings. + + """ + try: + cursor.close() + except Exception: + # log the error through the connection pool's logger. + self.engine.pool.logger.error( + "Error closing cursor", exc_info=True + ) + + _reentrant_error = False + _is_disconnect = False + + def _handle_dbapi_exception( + self, e, statement, parameters, cursor, context + ): + exc_info = sys.exc_info() + + is_exit_exception = util.is_exit_exception(e) + + if not self._is_disconnect: + self._is_disconnect = ( + isinstance(e, self.dialect.dbapi.Error) + and not self.closed + and self.dialect.is_disconnect( + e, + self._dbapi_connection if not self.invalidated else None, + cursor, + ) + ) or (is_exit_exception and not self.closed) + + invalidate_pool_on_disconnect = not is_exit_exception + + if self._reentrant_error: + util.raise_( + exc.DBAPIError.instance( + statement, + parameters, + e, + self.dialect.dbapi.Error, + hide_parameters=self.engine.hide_parameters, + dialect=self.dialect, + ismulti=context.executemany + if context is not None + else None, + ), + with_traceback=exc_info[2], + from_=e, + ) + self._reentrant_error = True + try: + # non-DBAPI error - if we already got a context, + # or there's no string statement, don't wrap it + should_wrap = isinstance(e, self.dialect.dbapi.Error) or ( + statement is not None + and context is None + and not is_exit_exception + ) + + if should_wrap: + sqlalchemy_exception = exc.DBAPIError.instance( + statement, + parameters, + e, + self.dialect.dbapi.Error, + hide_parameters=self.engine.hide_parameters, + connection_invalidated=self._is_disconnect, + dialect=self.dialect, + ismulti=context.executemany + if context is not None + else None, + ) + else: + sqlalchemy_exception = None + + newraise = None + + if ( + self._has_events or self.engine._has_events + ) and not self._execution_options.get( + "skip_user_error_events", False + ): + ctx = ExceptionContextImpl( + e, + sqlalchemy_exception, + self.engine, + self, + cursor, + statement, + parameters, + context, + self._is_disconnect, + invalidate_pool_on_disconnect, + ) + + for fn in self.dispatch.handle_error: + try: + # handler returns an exception; + # call next handler in a chain + per_fn = fn(ctx) + if per_fn is not None: + ctx.chained_exception = newraise = per_fn + except Exception as _raised: + # handler raises an exception - stop processing + newraise = _raised + break + + if self._is_disconnect != ctx.is_disconnect: + self._is_disconnect = ctx.is_disconnect + if sqlalchemy_exception: + sqlalchemy_exception.connection_invalidated = ( + ctx.is_disconnect + ) + + # set up potentially user-defined value for + # invalidate pool. + invalidate_pool_on_disconnect = ( + ctx.invalidate_pool_on_disconnect + ) + + if should_wrap and context: + context.handle_dbapi_exception(e) + + if not self._is_disconnect: + if cursor: + self._safe_close_cursor(cursor) + with util.safe_reraise(warn_only=True): + self._autorollback() + + if newraise: + util.raise_(newraise, with_traceback=exc_info[2], from_=e) + elif should_wrap: + util.raise_( + sqlalchemy_exception, with_traceback=exc_info[2], from_=e + ) + else: + util.raise_(exc_info[1], with_traceback=exc_info[2]) + + finally: + del self._reentrant_error + if self._is_disconnect: + del self._is_disconnect + if not self.invalidated: + dbapi_conn_wrapper = self._dbapi_connection + if invalidate_pool_on_disconnect: + self.engine.pool._invalidate(dbapi_conn_wrapper, e) + self.invalidate(e) + if self.should_close_with_result: + assert not self._is_future + self.close() + + @classmethod + def _handle_dbapi_exception_noconnection(cls, e, dialect, engine): + exc_info = sys.exc_info() + + is_disconnect = dialect.is_disconnect(e, None, None) + + should_wrap = isinstance(e, dialect.dbapi.Error) + + if should_wrap: + sqlalchemy_exception = exc.DBAPIError.instance( + None, + None, + e, + dialect.dbapi.Error, + hide_parameters=engine.hide_parameters, + connection_invalidated=is_disconnect, + ) + else: + sqlalchemy_exception = None + + newraise = None + + if engine._has_events: + ctx = ExceptionContextImpl( + e, + sqlalchemy_exception, + engine, + None, + None, + None, + None, + None, + is_disconnect, + True, + ) + for fn in engine.dispatch.handle_error: + try: + # handler returns an exception; + # call next handler in a chain + per_fn = fn(ctx) + if per_fn is not None: + ctx.chained_exception = newraise = per_fn + except Exception as _raised: + # handler raises an exception - stop processing + newraise = _raised + break + + if sqlalchemy_exception and is_disconnect != ctx.is_disconnect: + sqlalchemy_exception.connection_invalidated = ( + is_disconnect + ) = ctx.is_disconnect + + if newraise: + util.raise_(newraise, with_traceback=exc_info[2], from_=e) + elif should_wrap: + util.raise_( + sqlalchemy_exception, with_traceback=exc_info[2], from_=e + ) + else: + util.raise_(exc_info[1], with_traceback=exc_info[2]) + + def _run_ddl_visitor(self, visitorcallable, element, **kwargs): + """run a DDL visitor. + + This method is only here so that the MockConnection can change the + options given to the visitor so that "checkfirst" is skipped. + + """ + visitorcallable(self.dialect, self, **kwargs).traverse_single(element) + + @util.deprecated( + "1.4", + "The :meth:`_engine.Connection.transaction` " + "method is deprecated and will be " + "removed in a future release. Use the :meth:`_engine.Engine.begin` " + "context manager instead.", + ) + def transaction(self, callable_, *args, **kwargs): + r"""Execute the given function within a transaction boundary. + + The function is passed this :class:`_engine.Connection` + as the first argument, followed by the given \*args and \**kwargs, + e.g.:: + + def do_something(conn, x, y): + conn.execute(text("some statement"), {'x':x, 'y':y}) + + conn.transaction(do_something, 5, 10) + + The operations inside the function are all invoked within the + context of a single :class:`.Transaction`. + Upon success, the transaction is committed. If an + exception is raised, the transaction is rolled back + before propagating the exception. + + .. note:: + + The :meth:`.transaction` method is superseded by + the usage of the Python ``with:`` statement, which can + be used with :meth:`_engine.Connection.begin`:: + + with conn.begin(): + conn.execute(text("some statement"), {'x':5, 'y':10}) + + As well as with :meth:`_engine.Engine.begin`:: + + with engine.begin() as conn: + conn.execute(text("some statement"), {'x':5, 'y':10}) + + .. seealso:: + + :meth:`_engine.Engine.begin` - engine-level transactional + context + + :meth:`_engine.Engine.transaction` - engine-level version of + :meth:`_engine.Connection.transaction` + + """ + + kwargs["_sa_skip_warning"] = True + trans = self.begin() + try: + ret = self.run_callable(callable_, *args, **kwargs) + trans.commit() + return ret + except: + with util.safe_reraise(): + trans.rollback() + + @util.deprecated( + "1.4", + "The :meth:`_engine.Connection.run_callable` " + "method is deprecated and will " + "be removed in a future release. Invoke the callable function " + "directly, passing the Connection.", + ) + def run_callable(self, callable_, *args, **kwargs): + r"""Given a callable object or function, execute it, passing + a :class:`_engine.Connection` as the first argument. + + The given \*args and \**kwargs are passed subsequent + to the :class:`_engine.Connection` argument. + + This function, along with :meth:`_engine.Engine.run_callable`, + allows a function to be run with a :class:`_engine.Connection` + or :class:`_engine.Engine` object without the need to know + which one is being dealt with. + + """ + return callable_(self, *args, **kwargs) + + +class ExceptionContextImpl(ExceptionContext): + """Implement the :class:`.ExceptionContext` interface.""" + + def __init__( + self, + exception, + sqlalchemy_exception, + engine, + connection, + cursor, + statement, + parameters, + context, + is_disconnect, + invalidate_pool_on_disconnect, + ): + self.engine = engine + self.connection = connection + self.sqlalchemy_exception = sqlalchemy_exception + self.original_exception = exception + self.execution_context = context + self.statement = statement + self.parameters = parameters + self.is_disconnect = is_disconnect + self.invalidate_pool_on_disconnect = invalidate_pool_on_disconnect + + +class Transaction(TransactionalContext): + """Represent a database transaction in progress. + + The :class:`.Transaction` object is procured by + calling the :meth:`_engine.Connection.begin` method of + :class:`_engine.Connection`:: + + from sqlalchemy import create_engine + engine = create_engine("postgresql://scott:tiger@localhost/test") + connection = engine.connect() + trans = connection.begin() + connection.execute(text("insert into x (a, b) values (1, 2)")) + trans.commit() + + The object provides :meth:`.rollback` and :meth:`.commit` + methods in order to control transaction boundaries. It + also implements a context manager interface so that + the Python ``with`` statement can be used with the + :meth:`_engine.Connection.begin` method:: + + with connection.begin(): + connection.execute(text("insert into x (a, b) values (1, 2)")) + + The Transaction object is **not** threadsafe. + + .. seealso:: + + :meth:`_engine.Connection.begin` + + :meth:`_engine.Connection.begin_twophase` + + :meth:`_engine.Connection.begin_nested` + + .. index:: + single: thread safety; Transaction + """ + + __slots__ = () + + _is_root = False + + def __init__(self, connection): + raise NotImplementedError() + + def _do_deactivate(self): + """do whatever steps are necessary to set this transaction as + "deactive", however leave this transaction object in place as far + as the connection's state. + + for a "real" transaction this should roll back the transaction + and ensure this transaction is no longer a reset agent. + + this is used for nesting of marker transactions where the marker + can set the "real" transaction as rolled back, however it stays + in place. + + for 2.0 we hope to remove this nesting feature. + + """ + raise NotImplementedError() + + @property + def _deactivated_from_connection(self): + """True if this transaction is totally deactivated from the connection + and therefore can no longer affect its state. + + """ + raise NotImplementedError() + + def _do_close(self): + raise NotImplementedError() + + def _do_rollback(self): + raise NotImplementedError() + + def _do_commit(self): + raise NotImplementedError() + + @property + def is_valid(self): + return self.is_active and not self.connection.invalidated + + def close(self): + """Close this :class:`.Transaction`. + + If this transaction is the base transaction in a begin/commit + nesting, the transaction will rollback(). Otherwise, the + method returns. + + This is used to cancel a Transaction without affecting the scope of + an enclosing transaction. + + """ + try: + self._do_close() + finally: + assert not self.is_active + + def rollback(self): + """Roll back this :class:`.Transaction`. + + The implementation of this may vary based on the type of transaction in + use: + + * For a simple database transaction (e.g. :class:`.RootTransaction`), + it corresponds to a ROLLBACK. + + * For a :class:`.NestedTransaction`, it corresponds to a + "ROLLBACK TO SAVEPOINT" operation. + + * For a :class:`.TwoPhaseTransaction`, DBAPI-specific methods for two + phase transactions may be used. + + + """ + try: + self._do_rollback() + finally: + assert not self.is_active + + def commit(self): + """Commit this :class:`.Transaction`. + + The implementation of this may vary based on the type of transaction in + use: + + * For a simple database transaction (e.g. :class:`.RootTransaction`), + it corresponds to a COMMIT. + + * For a :class:`.NestedTransaction`, it corresponds to a + "RELEASE SAVEPOINT" operation. + + * For a :class:`.TwoPhaseTransaction`, DBAPI-specific methods for two + phase transactions may be used. + + """ + try: + self._do_commit() + finally: + assert not self.is_active + + def _get_subject(self): + return self.connection + + def _transaction_is_active(self): + return self.is_active + + def _transaction_is_closed(self): + return not self._deactivated_from_connection + + def _rollback_can_be_called(self): + # for RootTransaction / NestedTransaction, it's safe to call + # rollback() even if the transaction is deactive and no warnings + # will be emitted. tested in + # test_transaction.py -> test_no_rollback_in_deactive(?:_savepoint)? + return True + + +class MarkerTransaction(Transaction): + """A 'marker' transaction that is used for nested begin() calls. + + .. deprecated:: 1.4 future connection for 2.0 won't support this pattern. + + """ + + __slots__ = ("connection", "_is_active", "_transaction") + + def __init__(self, connection): + assert connection._transaction is not None + if not connection._transaction.is_active: + raise exc.InvalidRequestError( + "the current transaction on this connection is inactive. " + "Please issue a rollback first." + ) + + assert not connection._is_future + util.warn_deprecated_20( + "Calling .begin() when a transaction is already begun, creating " + "a 'sub' transaction, is deprecated " + "and will be removed in 2.0. See the documentation section " + "'Migrating from the nesting pattern' for background on how " + "to migrate from this pattern." + ) + + self.connection = connection + + if connection._trans_context_manager: + TransactionalContext._trans_ctx_check(connection) + + if connection._nested_transaction is not None: + self._transaction = connection._nested_transaction + else: + self._transaction = connection._transaction + self._is_active = True + + @property + def _deactivated_from_connection(self): + return not self.is_active + + @property + def is_active(self): + return self._is_active and self._transaction.is_active + + def _deactivate(self): + self._is_active = False + + def _do_close(self): + # does not actually roll back the root + self._deactivate() + + def _do_rollback(self): + # does roll back the root + if self._is_active: + try: + self._transaction._do_deactivate() + finally: + self._deactivate() + + def _do_commit(self): + self._deactivate() + + +class RootTransaction(Transaction): + """Represent the "root" transaction on a :class:`_engine.Connection`. + + This corresponds to the current "BEGIN/COMMIT/ROLLBACK" that's occurring + for the :class:`_engine.Connection`. The :class:`_engine.RootTransaction` + is created by calling upon the :meth:`_engine.Connection.begin` method, and + remains associated with the :class:`_engine.Connection` throughout its + active span. The current :class:`_engine.RootTransaction` in use is + accessible via the :attr:`_engine.Connection.get_transaction` method of + :class:`_engine.Connection`. + + In :term:`2.0 style` use, the :class:`_future.Connection` also employs + "autobegin" behavior that will create a new + :class:`_engine.RootTransaction` whenever a connection in a + non-transactional state is used to emit commands on the DBAPI connection. + The scope of the :class:`_engine.RootTransaction` in 2.0 style + use can be controlled using the :meth:`_future.Connection.commit` and + :meth:`_future.Connection.rollback` methods. + + + """ + + _is_root = True + + __slots__ = ("connection", "is_active") + + def __init__(self, connection): + assert connection._transaction is None + if connection._trans_context_manager: + TransactionalContext._trans_ctx_check(connection) + self.connection = connection + self._connection_begin_impl() + connection._transaction = self + + self.is_active = True + + def _deactivate_from_connection(self): + if self.is_active: + assert self.connection._transaction is self + self.is_active = False + + elif self.connection._transaction is not self: + util.warn("transaction already deassociated from connection") + + @property + def _deactivated_from_connection(self): + return self.connection._transaction is not self + + def _do_deactivate(self): + # called from a MarkerTransaction to cancel this root transaction. + # the transaction stays in place as connection._transaction, but + # is no longer active and is no longer the reset agent for the + # pooled connection. the connection won't support a new begin() + # until this transaction is explicitly closed, rolled back, + # or committed. + + assert self.connection._transaction is self + + if self.is_active: + self._connection_rollback_impl() + + # handle case where a savepoint was created inside of a marker + # transaction that refers to a root. nested has to be cancelled + # also. + if self.connection._nested_transaction: + self.connection._nested_transaction._cancel() + + self._deactivate_from_connection() + + def _connection_begin_impl(self): + self.connection._begin_impl(self) + + def _connection_rollback_impl(self): + self.connection._rollback_impl() + + def _connection_commit_impl(self): + self.connection._commit_impl() + + def _close_impl(self, try_deactivate=False): + try: + if self.is_active: + self._connection_rollback_impl() + + if self.connection._nested_transaction: + self.connection._nested_transaction._cancel() + finally: + if self.is_active or try_deactivate: + self._deactivate_from_connection() + if self.connection._transaction is self: + self.connection._transaction = None + + assert not self.is_active + assert self.connection._transaction is not self + + def _do_close(self): + self._close_impl() + + def _do_rollback(self): + self._close_impl(try_deactivate=True) + + def _do_commit(self): + if self.is_active: + assert self.connection._transaction is self + + try: + self._connection_commit_impl() + finally: + # whether or not commit succeeds, cancel any + # nested transactions, make this transaction "inactive" + # and remove it as a reset agent + if self.connection._nested_transaction: + self.connection._nested_transaction._cancel() + + self._deactivate_from_connection() + + # ...however only remove as the connection's current transaction + # if commit succeeded. otherwise it stays on so that a rollback + # needs to occur. + self.connection._transaction = None + else: + if self.connection._transaction is self: + self.connection._invalid_transaction() + else: + raise exc.InvalidRequestError("This transaction is inactive") + + assert not self.is_active + assert self.connection._transaction is not self + + +class NestedTransaction(Transaction): + """Represent a 'nested', or SAVEPOINT transaction. + + The :class:`.NestedTransaction` object is created by calling the + :meth:`_engine.Connection.begin_nested` method of + :class:`_engine.Connection`. + + When using :class:`.NestedTransaction`, the semantics of "begin" / + "commit" / "rollback" are as follows: + + * the "begin" operation corresponds to the "BEGIN SAVEPOINT" command, where + the savepoint is given an explicit name that is part of the state + of this object. + + * The :meth:`.NestedTransaction.commit` method corresponds to a + "RELEASE SAVEPOINT" operation, using the savepoint identifier associated + with this :class:`.NestedTransaction`. + + * The :meth:`.NestedTransaction.rollback` method corresponds to a + "ROLLBACK TO SAVEPOINT" operation, using the savepoint identifier + associated with this :class:`.NestedTransaction`. + + The rationale for mimicking the semantics of an outer transaction in + terms of savepoints so that code may deal with a "savepoint" transaction + and an "outer" transaction in an agnostic way. + + .. seealso:: + + :ref:`session_begin_nested` - ORM version of the SAVEPOINT API. + + """ + + __slots__ = ("connection", "is_active", "_savepoint", "_previous_nested") + + def __init__(self, connection): + assert connection._transaction is not None + if connection._trans_context_manager: + TransactionalContext._trans_ctx_check(connection) + self.connection = connection + self._savepoint = self.connection._savepoint_impl() + self.is_active = True + self._previous_nested = connection._nested_transaction + connection._nested_transaction = self + + def _deactivate_from_connection(self, warn=True): + if self.connection._nested_transaction is self: + self.connection._nested_transaction = self._previous_nested + elif warn: + util.warn( + "nested transaction already deassociated from connection" + ) + + @property + def _deactivated_from_connection(self): + return self.connection._nested_transaction is not self + + def _cancel(self): + # called by RootTransaction when the outer transaction is + # committed, rolled back, or closed to cancel all savepoints + # without any action being taken + self.is_active = False + self._deactivate_from_connection() + if self._previous_nested: + self._previous_nested._cancel() + + def _close_impl(self, deactivate_from_connection, warn_already_deactive): + try: + if self.is_active and self.connection._transaction.is_active: + self.connection._rollback_to_savepoint_impl(self._savepoint) + finally: + self.is_active = False + + if deactivate_from_connection: + self._deactivate_from_connection(warn=warn_already_deactive) + + assert not self.is_active + if deactivate_from_connection: + assert self.connection._nested_transaction is not self + + def _do_deactivate(self): + self._close_impl(False, False) + + def _do_close(self): + self._close_impl(True, False) + + def _do_rollback(self): + self._close_impl(True, True) + + def _do_commit(self): + if self.is_active: + try: + self.connection._release_savepoint_impl(self._savepoint) + finally: + # nested trans becomes inactive on failed release + # unconditionally. this prevents it from trying to + # emit SQL when it rolls back. + self.is_active = False + + # but only de-associate from connection if it succeeded + self._deactivate_from_connection() + else: + if self.connection._nested_transaction is self: + self.connection._invalid_transaction() + else: + raise exc.InvalidRequestError( + "This nested transaction is inactive" + ) + + +class TwoPhaseTransaction(RootTransaction): + """Represent a two-phase transaction. + + A new :class:`.TwoPhaseTransaction` object may be procured + using the :meth:`_engine.Connection.begin_twophase` method. + + The interface is the same as that of :class:`.Transaction` + with the addition of the :meth:`prepare` method. + + """ + + __slots__ = ("connection", "is_active", "xid", "_is_prepared") + + def __init__(self, connection, xid): + self._is_prepared = False + self.xid = xid + super(TwoPhaseTransaction, self).__init__(connection) + + def prepare(self): + """Prepare this :class:`.TwoPhaseTransaction`. + + After a PREPARE, the transaction can be committed. + + """ + if not self.is_active: + raise exc.InvalidRequestError("This transaction is inactive") + self.connection._prepare_twophase_impl(self.xid) + self._is_prepared = True + + def _connection_begin_impl(self): + self.connection._begin_twophase_impl(self) + + def _connection_rollback_impl(self): + self.connection._rollback_twophase_impl(self.xid, self._is_prepared) + + def _connection_commit_impl(self): + self.connection._commit_twophase_impl(self.xid, self._is_prepared) + + +class Engine(Connectable, log.Identified): + """ + Connects a :class:`~sqlalchemy.pool.Pool` and + :class:`~sqlalchemy.engine.interfaces.Dialect` together to provide a + source of database connectivity and behavior. + + This is the **SQLAlchemy 1.x version** of :class:`_engine.Engine`. For + the :term:`2.0 style` version, which includes some API differences, + see :class:`_future.Engine`. + + An :class:`_engine.Engine` object is instantiated publicly using the + :func:`~sqlalchemy.create_engine` function. + + .. seealso:: + + :doc:`/core/engines` + + :ref:`connections_toplevel` + + """ + + _execution_options = _EMPTY_EXECUTION_OPTS + _has_events = False + _connection_cls = Connection + _sqla_logger_namespace = "sqlalchemy.engine.Engine" + _is_future = False + + _schema_translate_map = None + + def __init__( + self, + pool, + dialect, + url, + logging_name=None, + echo=None, + query_cache_size=500, + execution_options=None, + hide_parameters=False, + ): + self.pool = pool + self.url = url + self.dialect = dialect + if logging_name: + self.logging_name = logging_name + self.echo = echo + self.hide_parameters = hide_parameters + if query_cache_size != 0: + self._compiled_cache = util.LRUCache( + query_cache_size, size_alert=self._lru_size_alert + ) + else: + self._compiled_cache = None + log.instance_logger(self, echoflag=echo) + if execution_options: + self.update_execution_options(**execution_options) + + def _lru_size_alert(self, cache): + if self._should_log_info: + self.logger.info( + "Compiled cache size pruning from %d items to %d. " + "Increase cache size to reduce the frequency of pruning.", + len(cache), + cache.capacity, + ) + + @property + def engine(self): + return self + + def clear_compiled_cache(self): + """Clear the compiled cache associated with the dialect. + + This applies **only** to the built-in cache that is established + via the :paramref:`_engine.create_engine.query_cache_size` parameter. + It will not impact any dictionary caches that were passed via the + :paramref:`.Connection.execution_options.query_cache` parameter. + + .. versionadded:: 1.4 + + """ + if self._compiled_cache: + self._compiled_cache.clear() + + def update_execution_options(self, **opt): + r"""Update the default execution_options dictionary + of this :class:`_engine.Engine`. + + The given keys/values in \**opt are added to the + default execution options that will be used for + all connections. The initial contents of this dictionary + can be sent via the ``execution_options`` parameter + to :func:`_sa.create_engine`. + + .. seealso:: + + :meth:`_engine.Connection.execution_options` + + :meth:`_engine.Engine.execution_options` + + """ + self._execution_options = self._execution_options.union(opt) + self.dispatch.set_engine_execution_options(self, opt) + self.dialect.set_engine_execution_options(self, opt) + + def execution_options(self, **opt): + """Return a new :class:`_engine.Engine` that will provide + :class:`_engine.Connection` objects with the given execution options. + + The returned :class:`_engine.Engine` remains related to the original + :class:`_engine.Engine` in that it shares the same connection pool and + other state: + + * The :class:`_pool.Pool` used by the new :class:`_engine.Engine` + is the + same instance. The :meth:`_engine.Engine.dispose` + method will replace + the connection pool instance for the parent engine as well + as this one. + * Event listeners are "cascaded" - meaning, the new + :class:`_engine.Engine` + inherits the events of the parent, and new events can be associated + with the new :class:`_engine.Engine` individually. + * The logging configuration and logging_name is copied from the parent + :class:`_engine.Engine`. + + The intent of the :meth:`_engine.Engine.execution_options` method is + to implement "sharding" schemes where multiple :class:`_engine.Engine` + objects refer to the same connection pool, but are differentiated + by options that would be consumed by a custom event:: + + primary_engine = create_engine("mysql://") + shard1 = primary_engine.execution_options(shard_id="shard1") + shard2 = primary_engine.execution_options(shard_id="shard2") + + Above, the ``shard1`` engine serves as a factory for + :class:`_engine.Connection` + objects that will contain the execution option + ``shard_id=shard1``, and ``shard2`` will produce + :class:`_engine.Connection` + objects that contain the execution option ``shard_id=shard2``. + + An event handler can consume the above execution option to perform + a schema switch or other operation, given a connection. Below + we emit a MySQL ``use`` statement to switch databases, at the same + time keeping track of which database we've established using the + :attr:`_engine.Connection.info` dictionary, + which gives us a persistent + storage space that follows the DBAPI connection:: + + from sqlalchemy import event + from sqlalchemy.engine import Engine + + shards = {"default": "base", shard_1: "db1", "shard_2": "db2"} + + @event.listens_for(Engine, "before_cursor_execute") + def _switch_shard(conn, cursor, stmt, + params, context, executemany): + shard_id = conn._execution_options.get('shard_id', "default") + current_shard = conn.info.get("current_shard", None) + + if current_shard != shard_id: + cursor.execute("use %s" % shards[shard_id]) + conn.info["current_shard"] = shard_id + + .. seealso:: + + :meth:`_engine.Connection.execution_options` + - update execution options + on a :class:`_engine.Connection` object. + + :meth:`_engine.Engine.update_execution_options` + - update the execution + options for a given :class:`_engine.Engine` in place. + + :meth:`_engine.Engine.get_execution_options` + + + """ + return self._option_cls(self, opt) + + def get_execution_options(self): + """Get the non-SQL options which will take effect during execution. + + .. versionadded: 1.3 + + .. seealso:: + + :meth:`_engine.Engine.execution_options` + """ + return self._execution_options + + @property + def name(self): + """String name of the :class:`~sqlalchemy.engine.interfaces.Dialect` + in use by this :class:`Engine`.""" + + return self.dialect.name + + @property + def driver(self): + """Driver name of the :class:`~sqlalchemy.engine.interfaces.Dialect` + in use by this :class:`Engine`.""" + + return self.dialect.driver + + echo = log.echo_property() + + def __repr__(self): + return "Engine(%r)" % (self.url,) + + def dispose(self, close=True): + """Dispose of the connection pool used by this + :class:`_engine.Engine`. + + A new connection pool is created immediately after the old one has been + disposed. The previous connection pool is disposed either actively, by + closing out all currently checked-in connections in that pool, or + passively, by losing references to it but otherwise not closing any + connections. The latter strategy is more appropriate for an initializer + in a forked Python process. + + :param close: if left at its default of ``True``, has the + effect of fully closing all **currently checked in** + database connections. Connections that are still checked out + will **not** be closed, however they will no longer be associated + with this :class:`_engine.Engine`, + so when they are closed individually, eventually the + :class:`_pool.Pool` which they are associated with will + be garbage collected and they will be closed out fully, if + not already closed on checkin. + + If set to ``False``, the previous connection pool is de-referenced, + and otherwise not touched in any way. + + .. versionadded:: 1.4.33 Added the :paramref:`.Engine.dispose.close` + parameter to allow the replacement of a connection pool in a child + process without interfering with the connections used by the parent + process. + + + .. seealso:: + + :ref:`engine_disposal` + + :ref:`pooling_multiprocessing` + + """ + if close: + self.pool.dispose() + self.pool = self.pool.recreate() + self.dispatch.engine_disposed(self) + + def _execute_default( + self, default, multiparams=(), params=util.EMPTY_DICT + ): + with self.connect() as conn: + return conn._execute_default(default, multiparams, params) + + @contextlib.contextmanager + def _optional_conn_ctx_manager(self, connection=None): + if connection is None: + with self.connect() as conn: + yield conn + else: + yield connection + + class _trans_ctx(object): + def __init__(self, conn, transaction, close_with_result): + self.conn = conn + self.transaction = transaction + self.close_with_result = close_with_result + + def __enter__(self): + self.transaction.__enter__() + return self.conn + + def __exit__(self, type_, value, traceback): + try: + self.transaction.__exit__(type_, value, traceback) + finally: + if not self.close_with_result: + self.conn.close() + + def begin(self, close_with_result=False): + """Return a context manager delivering a :class:`_engine.Connection` + with a :class:`.Transaction` established. + + E.g.:: + + with engine.begin() as conn: + conn.execute( + text("insert into table (x, y, z) values (1, 2, 3)") + ) + conn.execute(text("my_special_procedure(5)")) + + Upon successful operation, the :class:`.Transaction` + is committed. If an error is raised, the :class:`.Transaction` + is rolled back. + + Legacy use only: the ``close_with_result`` flag is normally ``False``, + and indicates that the :class:`_engine.Connection` will be closed when + the operation is complete. When set to ``True``, it indicates the + :class:`_engine.Connection` is in "single use" mode, where the + :class:`_engine.CursorResult` returned by the first call to + :meth:`_engine.Connection.execute` will close the + :class:`_engine.Connection` when that :class:`_engine.CursorResult` has + exhausted all result rows. + + .. seealso:: + + :meth:`_engine.Engine.connect` - procure a + :class:`_engine.Connection` from + an :class:`_engine.Engine`. + + :meth:`_engine.Connection.begin` - start a :class:`.Transaction` + for a particular :class:`_engine.Connection`. + + """ + if self._connection_cls._is_future: + conn = self.connect() + else: + conn = self.connect(close_with_result=close_with_result) + try: + trans = conn.begin() + except: + with util.safe_reraise(): + conn.close() + return Engine._trans_ctx(conn, trans, close_with_result) + + @util.deprecated( + "1.4", + "The :meth:`_engine.Engine.transaction` " + "method is deprecated and will be " + "removed in a future release. Use the :meth:`_engine.Engine.begin` " + "context " + "manager instead.", + ) + def transaction(self, callable_, *args, **kwargs): + r"""Execute the given function within a transaction boundary. + + The function is passed a :class:`_engine.Connection` newly procured + from :meth:`_engine.Engine.connect` as the first argument, + followed by the given \*args and \**kwargs. + + e.g.:: + + def do_something(conn, x, y): + conn.execute(text("some statement"), {'x':x, 'y':y}) + + engine.transaction(do_something, 5, 10) + + The operations inside the function are all invoked within the + context of a single :class:`.Transaction`. + Upon success, the transaction is committed. If an + exception is raised, the transaction is rolled back + before propagating the exception. + + .. note:: + + The :meth:`.transaction` method is superseded by + the usage of the Python ``with:`` statement, which can + be used with :meth:`_engine.Engine.begin`:: + + with engine.begin() as conn: + conn.execute(text("some statement"), {'x':5, 'y':10}) + + .. seealso:: + + :meth:`_engine.Engine.begin` - engine-level transactional + context + + :meth:`_engine.Connection.transaction` + - connection-level version of + :meth:`_engine.Engine.transaction` + + """ + kwargs["_sa_skip_warning"] = True + with self.connect() as conn: + return conn.transaction(callable_, *args, **kwargs) + + @util.deprecated( + "1.4", + "The :meth:`_engine.Engine.run_callable` " + "method is deprecated and will be " + "removed in a future release. Use the :meth:`_engine.Engine.begin` " + "context manager instead.", + ) + def run_callable(self, callable_, *args, **kwargs): + r"""Given a callable object or function, execute it, passing + a :class:`_engine.Connection` as the first argument. + + The given \*args and \**kwargs are passed subsequent + to the :class:`_engine.Connection` argument. + + This function, along with :meth:`_engine.Connection.run_callable`, + allows a function to be run with a :class:`_engine.Connection` + or :class:`_engine.Engine` object without the need to know + which one is being dealt with. + + """ + kwargs["_sa_skip_warning"] = True + with self.connect() as conn: + return conn.run_callable(callable_, *args, **kwargs) + + def _run_ddl_visitor(self, visitorcallable, element, **kwargs): + with self.begin() as conn: + conn._run_ddl_visitor(visitorcallable, element, **kwargs) + + @util.deprecated_20( + ":meth:`_engine.Engine.execute`", + alternative="All statement execution in SQLAlchemy 2.0 is performed " + "by the :meth:`_engine.Connection.execute` method of " + ":class:`_engine.Connection`, " + "or in the ORM by the :meth:`.Session.execute` method of " + ":class:`.Session`.", + ) + def execute(self, statement, *multiparams, **params): + """Executes the given construct and returns a + :class:`_engine.CursorResult`. + + The arguments are the same as those used by + :meth:`_engine.Connection.execute`. + + Here, a :class:`_engine.Connection` is acquired using the + :meth:`_engine.Engine.connect` method, and the statement executed + with that connection. The returned :class:`_engine.CursorResult` + is flagged + such that when the :class:`_engine.CursorResult` is exhausted and its + underlying cursor is closed, the :class:`_engine.Connection` + created here + will also be closed, which allows its associated DBAPI connection + resource to be returned to the connection pool. + + """ + connection = self.connect(close_with_result=True) + return connection.execute(statement, *multiparams, **params) + + @util.deprecated_20( + ":meth:`_engine.Engine.scalar`", + alternative="All statement execution in SQLAlchemy 2.0 is performed " + "by the :meth:`_engine.Connection.execute` method of " + ":class:`_engine.Connection`, " + "or in the ORM by the :meth:`.Session.execute` method of " + ":class:`.Session`; the :meth:`_future.Result.scalar` " + "method can then be " + "used to return a scalar result.", + ) + def scalar(self, statement, *multiparams, **params): + """Executes and returns the first column of the first row. + + The underlying result/cursor is closed after execution. + """ + return self.execute(statement, *multiparams, **params).scalar() + + def _execute_clauseelement( + self, + elem, + multiparams=None, + params=None, + execution_options=_EMPTY_EXECUTION_OPTS, + ): + connection = self.connect(close_with_result=True) + return connection._execute_clauseelement( + elem, multiparams, params, execution_options + ) + + def _execute_compiled( + self, + compiled, + multiparams, + params, + execution_options=_EMPTY_EXECUTION_OPTS, + ): + connection = self.connect(close_with_result=True) + return connection._execute_compiled( + compiled, multiparams, params, execution_options + ) + + def connect(self, close_with_result=False): + """Return a new :class:`_engine.Connection` object. + + The :class:`_engine.Connection` object is a facade that uses a DBAPI + connection internally in order to communicate with the database. This + connection is procured from the connection-holding :class:`_pool.Pool` + referenced by this :class:`_engine.Engine`. When the + :meth:`_engine.Connection.close` method of the + :class:`_engine.Connection` object + is called, the underlying DBAPI connection is then returned to the + connection pool, where it may be used again in a subsequent call to + :meth:`_engine.Engine.connect`. + + """ + + return self._connection_cls(self, close_with_result=close_with_result) + + @util.deprecated( + "1.4", + "The :meth:`_engine.Engine.table_names` " + "method is deprecated and will be " + "removed in a future release. Please refer to " + ":meth:`_reflection.Inspector.get_table_names`.", + ) + def table_names(self, schema=None, connection=None): + """Return a list of all table names available in the database. + + :param schema: Optional, retrieve names from a non-default schema. + + :param connection: Optional, use a specified connection. + """ + with self._optional_conn_ctx_manager(connection) as conn: + insp = inspection.inspect(conn) + return insp.get_table_names(schema) + + @util.deprecated( + "1.4", + "The :meth:`_engine.Engine.has_table` " + "method is deprecated and will be " + "removed in a future release. Please refer to " + ":meth:`_reflection.Inspector.has_table`.", + ) + def has_table(self, table_name, schema=None): + """Return True if the given backend has a table of the given name. + + .. seealso:: + + :ref:`metadata_reflection_inspector` - detailed schema inspection + using the :class:`_reflection.Inspector` interface. + + :class:`.quoted_name` - used to pass quoting information along + with a schema identifier. + + """ + with self._optional_conn_ctx_manager(None) as conn: + insp = inspection.inspect(conn) + return insp.has_table(table_name, schema=schema) + + def _wrap_pool_connect(self, fn, connection): + dialect = self.dialect + try: + return fn() + except dialect.dbapi.Error as e: + if connection is None: + Connection._handle_dbapi_exception_noconnection( + e, dialect, self + ) + else: + util.raise_( + sys.exc_info()[1], with_traceback=sys.exc_info()[2] + ) + + def raw_connection(self, _connection=None): + """Return a "raw" DBAPI connection from the connection pool. + + The returned object is a proxied version of the DBAPI + connection object used by the underlying driver in use. + The object will have all the same behavior as the real DBAPI + connection, except that its ``close()`` method will result in the + connection being returned to the pool, rather than being closed + for real. + + This method provides direct DBAPI connection access for + special situations when the API provided by + :class:`_engine.Connection` + is not needed. When a :class:`_engine.Connection` object is already + present, the DBAPI connection is available using + the :attr:`_engine.Connection.connection` accessor. + + .. seealso:: + + :ref:`dbapi_connections` + + """ + return self._wrap_pool_connect(self.pool.connect, _connection) + + +class OptionEngineMixin(object): + _sa_propagate_class_events = False + + def __init__(self, proxied, execution_options): + self._proxied = proxied + self.url = proxied.url + self.dialect = proxied.dialect + self.logging_name = proxied.logging_name + self.echo = proxied.echo + self._compiled_cache = proxied._compiled_cache + self.hide_parameters = proxied.hide_parameters + log.instance_logger(self, echoflag=self.echo) + + # note: this will propagate events that are assigned to the parent + # engine after this OptionEngine is created. Since we share + # the events of the parent we also disallow class-level events + # to apply to the OptionEngine class directly. + # + # the other way this can work would be to transfer existing + # events only, using: + # self.dispatch._update(proxied.dispatch) + # + # that might be more appropriate however it would be a behavioral + # change for logic that assigns events to the parent engine and + # would like it to take effect for the already-created sub-engine. + self.dispatch = self.dispatch._join(proxied.dispatch) + + self._execution_options = proxied._execution_options + self.update_execution_options(**execution_options) + + def _get_pool(self): + return self._proxied.pool + + def _set_pool(self, pool): + self._proxied.pool = pool + + pool = property(_get_pool, _set_pool) + + def _get_has_events(self): + return self._proxied._has_events or self.__dict__.get( + "_has_events", False + ) + + def _set_has_events(self, value): + self.__dict__["_has_events"] = value + + _has_events = property(_get_has_events, _set_has_events) + + +class OptionEngine(OptionEngineMixin, Engine): + pass + + +Engine._option_cls = OptionEngine |