Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# engine/base.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7"""Defines :class:`_engine.Connection` and :class:`_engine.Engine`."""
9from __future__ import annotations
11import contextlib
12import sys
13import typing
14from typing import Any
15from typing import Callable
16from typing import cast
17from typing import Iterable
18from typing import Iterator
19from typing import List
20from typing import Mapping
21from typing import NoReturn
22from typing import Optional
23from typing import overload
24from typing import Tuple
25from typing import Type
26from typing import TypeVar
27from typing import Union
29from .interfaces import BindTyping
30from .interfaces import ConnectionEventsTarget
31from .interfaces import DBAPICursor
32from .interfaces import ExceptionContext
33from .interfaces import ExecuteStyle
34from .interfaces import ExecutionContext
35from .interfaces import IsolationLevel
36from .util import _distill_params_20
37from .util import _distill_raw_params
38from .util import TransactionalContext
39from .. import exc
40from .. import inspection
41from .. import log
42from .. import util
43from ..sql import compiler
44from ..sql import util as sql_util
45from ..util.typing import Never
46from ..util.typing import TupleAny
47from ..util.typing import TypeVarTuple
48from ..util.typing import Unpack
50if typing.TYPE_CHECKING:
51 from . import CursorResult
52 from . import ScalarResult
53 from .interfaces import _AnyExecuteParams
54 from .interfaces import _AnyMultiExecuteParams
55 from .interfaces import _CoreAnyExecuteParams
56 from .interfaces import _CoreMultiExecuteParams
57 from .interfaces import _CoreSingleExecuteParams
58 from .interfaces import _DBAPIAnyExecuteParams
59 from .interfaces import _DBAPISingleExecuteParams
60 from .interfaces import _ExecuteOptions
61 from .interfaces import CompiledCacheType
62 from .interfaces import CoreExecuteOptionsParameter
63 from .interfaces import Dialect
64 from .interfaces import SchemaTranslateMapType
65 from .reflection import Inspector # noqa
66 from .url import URL
67 from ..event import dispatcher
68 from ..log import _EchoFlagType
69 from ..pool import _ConnectionFairy
70 from ..pool import Pool
71 from ..pool import PoolProxiedConnection
72 from ..sql import Executable
73 from ..sql._typing import _InfoType
74 from ..sql.compiler import Compiled
75 from ..sql.ddl import ExecutableDDLElement
76 from ..sql.ddl import InvokeDDLBase
77 from ..sql.functions import FunctionElement
78 from ..sql.schema import DefaultGenerator
79 from ..sql.schema import HasSchemaAttr
80 from ..sql.schema import SchemaVisitable
81 from ..sql.selectable import TypedReturnsRows
84_T = TypeVar("_T", bound=Any)
85_Ts = TypeVarTuple("_Ts")
86_EMPTY_EXECUTION_OPTS: _ExecuteOptions = util.EMPTY_DICT
87NO_OPTIONS: Mapping[str, Any] = util.EMPTY_DICT
90class Connection(ConnectionEventsTarget, inspection.Inspectable["Inspector"]):
91 """Provides high-level functionality for a wrapped DB-API connection.
93 The :class:`_engine.Connection` object is procured by calling the
94 :meth:`_engine.Engine.connect` method of the :class:`_engine.Engine`
95 object, and provides services for execution of SQL statements as well
96 as transaction control.
98 The Connection object is **not** thread-safe. While a Connection can be
99 shared among threads using properly synchronized access, it is still
100 possible that the underlying DBAPI connection may not support shared
101 access between threads. Check the DBAPI documentation for details.
103 The Connection object represents a single DBAPI connection checked out
104 from the connection pool. In this state, the connection pool has no
105 affect upon the connection, including its expiration or timeout state.
106 For the connection pool to properly manage connections, connections
107 should be returned to the connection pool (i.e. ``connection.close()``)
108 whenever the connection is not in use.
110 .. index::
111 single: thread safety; Connection
113 """
115 dialect: Dialect
116 dispatch: dispatcher[ConnectionEventsTarget]
118 _sqla_logger_namespace = "sqlalchemy.engine.Connection"
120 # used by sqlalchemy.engine.util.TransactionalContext
121 _trans_context_manager: Optional[TransactionalContext] = None
123 # legacy as of 2.0, should be eventually deprecated and
124 # removed. was used in the "pre_ping" recipe that's been in the docs
125 # a long time
126 should_close_with_result = False
128 _dbapi_connection: Optional[PoolProxiedConnection]
130 _execution_options: _ExecuteOptions
132 _transaction: Optional[RootTransaction]
133 _nested_transaction: Optional[NestedTransaction]
135 def __init__(
136 self,
137 engine: Engine,
138 connection: Optional[PoolProxiedConnection] = None,
139 _has_events: Optional[bool] = None,
140 _allow_revalidate: bool = True,
141 _allow_autobegin: bool = True,
142 ):
143 """Construct a new Connection."""
144 self.engine = engine
145 self.dialect = dialect = engine.dialect
147 if connection is None:
148 try:
149 self._dbapi_connection = engine.raw_connection()
150 except dialect.loaded_dbapi.Error as err:
151 Connection._handle_dbapi_exception_noconnection(
152 err, dialect, engine
153 )
154 raise
155 else:
156 self._dbapi_connection = connection
158 self._transaction = self._nested_transaction = None
159 self.__savepoint_seq = 0
160 self.__in_begin = False
162 self.__can_reconnect = _allow_revalidate
163 self._allow_autobegin = _allow_autobegin
164 self._echo = self.engine._should_log_info()
166 if _has_events is None:
167 # if _has_events is sent explicitly as False,
168 # then don't join the dispatch of the engine; we don't
169 # want to handle any of the engine's events in that case.
170 self.dispatch = self.dispatch._join(engine.dispatch)
171 self._has_events = _has_events or (
172 _has_events is None and engine._has_events
173 )
175 self._execution_options = engine._execution_options
177 if self._has_events or self.engine._has_events:
178 self.dispatch.engine_connect(self)
180 # this can be assigned differently via
181 # characteristics.LoggingTokenCharacteristic
182 _message_formatter: Any = None
184 def _log_info(self, message: str, *arg: Any, **kw: Any) -> None:
185 fmt = self._message_formatter
187 if fmt:
188 message = fmt(message)
190 if log.STACKLEVEL:
191 kw["stacklevel"] = 1 + log.STACKLEVEL_OFFSET
193 self.engine.logger.info(message, *arg, **kw)
195 def _log_debug(self, message: str, *arg: Any, **kw: Any) -> None:
196 fmt = self._message_formatter
198 if fmt:
199 message = fmt(message)
201 if log.STACKLEVEL:
202 kw["stacklevel"] = 1 + log.STACKLEVEL_OFFSET
204 self.engine.logger.debug(message, *arg, **kw)
206 @property
207 def _schema_translate_map(self) -> Optional[SchemaTranslateMapType]:
208 schema_translate_map: Optional[SchemaTranslateMapType] = (
209 self._execution_options.get("schema_translate_map", None)
210 )
212 return schema_translate_map
214 def schema_for_object(self, obj: HasSchemaAttr) -> Optional[str]:
215 """Return the schema name for the given schema item taking into
216 account current schema translate map.
218 """
220 name = obj.schema
221 schema_translate_map: Optional[SchemaTranslateMapType] = (
222 self._execution_options.get("schema_translate_map", None)
223 )
225 if (
226 schema_translate_map
227 and name in schema_translate_map
228 and obj._use_schema_map
229 ):
230 return schema_translate_map[name]
231 else:
232 return name
234 def __enter__(self) -> Connection:
235 return self
237 def __exit__(self, type_: Any, value: Any, traceback: Any) -> None:
238 self.close()
240 @overload
241 def execution_options(
242 self,
243 *,
244 compiled_cache: Optional[CompiledCacheType] = ...,
245 logging_token: str = ...,
246 isolation_level: IsolationLevel = ...,
247 no_parameters: bool = False,
248 stream_results: bool = False,
249 max_row_buffer: int = ...,
250 yield_per: int = ...,
251 insertmanyvalues_page_size: int = ...,
252 schema_translate_map: Optional[SchemaTranslateMapType] = ...,
253 preserve_rowcount: bool = False,
254 driver_column_names: bool = False,
255 **opt: Any,
256 ) -> Connection: ...
258 @overload
259 def execution_options(self, **opt: Any) -> Connection: ...
261 def execution_options(self, **opt: Any) -> Connection:
262 r"""Set non-SQL options for the connection which take effect
263 during execution.
265 This method modifies this :class:`_engine.Connection` **in-place**;
266 the return value is the same :class:`_engine.Connection` object
267 upon which the method is called. Note that this is in contrast
268 to the behavior of the ``execution_options`` methods on other
269 objects such as :meth:`_engine.Engine.execution_options` and
270 :meth:`_sql.Executable.execution_options`. The rationale is that many
271 such execution options necessarily modify the state of the base
272 DBAPI connection in any case so there is no feasible means of
273 keeping the effect of such an option localized to a "sub" connection.
275 .. versionchanged:: 2.0 The :meth:`_engine.Connection.execution_options`
276 method, in contrast to other objects with this method, modifies
277 the connection in-place without creating copy of it.
279 As discussed elsewhere, the :meth:`_engine.Connection.execution_options`
280 method accepts any arbitrary parameters including user defined names.
281 All parameters given are consumable in a number of ways including
282 by using the :meth:`_engine.Connection.get_execution_options` method.
283 See the examples at :meth:`_sql.Executable.execution_options`
284 and :meth:`_engine.Engine.execution_options`.
286 The keywords that are currently recognized by SQLAlchemy itself
287 include all those listed under :meth:`.Executable.execution_options`,
288 as well as others that are specific to :class:`_engine.Connection`.
290 :param compiled_cache: Available on: :class:`_engine.Connection`,
291 :class:`_engine.Engine`.
293 A dictionary where :class:`.Compiled` objects
294 will be cached when the :class:`_engine.Connection`
295 compiles a clause
296 expression into a :class:`.Compiled` object. This dictionary will
297 supersede the statement cache that may be configured on the
298 :class:`_engine.Engine` itself. If set to None, caching
299 is disabled, even if the engine has a configured cache size.
301 Note that the ORM makes use of its own "compiled" caches for
302 some operations, including flush operations. The caching
303 used by the ORM internally supersedes a cache dictionary
304 specified here.
306 :param logging_token: Available on: :class:`_engine.Connection`,
307 :class:`_engine.Engine`, :class:`_sql.Executable`.
309 Adds the specified string token surrounded by brackets in log
310 messages logged by the connection, i.e. the logging that's enabled
311 either via the :paramref:`_sa.create_engine.echo` flag or via the
312 ``logging.getLogger("sqlalchemy.engine")`` logger. This allows a
313 per-connection or per-sub-engine token to be available which is
314 useful for debugging concurrent connection scenarios.
316 .. versionadded:: 1.4.0b2
318 .. seealso::
320 :ref:`dbengine_logging_tokens` - usage example
322 :paramref:`_sa.create_engine.logging_name` - adds a name to the
323 name used by the Python logger object itself.
325 :param isolation_level: Available on: :class:`_engine.Connection`,
326 :class:`_engine.Engine`.
328 Set the transaction isolation level for the lifespan of this
329 :class:`_engine.Connection` object.
330 Valid values include those string
331 values accepted by the :paramref:`_sa.create_engine.isolation_level`
332 parameter passed to :func:`_sa.create_engine`. These levels are
333 semi-database specific; see individual dialect documentation for
334 valid levels.
336 The isolation level option applies the isolation level by emitting
337 statements on the DBAPI connection, and **necessarily affects the
338 original Connection object overall**. The isolation level will remain
339 at the given setting until explicitly changed, or when the DBAPI
340 connection itself is :term:`released` to the connection pool, i.e. the
341 :meth:`_engine.Connection.close` method is called, at which time an
342 event handler will emit additional statements on the DBAPI connection
343 in order to revert the isolation level change.
345 .. note:: The ``isolation_level`` execution option may only be
346 established before the :meth:`_engine.Connection.begin` method is
347 called, as well as before any SQL statements are emitted which
348 would otherwise trigger "autobegin", or directly after a call to
349 :meth:`_engine.Connection.commit` or
350 :meth:`_engine.Connection.rollback`. A database cannot change the
351 isolation level on a transaction in progress.
353 .. note:: The ``isolation_level`` execution option is implicitly
354 reset if the :class:`_engine.Connection` is invalidated, e.g. via
355 the :meth:`_engine.Connection.invalidate` method, or if a
356 disconnection error occurs. The new connection produced after the
357 invalidation will **not** have the selected isolation level
358 re-applied to it automatically.
360 .. seealso::
362 :ref:`dbapi_autocommit`
364 :meth:`_engine.Connection.get_isolation_level`
365 - view current actual level
367 :param no_parameters: Available on: :class:`_engine.Connection`,
368 :class:`_sql.Executable`.
370 When ``True``, if the final parameter
371 list or dictionary is totally empty, will invoke the
372 statement on the cursor as ``cursor.execute(statement)``,
373 not passing the parameter collection at all.
374 Some DBAPIs such as psycopg2 and mysql-python consider
375 percent signs as significant only when parameters are
376 present; this option allows code to generate SQL
377 containing percent signs (and possibly other characters)
378 that is neutral regarding whether it's executed by the DBAPI
379 or piped into a script that's later invoked by
380 command line tools.
382 :param stream_results: Available on: :class:`_engine.Connection`,
383 :class:`_sql.Executable`.
385 Indicate to the dialect that results should be "streamed" and not
386 pre-buffered, if possible. For backends such as PostgreSQL, MySQL
387 and MariaDB, this indicates the use of a "server side cursor" as
388 opposed to a client side cursor. Other backends such as that of
389 Oracle Database may already use server side cursors by default.
391 The usage of
392 :paramref:`_engine.Connection.execution_options.stream_results` is
393 usually combined with setting a fixed number of rows to to be fetched
394 in batches, to allow for efficient iteration of database rows while
395 at the same time not loading all result rows into memory at once;
396 this can be configured on a :class:`_engine.Result` object using the
397 :meth:`_engine.Result.yield_per` method, after execution has
398 returned a new :class:`_engine.Result`. If
399 :meth:`_engine.Result.yield_per` is not used,
400 the :paramref:`_engine.Connection.execution_options.stream_results`
401 mode of operation will instead use a dynamically sized buffer
402 which buffers sets of rows at a time, growing on each batch
403 based on a fixed growth size up until a limit which may
404 be configured using the
405 :paramref:`_engine.Connection.execution_options.max_row_buffer`
406 parameter.
408 When using the ORM to fetch ORM mapped objects from a result,
409 :meth:`_engine.Result.yield_per` should always be used with
410 :paramref:`_engine.Connection.execution_options.stream_results`,
411 so that the ORM does not fetch all rows into new ORM objects at once.
413 For typical use, the
414 :paramref:`_engine.Connection.execution_options.yield_per` execution
415 option should be preferred, which sets up both
416 :paramref:`_engine.Connection.execution_options.stream_results` and
417 :meth:`_engine.Result.yield_per` at once. This option is supported
418 both at a core level by :class:`_engine.Connection` as well as by the
419 ORM :class:`_engine.Session`; the latter is described at
420 :ref:`orm_queryguide_yield_per`.
422 .. seealso::
424 :ref:`engine_stream_results` - background on
425 :paramref:`_engine.Connection.execution_options.stream_results`
427 :paramref:`_engine.Connection.execution_options.max_row_buffer`
429 :paramref:`_engine.Connection.execution_options.yield_per`
431 :ref:`orm_queryguide_yield_per` - in the :ref:`queryguide_toplevel`
432 describing the ORM version of ``yield_per``
434 :param max_row_buffer: Available on: :class:`_engine.Connection`,
435 :class:`_sql.Executable`. Sets a maximum
436 buffer size to use when the
437 :paramref:`_engine.Connection.execution_options.stream_results`
438 execution option is used on a backend that supports server side
439 cursors. The default value if not specified is 1000.
441 .. seealso::
443 :paramref:`_engine.Connection.execution_options.stream_results`
445 :ref:`engine_stream_results`
448 :param yield_per: Available on: :class:`_engine.Connection`,
449 :class:`_sql.Executable`. Integer value applied which will
450 set the :paramref:`_engine.Connection.execution_options.stream_results`
451 execution option and invoke :meth:`_engine.Result.yield_per`
452 automatically at once. Allows equivalent functionality as
453 is present when using this parameter with the ORM.
455 .. versionadded:: 1.4.40
457 .. seealso::
459 :ref:`engine_stream_results` - background and examples
460 on using server side cursors with Core.
462 :ref:`orm_queryguide_yield_per` - in the :ref:`queryguide_toplevel`
463 describing the ORM version of ``yield_per``
465 :param insertmanyvalues_page_size: Available on: :class:`_engine.Connection`,
466 :class:`_engine.Engine`. Number of rows to format into an
467 INSERT statement when the statement uses "insertmanyvalues" mode,
468 which is a paged form of bulk insert that is used for many backends
469 when using :term:`executemany` execution typically in conjunction
470 with RETURNING. Defaults to 1000. May also be modified on a
471 per-engine basis using the
472 :paramref:`_sa.create_engine.insertmanyvalues_page_size` parameter.
474 .. versionadded:: 2.0
476 .. seealso::
478 :ref:`engine_insertmanyvalues`
480 :param schema_translate_map: Available on: :class:`_engine.Connection`,
481 :class:`_engine.Engine`, :class:`_sql.Executable`.
483 A dictionary mapping schema names to schema names, that will be
484 applied to the :paramref:`_schema.Table.schema` element of each
485 :class:`_schema.Table`
486 encountered when SQL or DDL expression elements
487 are compiled into strings; the resulting schema name will be
488 converted based on presence in the map of the original name.
490 .. seealso::
492 :ref:`schema_translating`
494 :param preserve_rowcount: Boolean; when True, the ``cursor.rowcount``
495 attribute will be unconditionally memoized within the result and
496 made available via the :attr:`.CursorResult.rowcount` attribute.
497 Normally, this attribute is only preserved for UPDATE and DELETE
498 statements. Using this option, the DBAPIs rowcount value can
499 be accessed for other kinds of statements such as INSERT and SELECT,
500 to the degree that the DBAPI supports these statements. See
501 :attr:`.CursorResult.rowcount` for notes regarding the behavior
502 of this attribute.
504 .. versionadded:: 2.0.28
506 .. seealso::
508 :meth:`_engine.Engine.execution_options`
510 :meth:`.Executable.execution_options`
512 :meth:`_engine.Connection.get_execution_options`
514 :ref:`orm_queryguide_execution_options` - documentation on all
515 ORM-specific execution options
517 :param driver_column_names: When True, the returned
518 :class:`_engine.CursorResult` will use the column names as written in
519 ``cursor.description`` to set up the keys for the result set,
520 including the names of columns for the :class:`_engine.Row` object as
521 well as the dictionary keys when using :attr:`_engine.Row._mapping`.
522 On backends that use "name normalization" such as Oracle Database to
523 correct for lower case names being converted to all uppercase, this
524 behavior is turned off and the raw UPPERCASE names in
525 cursor.description will be present.
527 .. versionadded:: 2.1
529 """ # noqa
530 if self._has_events or self.engine._has_events:
531 self.dispatch.set_connection_execution_options(self, opt)
532 self._execution_options = self._execution_options.union(opt)
533 self.dialect.set_connection_execution_options(self, opt)
534 return self
536 def get_execution_options(self) -> _ExecuteOptions:
537 """Get the non-SQL options which will take effect during execution.
539 .. seealso::
541 :meth:`_engine.Connection.execution_options`
542 """
543 return self._execution_options
545 @property
546 def _still_open_and_dbapi_connection_is_valid(self) -> bool:
547 pool_proxied_connection = self._dbapi_connection
548 return (
549 pool_proxied_connection is not None
550 and pool_proxied_connection.is_valid
551 )
553 @property
554 def closed(self) -> bool:
555 """Return True if this connection is closed."""
557 return self._dbapi_connection is None and not self.__can_reconnect
559 @property
560 def invalidated(self) -> bool:
561 """Return True if this connection was invalidated.
563 This does not indicate whether or not the connection was
564 invalidated at the pool level, however
566 """
568 # prior to 1.4, "invalid" was stored as a state independent of
569 # "closed", meaning an invalidated connection could be "closed",
570 # the _dbapi_connection would be None and closed=True, yet the
571 # "invalid" flag would stay True. This meant that there were
572 # three separate states (open/valid, closed/valid, closed/invalid)
573 # when there is really no reason for that; a connection that's
574 # "closed" does not need to be "invalid". So the state is now
575 # represented by the two facts alone.
577 pool_proxied_connection = self._dbapi_connection
578 return pool_proxied_connection is None and self.__can_reconnect
580 @property
581 def connection(self) -> PoolProxiedConnection:
582 """The underlying DB-API connection managed by this Connection.
584 This is a SQLAlchemy connection-pool proxied connection
585 which then has the attribute
586 :attr:`_pool._ConnectionFairy.dbapi_connection` that refers to the
587 actual driver connection.
589 .. seealso::
592 :ref:`dbapi_connections`
594 """
596 if self._dbapi_connection is None:
597 try:
598 return self._revalidate_connection()
599 except (exc.PendingRollbackError, exc.ResourceClosedError):
600 raise
601 except BaseException as e:
602 self._handle_dbapi_exception(e, None, None, None, None)
603 else:
604 return self._dbapi_connection
606 def get_isolation_level(self) -> IsolationLevel:
607 """Return the current **actual** isolation level that's present on
608 the database within the scope of this connection.
610 This attribute will perform a live SQL operation against the database
611 in order to procure the current isolation level, so the value returned
612 is the actual level on the underlying DBAPI connection regardless of
613 how this state was set. This will be one of the four actual isolation
614 modes ``READ UNCOMMITTED``, ``READ COMMITTED``, ``REPEATABLE READ``,
615 ``SERIALIZABLE``. It will **not** include the ``AUTOCOMMIT`` isolation
616 level setting. Third party dialects may also feature additional
617 isolation level settings.
619 .. note:: This method **will not report** on the ``AUTOCOMMIT``
620 isolation level, which is a separate :term:`dbapi` setting that's
621 independent of **actual** isolation level. When ``AUTOCOMMIT`` is
622 in use, the database connection still has a "traditional" isolation
623 mode in effect, that is typically one of the four values
624 ``READ UNCOMMITTED``, ``READ COMMITTED``, ``REPEATABLE READ``,
625 ``SERIALIZABLE``.
627 Compare to the :attr:`_engine.Connection.default_isolation_level`
628 accessor which returns the isolation level that is present on the
629 database at initial connection time.
631 .. seealso::
633 :attr:`_engine.Connection.default_isolation_level`
634 - view default level
636 :paramref:`_sa.create_engine.isolation_level`
637 - set per :class:`_engine.Engine` isolation level
639 :paramref:`.Connection.execution_options.isolation_level`
640 - set per :class:`_engine.Connection` isolation level
642 """
643 dbapi_connection = self.connection.dbapi_connection
644 assert dbapi_connection is not None
645 try:
646 return self.dialect.get_isolation_level(dbapi_connection)
647 except BaseException as e:
648 self._handle_dbapi_exception(e, None, None, None, None)
650 @property
651 def default_isolation_level(self) -> Optional[IsolationLevel]:
652 """The initial-connection time isolation level associated with the
653 :class:`_engine.Dialect` in use.
655 This value is independent of the
656 :paramref:`.Connection.execution_options.isolation_level` and
657 :paramref:`.Engine.execution_options.isolation_level` execution
658 options, and is determined by the :class:`_engine.Dialect` when the
659 first connection is created, by performing a SQL query against the
660 database for the current isolation level before any additional commands
661 have been emitted.
663 Calling this accessor does not invoke any new SQL queries.
665 .. seealso::
667 :meth:`_engine.Connection.get_isolation_level`
668 - view current actual isolation level
670 :paramref:`_sa.create_engine.isolation_level`
671 - set per :class:`_engine.Engine` isolation level
673 :paramref:`.Connection.execution_options.isolation_level`
674 - set per :class:`_engine.Connection` isolation level
676 """
677 return self.dialect.default_isolation_level
679 def _invalid_transaction(self) -> NoReturn:
680 raise exc.PendingRollbackError(
681 "Can't reconnect until invalid %stransaction is rolled "
682 "back. Please rollback() fully before proceeding"
683 % ("savepoint " if self._nested_transaction is not None else ""),
684 code="8s2b",
685 )
687 def _revalidate_connection(self) -> PoolProxiedConnection:
688 if self.__can_reconnect and self.invalidated:
689 if self._transaction is not None:
690 self._invalid_transaction()
691 self._dbapi_connection = self.engine.raw_connection()
692 return self._dbapi_connection
693 raise exc.ResourceClosedError("This Connection is closed")
695 @property
696 def info(self) -> _InfoType:
697 """Info dictionary associated with the underlying DBAPI connection
698 referred to by this :class:`_engine.Connection`, allowing user-defined
699 data to be associated with the connection.
701 The data here will follow along with the DBAPI connection including
702 after it is returned to the connection pool and used again
703 in subsequent instances of :class:`_engine.Connection`.
705 """
707 return self.connection.info
709 def invalidate(self, exception: Optional[BaseException] = None) -> None:
710 """Invalidate the underlying DBAPI connection associated with
711 this :class:`_engine.Connection`.
713 An attempt will be made to close the underlying DBAPI connection
714 immediately; however if this operation fails, the error is logged
715 but not raised. The connection is then discarded whether or not
716 close() succeeded.
718 Upon the next use (where "use" typically means using the
719 :meth:`_engine.Connection.execute` method or similar),
720 this :class:`_engine.Connection` will attempt to
721 procure a new DBAPI connection using the services of the
722 :class:`_pool.Pool` as a source of connectivity (e.g.
723 a "reconnection").
725 If a transaction was in progress (e.g. the
726 :meth:`_engine.Connection.begin` method has been called) when
727 :meth:`_engine.Connection.invalidate` method is called, at the DBAPI
728 level all state associated with this transaction is lost, as
729 the DBAPI connection is closed. The :class:`_engine.Connection`
730 will not allow a reconnection to proceed until the
731 :class:`.Transaction` object is ended, by calling the
732 :meth:`.Transaction.rollback` method; until that point, any attempt at
733 continuing to use the :class:`_engine.Connection` will raise an
734 :class:`~sqlalchemy.exc.InvalidRequestError`.
735 This is to prevent applications from accidentally
736 continuing an ongoing transactional operations despite the
737 fact that the transaction has been lost due to an
738 invalidation.
740 The :meth:`_engine.Connection.invalidate` method,
741 just like auto-invalidation,
742 will at the connection pool level invoke the
743 :meth:`_events.PoolEvents.invalidate` event.
745 :param exception: an optional ``Exception`` instance that's the
746 reason for the invalidation. is passed along to event handlers
747 and logging functions.
749 .. seealso::
751 :ref:`pool_connection_invalidation`
753 """
755 if self.invalidated:
756 return
758 if self.closed:
759 raise exc.ResourceClosedError("This Connection is closed")
761 if self._still_open_and_dbapi_connection_is_valid:
762 pool_proxied_connection = self._dbapi_connection
763 assert pool_proxied_connection is not None
764 pool_proxied_connection.invalidate(exception)
766 self._dbapi_connection = None
768 def detach(self) -> None:
769 """Detach the underlying DB-API connection from its connection pool.
771 E.g.::
773 with engine.connect() as conn:
774 conn.detach()
775 conn.execute(text("SET search_path TO schema1, schema2"))
777 # work with connection
779 # connection is fully closed (since we used "with:", can
780 # also call .close())
782 This :class:`_engine.Connection` instance will remain usable.
783 When closed
784 (or exited from a context manager context as above),
785 the DB-API connection will be literally closed and not
786 returned to its originating pool.
788 This method can be used to insulate the rest of an application
789 from a modified state on a connection (such as a transaction
790 isolation level or similar).
792 """
794 if self.closed:
795 raise exc.ResourceClosedError("This Connection is closed")
797 pool_proxied_connection = self._dbapi_connection
798 if pool_proxied_connection is None:
799 raise exc.InvalidRequestError(
800 "Can't detach an invalidated Connection"
801 )
802 pool_proxied_connection.detach()
804 def _autobegin(self) -> None:
805 if self._allow_autobegin and not self.__in_begin:
806 self.begin()
808 def begin(self) -> RootTransaction:
809 """Begin a transaction prior to autobegin occurring.
811 E.g.::
813 with engine.connect() as conn:
814 with conn.begin() as trans:
815 conn.execute(table.insert(), {"username": "sandy"})
817 The returned object is an instance of :class:`_engine.RootTransaction`.
818 This object represents the "scope" of the transaction,
819 which completes when either the :meth:`_engine.Transaction.rollback`
820 or :meth:`_engine.Transaction.commit` method is called; the object
821 also works as a context manager as illustrated above.
823 The :meth:`_engine.Connection.begin` method begins a
824 transaction that normally will be begun in any case when the connection
825 is first used to execute a statement. The reason this method might be
826 used would be to invoke the :meth:`_events.ConnectionEvents.begin`
827 event at a specific time, or to organize code within the scope of a
828 connection checkout in terms of context managed blocks, such as::
830 with engine.connect() as conn:
831 with conn.begin():
832 conn.execute(...)
833 conn.execute(...)
835 with conn.begin():
836 conn.execute(...)
837 conn.execute(...)
839 The above code is not fundamentally any different in its behavior than
840 the following code which does not use
841 :meth:`_engine.Connection.begin`; the below style is known
842 as "commit as you go" style::
844 with engine.connect() as conn:
845 conn.execute(...)
846 conn.execute(...)
847 conn.commit()
849 conn.execute(...)
850 conn.execute(...)
851 conn.commit()
853 From a database point of view, the :meth:`_engine.Connection.begin`
854 method does not emit any SQL or change the state of the underlying
855 DBAPI connection in any way; the Python DBAPI does not have any
856 concept of explicit transaction begin.
858 .. seealso::
860 :ref:`tutorial_working_with_transactions` - in the
861 :ref:`unified_tutorial`
863 :meth:`_engine.Connection.begin_nested` - use a SAVEPOINT
865 :meth:`_engine.Connection.begin_twophase` -
866 use a two phase /XID transaction
868 :meth:`_engine.Engine.begin` - context manager available from
869 :class:`_engine.Engine`
871 """
872 if self._transaction is None:
873 self._transaction = RootTransaction(self)
874 return self._transaction
875 else:
876 raise exc.InvalidRequestError(
877 "This connection has already initialized a SQLAlchemy "
878 "Transaction() object via begin() or autobegin; can't "
879 "call begin() here unless rollback() or commit() "
880 "is called first."
881 )
883 def begin_nested(self) -> NestedTransaction:
884 """Begin a nested transaction (i.e. SAVEPOINT) and return a transaction
885 handle that controls the scope of the SAVEPOINT.
887 E.g.::
889 with engine.begin() as connection:
890 with connection.begin_nested():
891 connection.execute(table.insert(), {"username": "sandy"})
893 The returned object is an instance of
894 :class:`_engine.NestedTransaction`, which includes transactional
895 methods :meth:`_engine.NestedTransaction.commit` and
896 :meth:`_engine.NestedTransaction.rollback`; for a nested transaction,
897 these methods correspond to the operations "RELEASE SAVEPOINT <name>"
898 and "ROLLBACK TO SAVEPOINT <name>". The name of the savepoint is local
899 to the :class:`_engine.NestedTransaction` object and is generated
900 automatically. Like any other :class:`_engine.Transaction`, the
901 :class:`_engine.NestedTransaction` may be used as a context manager as
902 illustrated above which will "release" or "rollback" corresponding to
903 if the operation within the block were successful or raised an
904 exception.
906 Nested transactions require SAVEPOINT support in the underlying
907 database, else the behavior is undefined. SAVEPOINT is commonly used to
908 run operations within a transaction that may fail, while continuing the
909 outer transaction. E.g.::
911 from sqlalchemy import exc
913 with engine.begin() as connection:
914 trans = connection.begin_nested()
915 try:
916 connection.execute(table.insert(), {"username": "sandy"})
917 trans.commit()
918 except exc.IntegrityError: # catch for duplicate username
919 trans.rollback() # rollback to savepoint
921 # outer transaction continues
922 connection.execute(...)
924 If :meth:`_engine.Connection.begin_nested` is called without first
925 calling :meth:`_engine.Connection.begin` or
926 :meth:`_engine.Engine.begin`, the :class:`_engine.Connection` object
927 will "autobegin" the outer transaction first. This outer transaction
928 may be committed using "commit-as-you-go" style, e.g.::
930 with engine.connect() as connection: # begin() wasn't called
932 with connection.begin_nested(): # will auto-"begin()" first
933 connection.execute(...)
934 # savepoint is released
936 connection.execute(...)
938 # explicitly commit outer transaction
939 connection.commit()
941 # can continue working with connection here
943 .. versionchanged:: 2.0
945 :meth:`_engine.Connection.begin_nested` will now participate
946 in the connection "autobegin" behavior that is new as of
947 2.0 / "future" style connections in 1.4.
949 .. seealso::
951 :meth:`_engine.Connection.begin`
953 :ref:`session_begin_nested` - ORM support for SAVEPOINT
955 """
956 if self._transaction is None:
957 self._autobegin()
959 return NestedTransaction(self)
961 def begin_twophase(self, xid: Optional[Any] = None) -> TwoPhaseTransaction:
962 """Begin a two-phase or XA transaction and return a transaction
963 handle.
965 The returned object is an instance of :class:`.TwoPhaseTransaction`,
966 which in addition to the methods provided by
967 :class:`.Transaction`, also provides a
968 :meth:`~.TwoPhaseTransaction.prepare` method.
970 :param xid: the two phase transaction id. If not supplied, a
971 random id will be generated. The accepted type and value depends on
972 the driver in use.
974 .. seealso::
976 :meth:`_engine.Connection.begin`
978 :meth:`_engine.Connection.begin_twophase`
980 """
982 if self._transaction is not None:
983 raise exc.InvalidRequestError(
984 "Cannot start a two phase transaction when a transaction "
985 "is already in progress."
986 )
987 if xid is None:
988 xid = self.engine.dialect.create_xid()
989 return TwoPhaseTransaction(self, xid)
991 def commit(self) -> None:
992 """Commit the transaction that is currently in progress.
994 This method commits the current transaction if one has been started.
995 If no transaction was started, the method has no effect, assuming
996 the connection is in a non-invalidated state.
998 A transaction is begun on a :class:`_engine.Connection` automatically
999 whenever a statement is first executed, or when the
1000 :meth:`_engine.Connection.begin` method is called.
1002 .. note:: The :meth:`_engine.Connection.commit` method only acts upon
1003 the primary database transaction that is linked to the
1004 :class:`_engine.Connection` object. It does not operate upon a
1005 SAVEPOINT that would have been invoked from the
1006 :meth:`_engine.Connection.begin_nested` method; for control of a
1007 SAVEPOINT, call :meth:`_engine.NestedTransaction.commit` on the
1008 :class:`_engine.NestedTransaction` that is returned by the
1009 :meth:`_engine.Connection.begin_nested` method itself.
1012 """
1013 if self._transaction:
1014 self._transaction.commit()
1016 def rollback(self) -> None:
1017 """Roll back the transaction that is currently in progress.
1019 This method rolls back the current transaction if one has been started.
1020 If no transaction was started, the method has no effect. If a
1021 transaction was started and the connection is in an invalidated state,
1022 the transaction is cleared using this method.
1024 A transaction is begun on a :class:`_engine.Connection` automatically
1025 whenever a statement is first executed, or when the
1026 :meth:`_engine.Connection.begin` method is called.
1028 .. note:: The :meth:`_engine.Connection.rollback` method only acts
1029 upon the primary database transaction that is linked to the
1030 :class:`_engine.Connection` object. It does not operate upon a
1031 SAVEPOINT that would have been invoked from the
1032 :meth:`_engine.Connection.begin_nested` method; for control of a
1033 SAVEPOINT, call :meth:`_engine.NestedTransaction.rollback` on the
1034 :class:`_engine.NestedTransaction` that is returned by the
1035 :meth:`_engine.Connection.begin_nested` method itself.
1038 """
1039 if self._transaction:
1040 self._transaction.rollback()
1042 def recover_twophase(self) -> List[Any]:
1043 return self.engine.dialect.do_recover_twophase(self)
1045 def rollback_prepared(self, xid: Any, recover: bool = False) -> None:
1046 self.engine.dialect.do_rollback_twophase(self, xid, recover=recover)
1048 def commit_prepared(self, xid: Any, recover: bool = False) -> None:
1049 self.engine.dialect.do_commit_twophase(self, xid, recover=recover)
1051 def in_transaction(self) -> bool:
1052 """Return True if a transaction is in progress."""
1053 return self._transaction is not None and self._transaction.is_active
1055 def in_nested_transaction(self) -> bool:
1056 """Return True if a transaction is in progress."""
1057 return (
1058 self._nested_transaction is not None
1059 and self._nested_transaction.is_active
1060 )
1062 def _is_autocommit_isolation(self) -> bool:
1063 opt_iso = self._execution_options.get("isolation_level", None)
1064 return bool(
1065 opt_iso == "AUTOCOMMIT"
1066 or (
1067 opt_iso is None
1068 and self.engine.dialect._on_connect_isolation_level
1069 == "AUTOCOMMIT"
1070 )
1071 )
1073 def _get_required_transaction(self) -> RootTransaction:
1074 trans = self._transaction
1075 if trans is None:
1076 raise exc.InvalidRequestError("connection is not in a transaction")
1077 return trans
1079 def _get_required_nested_transaction(self) -> NestedTransaction:
1080 trans = self._nested_transaction
1081 if trans is None:
1082 raise exc.InvalidRequestError(
1083 "connection is not in a nested transaction"
1084 )
1085 return trans
1087 def get_transaction(self) -> Optional[RootTransaction]:
1088 """Return the current root transaction in progress, if any.
1090 .. versionadded:: 1.4
1092 """
1094 return self._transaction
1096 def get_nested_transaction(self) -> Optional[NestedTransaction]:
1097 """Return the current nested transaction in progress, if any.
1099 .. versionadded:: 1.4
1101 """
1102 return self._nested_transaction
1104 def _begin_impl(self, transaction: RootTransaction) -> None:
1105 if self._echo:
1106 if self._is_autocommit_isolation():
1107 self._log_info(
1108 "BEGIN (implicit; DBAPI should not BEGIN due to "
1109 "autocommit mode)"
1110 )
1111 else:
1112 self._log_info("BEGIN (implicit)")
1114 self.__in_begin = True
1116 if self._has_events or self.engine._has_events:
1117 self.dispatch.begin(self)
1119 try:
1120 self.engine.dialect.do_begin(self.connection)
1121 except BaseException as e:
1122 self._handle_dbapi_exception(e, None, None, None, None)
1123 finally:
1124 self.__in_begin = False
1126 def _rollback_impl(self) -> None:
1127 if self._has_events or self.engine._has_events:
1128 self.dispatch.rollback(self)
1130 if self._still_open_and_dbapi_connection_is_valid:
1131 if self._echo:
1132 if self._is_autocommit_isolation():
1133 if self.dialect.skip_autocommit_rollback:
1134 self._log_info(
1135 "ROLLBACK will be skipped by "
1136 "skip_autocommit_rollback"
1137 )
1138 else:
1139 self._log_info(
1140 "ROLLBACK using DBAPI connection.rollback(); "
1141 "set skip_autocommit_rollback to prevent fully"
1142 )
1143 else:
1144 self._log_info("ROLLBACK")
1145 try:
1146 self.engine.dialect.do_rollback(self.connection)
1147 except BaseException as e:
1148 self._handle_dbapi_exception(e, None, None, None, None)
1150 def _commit_impl(self) -> None:
1151 if self._has_events or self.engine._has_events:
1152 self.dispatch.commit(self)
1154 if self._echo:
1155 if self._is_autocommit_isolation():
1156 self._log_info(
1157 "COMMIT using DBAPI connection.commit(), "
1158 "has no effect due to autocommit mode"
1159 )
1160 else:
1161 self._log_info("COMMIT")
1162 try:
1163 self.engine.dialect.do_commit(self.connection)
1164 except BaseException as e:
1165 self._handle_dbapi_exception(e, None, None, None, None)
1167 def _savepoint_impl(self, name: Optional[str] = None) -> str:
1168 if self._has_events or self.engine._has_events:
1169 self.dispatch.savepoint(self, name)
1171 if name is None:
1172 self.__savepoint_seq += 1
1173 name = "sa_savepoint_%s" % self.__savepoint_seq
1174 self.engine.dialect.do_savepoint(self, name)
1175 return name
1177 def _rollback_to_savepoint_impl(self, name: str) -> None:
1178 if self._has_events or self.engine._has_events:
1179 self.dispatch.rollback_savepoint(self, name, None)
1181 if self._still_open_and_dbapi_connection_is_valid:
1182 self.engine.dialect.do_rollback_to_savepoint(self, name)
1184 def _release_savepoint_impl(self, name: str) -> None:
1185 if self._has_events or self.engine._has_events:
1186 self.dispatch.release_savepoint(self, name, None)
1188 self.engine.dialect.do_release_savepoint(self, name)
1190 def _begin_twophase_impl(self, transaction: TwoPhaseTransaction) -> None:
1191 if self._echo:
1192 self._log_info("BEGIN TWOPHASE (implicit)")
1193 if self._has_events or self.engine._has_events:
1194 self.dispatch.begin_twophase(self, transaction.xid)
1196 self.__in_begin = True
1197 try:
1198 self.engine.dialect.do_begin_twophase(self, transaction.xid)
1199 except BaseException as e:
1200 self._handle_dbapi_exception(e, None, None, None, None)
1201 finally:
1202 self.__in_begin = False
1204 def _prepare_twophase_impl(self, xid: Any) -> None:
1205 if self._has_events or self.engine._has_events:
1206 self.dispatch.prepare_twophase(self, xid)
1208 assert isinstance(self._transaction, TwoPhaseTransaction)
1209 try:
1210 self.engine.dialect.do_prepare_twophase(self, xid)
1211 except BaseException as e:
1212 self._handle_dbapi_exception(e, None, None, None, None)
1214 def _rollback_twophase_impl(self, xid: Any, is_prepared: bool) -> None:
1215 if self._has_events or self.engine._has_events:
1216 self.dispatch.rollback_twophase(self, xid, is_prepared)
1218 if self._still_open_and_dbapi_connection_is_valid:
1219 assert isinstance(self._transaction, TwoPhaseTransaction)
1220 try:
1221 self.engine.dialect.do_rollback_twophase(
1222 self, xid, is_prepared
1223 )
1224 except BaseException as e:
1225 self._handle_dbapi_exception(e, None, None, None, None)
1227 def _commit_twophase_impl(self, xid: Any, is_prepared: bool) -> None:
1228 if self._has_events or self.engine._has_events:
1229 self.dispatch.commit_twophase(self, xid, is_prepared)
1231 assert isinstance(self._transaction, TwoPhaseTransaction)
1232 try:
1233 self.engine.dialect.do_commit_twophase(self, xid, is_prepared)
1234 except BaseException as e:
1235 self._handle_dbapi_exception(e, None, None, None, None)
1237 def close(self) -> None:
1238 """Close this :class:`_engine.Connection`.
1240 This results in a release of the underlying database
1241 resources, that is, the DBAPI connection referenced
1242 internally. The DBAPI connection is typically restored
1243 back to the connection-holding :class:`_pool.Pool` referenced
1244 by the :class:`_engine.Engine` that produced this
1245 :class:`_engine.Connection`. Any transactional state present on
1246 the DBAPI connection is also unconditionally released via
1247 the DBAPI connection's ``rollback()`` method, regardless
1248 of any :class:`.Transaction` object that may be
1249 outstanding with regards to this :class:`_engine.Connection`.
1251 This has the effect of also calling :meth:`_engine.Connection.rollback`
1252 if any transaction is in place.
1254 After :meth:`_engine.Connection.close` is called, the
1255 :class:`_engine.Connection` is permanently in a closed state,
1256 and will allow no further operations.
1258 """
1260 # if the rollback here raises, skip_reset stays False and the pool
1261 # does its own reset of the connection, invalidating it if that
1262 # fails also. see #13657
1263 skip_reset = False
1264 try:
1265 if self._transaction:
1266 self._transaction.close()
1267 skip_reset = True
1268 finally:
1269 if self._dbapi_connection is not None:
1270 conn = self._dbapi_connection
1272 # as we just closed the transaction, close the connection
1273 # pool connection without doing an additional reset
1274 if skip_reset:
1275 cast("_ConnectionFairy", conn)._close_special(
1276 transaction_reset=True
1277 )
1278 else:
1279 conn.close()
1281 # There is a slight chance that conn.close() may have
1282 # triggered an invalidation here in which case
1283 # _dbapi_connection would already be None, however usually
1284 # it will be non-None here and in a "closed" state.
1285 self._dbapi_connection = None
1286 self.__can_reconnect = False
1288 # special case to handle mypy issue:
1289 # https://github.com/python/mypy/issues/20651
1290 @overload
1291 def scalar(
1292 self,
1293 statement: TypedReturnsRows[Never],
1294 parameters: Optional[_CoreSingleExecuteParams] = None,
1295 *,
1296 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1297 ) -> Optional[Any]: ...
1299 @overload
1300 def scalar(
1301 self,
1302 statement: TypedReturnsRows[_T],
1303 parameters: Optional[_CoreSingleExecuteParams] = None,
1304 *,
1305 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1306 ) -> Optional[_T]: ...
1308 @overload
1309 def scalar(
1310 self,
1311 statement: Executable,
1312 parameters: Optional[_CoreSingleExecuteParams] = None,
1313 *,
1314 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1315 ) -> Any: ...
1317 def scalar(
1318 self,
1319 statement: Executable,
1320 parameters: Optional[_CoreSingleExecuteParams] = None,
1321 *,
1322 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1323 ) -> Any:
1324 r"""Executes a SQL statement construct and returns a scalar object.
1326 This method is shorthand for invoking the
1327 :meth:`_engine.Result.scalar` method after invoking the
1328 :meth:`_engine.Connection.execute` method. Parameters are equivalent.
1330 :return: a scalar Python value representing the first column of the
1331 first row returned.
1333 """
1334 distilled_parameters = _distill_params_20(parameters)
1335 try:
1336 meth = statement._execute_on_scalar
1337 except AttributeError as err:
1338 raise exc.ObjectNotExecutableError(statement) from err
1339 else:
1340 return meth(
1341 self,
1342 distilled_parameters,
1343 execution_options or NO_OPTIONS,
1344 )
1346 @overload
1347 def scalars(
1348 self,
1349 statement: TypedReturnsRows[_T],
1350 parameters: Optional[_CoreAnyExecuteParams] = None,
1351 *,
1352 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1353 ) -> ScalarResult[_T]: ...
1355 @overload
1356 def scalars(
1357 self,
1358 statement: Executable,
1359 parameters: Optional[_CoreAnyExecuteParams] = None,
1360 *,
1361 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1362 ) -> ScalarResult[Any]: ...
1364 def scalars(
1365 self,
1366 statement: Executable,
1367 parameters: Optional[_CoreAnyExecuteParams] = None,
1368 *,
1369 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1370 ) -> ScalarResult[Any]:
1371 """Executes and returns a scalar result set, which yields scalar values
1372 from the first column of each row.
1374 This method is equivalent to calling :meth:`_engine.Connection.execute`
1375 to receive a :class:`_result.Result` object, then invoking the
1376 :meth:`_result.Result.scalars` method to produce a
1377 :class:`_result.ScalarResult` instance.
1379 :return: a :class:`_result.ScalarResult`
1381 .. versionadded:: 1.4.24
1383 """
1385 return self.execute(
1386 statement, parameters, execution_options=execution_options
1387 ).scalars()
1389 @overload
1390 def execute(
1391 self,
1392 statement: TypedReturnsRows[Unpack[_Ts]],
1393 parameters: Optional[_CoreAnyExecuteParams] = None,
1394 *,
1395 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1396 ) -> CursorResult[Unpack[_Ts]]: ...
1398 @overload
1399 def execute(
1400 self,
1401 statement: Executable,
1402 parameters: Optional[_CoreAnyExecuteParams] = None,
1403 *,
1404 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1405 ) -> CursorResult[Unpack[TupleAny]]: ...
1407 def execute(
1408 self,
1409 statement: Executable,
1410 parameters: Optional[_CoreAnyExecuteParams] = None,
1411 *,
1412 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1413 ) -> CursorResult[Unpack[TupleAny]]:
1414 r"""Executes a SQL statement construct and returns a
1415 :class:`_engine.CursorResult`.
1417 :param statement: The statement to be executed. This is always
1418 an object that is in both the :class:`_expression.ClauseElement` and
1419 :class:`_expression.Executable` hierarchies, including:
1421 * :class:`_expression.Select`
1422 * :class:`_expression.Insert`, :class:`_expression.Update`,
1423 :class:`_expression.Delete`
1424 * :class:`_expression.TextClause` and
1425 :class:`_expression.TextualSelect`
1426 * :class:`_schema.DDL` and objects which inherit from
1427 :class:`_schema.ExecutableDDLElement`
1429 :param parameters: parameters which will be bound into the statement.
1430 This may be either a dictionary of parameter names to values,
1431 or a mutable sequence (e.g. a list) of dictionaries. When a
1432 list of dictionaries is passed, the underlying statement execution
1433 will make use of the DBAPI ``cursor.executemany()`` method.
1434 When a single dictionary is passed, the DBAPI ``cursor.execute()``
1435 method will be used.
1437 :param execution_options: optional dictionary of execution options,
1438 which will be associated with the statement execution. This
1439 dictionary can provide a subset of the options that are accepted
1440 by :meth:`_engine.Connection.execution_options`.
1442 :return: a :class:`_engine.Result` object.
1444 """
1445 distilled_parameters = _distill_params_20(parameters)
1446 try:
1447 meth = statement._execute_on_connection
1448 except AttributeError as err:
1449 raise exc.ObjectNotExecutableError(statement) from err
1450 else:
1451 return meth(
1452 self,
1453 distilled_parameters,
1454 execution_options or NO_OPTIONS,
1455 )
1457 def _execute_function(
1458 self,
1459 func: FunctionElement[Any],
1460 distilled_parameters: _CoreMultiExecuteParams,
1461 execution_options: CoreExecuteOptionsParameter,
1462 ) -> CursorResult[Unpack[TupleAny]]:
1463 """Execute a sql.FunctionElement object."""
1465 return self._execute_clauseelement(
1466 func.select(), distilled_parameters, execution_options
1467 )
1469 def _execute_default(
1470 self,
1471 default: DefaultGenerator,
1472 distilled_parameters: _CoreMultiExecuteParams,
1473 execution_options: CoreExecuteOptionsParameter,
1474 ) -> Any:
1475 """Execute a schema.ColumnDefault object."""
1477 exec_opts = self._execution_options.merge_with(execution_options)
1479 event_multiparams: Optional[_CoreMultiExecuteParams]
1480 event_params: Optional[_CoreAnyExecuteParams]
1482 # note for event handlers, the "distilled parameters" which is always
1483 # a list of dicts is broken out into separate "multiparams" and
1484 # "params" collections, which allows the handler to distinguish
1485 # between an executemany and execute style set of parameters.
1486 if self._has_events or self.engine._has_events:
1487 (
1488 default,
1489 distilled_parameters,
1490 event_multiparams,
1491 event_params,
1492 ) = self._invoke_before_exec_event(
1493 default, distilled_parameters, exec_opts
1494 )
1495 else:
1496 event_multiparams = event_params = None
1498 try:
1499 conn = self._dbapi_connection
1500 if conn is None:
1501 conn = self._revalidate_connection()
1503 dialect = self.dialect
1504 ctx = dialect.execution_ctx_cls._init_default(
1505 dialect, self, conn, exec_opts
1506 )
1507 except (exc.PendingRollbackError, exc.ResourceClosedError):
1508 raise
1509 except BaseException as e:
1510 self._handle_dbapi_exception(e, None, None, None, None)
1512 ret = ctx._exec_default(None, default, None)
1514 if self._has_events or self.engine._has_events:
1515 self.dispatch.after_execute(
1516 self,
1517 default,
1518 event_multiparams,
1519 event_params,
1520 exec_opts,
1521 ret,
1522 )
1524 return ret
1526 def _execute_ddl(
1527 self,
1528 ddl: ExecutableDDLElement,
1529 distilled_parameters: _CoreMultiExecuteParams,
1530 execution_options: CoreExecuteOptionsParameter,
1531 ) -> CursorResult[Unpack[TupleAny]]:
1532 """Execute a schema.DDL object."""
1534 exec_opts = ddl._execution_options.merge_with(
1535 self._execution_options, execution_options
1536 )
1538 event_multiparams: Optional[_CoreMultiExecuteParams]
1539 event_params: Optional[_CoreSingleExecuteParams]
1541 if self._has_events or self.engine._has_events:
1542 (
1543 ddl,
1544 distilled_parameters,
1545 event_multiparams,
1546 event_params,
1547 ) = self._invoke_before_exec_event(
1548 ddl, distilled_parameters, exec_opts
1549 )
1550 else:
1551 event_multiparams = event_params = None
1553 schema_translate_map = exec_opts.get("schema_translate_map", None)
1555 dialect = self.dialect
1557 compiled = ddl.compile(
1558 dialect=dialect, schema_translate_map=schema_translate_map
1559 )
1560 ret = self._execute_context(
1561 dialect,
1562 dialect.execution_ctx_cls._init_ddl,
1563 compiled,
1564 None,
1565 exec_opts,
1566 compiled,
1567 )
1568 if self._has_events or self.engine._has_events:
1569 self.dispatch.after_execute(
1570 self,
1571 ddl,
1572 event_multiparams,
1573 event_params,
1574 exec_opts,
1575 ret,
1576 )
1577 return ret
1579 def _invoke_before_exec_event(
1580 self,
1581 elem: Any,
1582 distilled_params: _CoreMultiExecuteParams,
1583 execution_options: _ExecuteOptions,
1584 ) -> Tuple[
1585 Any,
1586 _CoreMultiExecuteParams,
1587 _CoreMultiExecuteParams,
1588 _CoreSingleExecuteParams,
1589 ]:
1590 event_multiparams: _CoreMultiExecuteParams
1591 event_params: _CoreSingleExecuteParams
1593 if len(distilled_params) == 1:
1594 event_multiparams, event_params = [], distilled_params[0]
1595 else:
1596 event_multiparams, event_params = distilled_params, {}
1598 for fn in self.dispatch.before_execute:
1599 elem, event_multiparams, event_params = fn(
1600 self,
1601 elem,
1602 event_multiparams,
1603 event_params,
1604 execution_options,
1605 )
1607 if event_multiparams:
1608 distilled_params = list(event_multiparams)
1609 if event_params:
1610 raise exc.InvalidRequestError(
1611 "Event handler can't return non-empty multiparams "
1612 "and params at the same time"
1613 )
1614 elif event_params:
1615 distilled_params = [event_params]
1616 else:
1617 distilled_params = []
1619 return elem, distilled_params, event_multiparams, event_params
1621 def _execute_clauseelement(
1622 self,
1623 elem: Executable,
1624 distilled_parameters: _CoreMultiExecuteParams,
1625 execution_options: CoreExecuteOptionsParameter,
1626 ) -> CursorResult[Unpack[TupleAny]]:
1627 """Execute a sql.ClauseElement object."""
1629 exec_opts = elem._execution_options.merge_with(
1630 self._execution_options, execution_options
1631 )
1633 has_events = self._has_events or self.engine._has_events
1634 if has_events:
1635 (
1636 elem,
1637 distilled_parameters,
1638 event_multiparams,
1639 event_params,
1640 ) = self._invoke_before_exec_event(
1641 elem, distilled_parameters, exec_opts
1642 )
1644 if distilled_parameters:
1645 # ensure we don't retain a link to the view object for keys()
1646 # which links to the values, which we don't want to cache
1647 keys = sorted(distilled_parameters[0])
1648 for_executemany = len(distilled_parameters) > 1
1649 else:
1650 keys = []
1651 for_executemany = False
1653 dialect = self.dialect
1655 schema_translate_map = exec_opts.get("schema_translate_map", None)
1657 compiled_cache: Optional[CompiledCacheType] = exec_opts.get(
1658 "compiled_cache", self.engine._compiled_cache
1659 )
1661 compiled_sql, extracted_params, param_dict, cache_hit = (
1662 elem._compile_w_cache(
1663 dialect=dialect,
1664 compiled_cache=compiled_cache,
1665 column_keys=keys,
1666 for_executemany=for_executemany,
1667 schema_translate_map=schema_translate_map,
1668 linting=self.dialect.compiler_linting | compiler.WARN_LINTING,
1669 )
1670 )
1671 ret = self._execute_context(
1672 dialect,
1673 dialect.execution_ctx_cls._init_compiled,
1674 compiled_sql,
1675 distilled_parameters,
1676 exec_opts,
1677 compiled_sql,
1678 distilled_parameters,
1679 elem,
1680 extracted_params,
1681 cache_hit=cache_hit,
1682 param_dict=param_dict,
1683 )
1684 if has_events:
1685 self.dispatch.after_execute(
1686 self,
1687 elem,
1688 event_multiparams,
1689 event_params,
1690 exec_opts,
1691 ret,
1692 )
1693 return ret
1695 def exec_driver_sql(
1696 self,
1697 statement: str,
1698 parameters: Optional[_DBAPIAnyExecuteParams] = None,
1699 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1700 ) -> CursorResult[Unpack[TupleAny]]:
1701 r"""Executes a string SQL statement on the DBAPI cursor directly,
1702 without any SQL compilation steps.
1704 This can be used to pass any string directly to the
1705 ``cursor.execute()`` method of the DBAPI in use.
1707 :param statement: The statement str to be executed. Bound parameters
1708 must use the underlying DBAPI's paramstyle, such as "qmark",
1709 "pyformat", "format", etc.
1711 :param parameters: represent bound parameter values to be used in the
1712 execution. The format is one of: a dictionary of named parameters,
1713 a tuple of positional parameters, or a list containing either
1714 dictionaries or tuples for multiple-execute support.
1716 :return: a :class:`_engine.CursorResult`.
1718 E.g. multiple dictionaries::
1721 conn.exec_driver_sql(
1722 "INSERT INTO table (id, value) VALUES (%(id)s, %(value)s)",
1723 [{"id": 1, "value": "v1"}, {"id": 2, "value": "v2"}],
1724 )
1726 Single dictionary::
1728 conn.exec_driver_sql(
1729 "INSERT INTO table (id, value) VALUES (%(id)s, %(value)s)",
1730 dict(id=1, value="v1"),
1731 )
1733 Single tuple::
1735 conn.exec_driver_sql(
1736 "INSERT INTO table (id, value) VALUES (?, ?)", (1, "v1")
1737 )
1739 .. note:: The :meth:`_engine.Connection.exec_driver_sql` method does
1740 not participate in the
1741 :meth:`_events.ConnectionEvents.before_execute` and
1742 :meth:`_events.ConnectionEvents.after_execute` events. To
1743 intercept calls to :meth:`_engine.Connection.exec_driver_sql`, use
1744 :meth:`_events.ConnectionEvents.before_cursor_execute` and
1745 :meth:`_events.ConnectionEvents.after_cursor_execute`.
1747 .. seealso::
1749 :pep:`249`
1751 """
1753 distilled_parameters = _distill_raw_params(parameters)
1755 exec_opts = self._execution_options.merge_with(execution_options)
1757 dialect = self.dialect
1758 ret = self._execute_context(
1759 dialect,
1760 dialect.execution_ctx_cls._init_statement,
1761 statement,
1762 None,
1763 exec_opts,
1764 statement,
1765 distilled_parameters,
1766 )
1768 return ret
1770 def _execute_context(
1771 self,
1772 dialect: Dialect,
1773 constructor: Callable[..., ExecutionContext],
1774 statement: Union[str, Compiled],
1775 parameters: Optional[_AnyMultiExecuteParams],
1776 execution_options: _ExecuteOptions,
1777 *args: Any,
1778 **kw: Any,
1779 ) -> CursorResult[Unpack[TupleAny]]:
1780 """Create an :class:`.ExecutionContext` and execute, returning
1781 a :class:`_engine.CursorResult`."""
1783 if execution_options:
1784 yp = execution_options.get("yield_per", None)
1785 if yp:
1786 execution_options = execution_options.union(
1787 {"stream_results": True, "max_row_buffer": yp}
1788 )
1789 try:
1790 conn = self._dbapi_connection
1791 if conn is None:
1792 conn = self._revalidate_connection()
1794 context = constructor(
1795 dialect, self, conn, execution_options, *args, **kw
1796 )
1797 except (exc.PendingRollbackError, exc.ResourceClosedError):
1798 raise
1799 except BaseException as e:
1800 self._handle_dbapi_exception(
1801 e, str(statement), parameters, None, None
1802 )
1804 if (
1805 self._transaction
1806 and not self._transaction.is_active
1807 or (
1808 self._nested_transaction
1809 and not self._nested_transaction.is_active
1810 )
1811 ):
1812 self._invalid_transaction()
1814 elif self._trans_context_manager:
1815 TransactionalContext._trans_ctx_check(self)
1817 if self._transaction is None:
1818 self._autobegin()
1820 context.pre_exec()
1822 if context.execute_style is ExecuteStyle.INSERTMANYVALUES:
1823 return self._exec_insertmany_context(dialect, context)
1824 else:
1825 return self._exec_single_context(
1826 dialect, context, statement, parameters
1827 )
1829 def _exec_single_context(
1830 self,
1831 dialect: Dialect,
1832 context: ExecutionContext,
1833 statement: Union[str, Compiled],
1834 parameters: Optional[_AnyMultiExecuteParams],
1835 ) -> CursorResult[Unpack[TupleAny]]:
1836 """continue the _execute_context() method for a single DBAPI
1837 cursor.execute() or cursor.executemany() call.
1839 """
1840 if dialect.bind_typing is BindTyping.SETINPUTSIZES:
1841 generic_setinputsizes = context._prepare_set_input_sizes()
1843 if generic_setinputsizes:
1844 try:
1845 dialect.do_set_input_sizes(
1846 context.cursor, generic_setinputsizes, context
1847 )
1848 except BaseException as e:
1849 self._handle_dbapi_exception(
1850 e, str(statement), parameters, None, context
1851 )
1853 cursor, str_statement, parameters = (
1854 context.cursor,
1855 context.statement,
1856 context.parameters,
1857 )
1859 effective_parameters: Optional[_AnyExecuteParams]
1861 if not context.executemany:
1862 effective_parameters = parameters[0]
1863 else:
1864 effective_parameters = parameters
1866 evt_handled: bool = False
1867 try:
1868 if self._has_events or self.engine._has_events:
1869 for fn in self.dispatch.before_cursor_execute:
1870 str_statement, effective_parameters = fn(
1871 self,
1872 cursor,
1873 str_statement,
1874 effective_parameters,
1875 context,
1876 context.executemany,
1877 )
1879 if self._echo:
1880 self._log_info(str_statement)
1882 stats = context._get_cache_stats()
1884 if not self.engine.hide_parameters:
1885 self._log_info(
1886 "[%s] %r",
1887 stats,
1888 sql_util._repr_params(
1889 effective_parameters,
1890 batches=10,
1891 ismulti=context.executemany,
1892 ),
1893 )
1894 else:
1895 self._log_info(
1896 "[%s] [SQL parameters hidden due to "
1897 "hide_parameters=True]",
1898 stats,
1899 )
1900 if context.execute_style is ExecuteStyle.EXECUTEMANY:
1901 effective_parameters = cast(
1902 "_CoreMultiExecuteParams", effective_parameters
1903 )
1904 if self.dialect._has_events:
1905 for fn in self.dialect.dispatch.do_executemany:
1906 if fn(
1907 cursor,
1908 str_statement,
1909 effective_parameters,
1910 context,
1911 ):
1912 evt_handled = True
1913 break
1914 if not evt_handled:
1915 self.dialect.do_executemany(
1916 cursor,
1917 str_statement,
1918 effective_parameters,
1919 context,
1920 )
1921 elif not effective_parameters and context.no_parameters:
1922 if self.dialect._has_events:
1923 for fn in self.dialect.dispatch.do_execute_no_params:
1924 if fn(cursor, str_statement, context):
1925 evt_handled = True
1926 break
1927 if not evt_handled:
1928 self.dialect.do_execute_no_params(
1929 cursor, str_statement, context
1930 )
1931 else:
1932 effective_parameters = cast(
1933 "_CoreSingleExecuteParams", effective_parameters
1934 )
1935 if self.dialect._has_events:
1936 for fn in self.dialect.dispatch.do_execute:
1937 if fn(
1938 cursor,
1939 str_statement,
1940 effective_parameters,
1941 context,
1942 ):
1943 evt_handled = True
1944 break
1945 if not evt_handled:
1946 self.dialect.do_execute(
1947 cursor, str_statement, effective_parameters, context
1948 )
1950 if self._has_events or self.engine._has_events:
1951 self.dispatch.after_cursor_execute(
1952 self,
1953 cursor,
1954 str_statement,
1955 effective_parameters,
1956 context,
1957 context.executemany,
1958 )
1960 context.post_exec()
1962 result = context._setup_result_proxy()
1964 except BaseException as e:
1965 self._handle_dbapi_exception(
1966 e, str_statement, effective_parameters, cursor, context
1967 )
1969 return result
1971 def _exec_insertmany_context(
1972 self,
1973 dialect: Dialect,
1974 context: ExecutionContext,
1975 ) -> CursorResult[Unpack[TupleAny]]:
1976 """continue the _execute_context() method for an "insertmanyvalues"
1977 operation, which will invoke DBAPI
1978 cursor.execute() one or more times with individual log and
1979 event hook calls.
1981 """
1983 if dialect.bind_typing is BindTyping.SETINPUTSIZES:
1984 generic_setinputsizes = context._prepare_set_input_sizes()
1985 else:
1986 generic_setinputsizes = None
1988 cursor, str_statement, parameters = (
1989 context.cursor,
1990 context.statement,
1991 context.parameters,
1992 )
1994 effective_parameters = parameters
1996 engine_events = self._has_events or self.engine._has_events
1997 if self.dialect._has_events:
1998 do_execute_dispatch: Iterable[Any] = (
1999 self.dialect.dispatch.do_execute
2000 )
2001 else:
2002 do_execute_dispatch = ()
2004 if self._echo:
2005 stats = context._get_cache_stats() + " (insertmanyvalues)"
2007 preserve_rowcount = context.execution_options.get(
2008 "preserve_rowcount", False
2009 )
2010 rowcount = 0
2012 for imv_batch in dialect._deliver_insertmanyvalues_batches(
2013 self,
2014 cursor,
2015 str_statement,
2016 effective_parameters,
2017 generic_setinputsizes,
2018 context,
2019 ):
2020 if imv_batch.processed_setinputsizes:
2021 try:
2022 dialect.do_set_input_sizes(
2023 context.cursor,
2024 imv_batch.processed_setinputsizes,
2025 context,
2026 )
2027 except BaseException as e:
2028 self._handle_dbapi_exception(
2029 e,
2030 sql_util._long_statement(imv_batch.replaced_statement),
2031 imv_batch.replaced_parameters,
2032 None,
2033 context,
2034 is_sub_exec=True,
2035 )
2037 sub_stmt = imv_batch.replaced_statement
2038 sub_params = imv_batch.replaced_parameters
2040 try:
2041 if engine_events:
2042 for fn in self.dispatch.before_cursor_execute:
2043 sub_stmt, sub_params = fn(
2044 self,
2045 cursor,
2046 sub_stmt,
2047 sub_params,
2048 context,
2049 True,
2050 )
2052 if self._echo:
2053 self._log_info(sql_util._long_statement(sub_stmt))
2055 imv_stats = f""" {imv_batch.batchnum}/{
2056 imv_batch.total_batches
2057 } ({
2058 'ordered'
2059 if imv_batch.rows_sorted else 'unordered'
2060 }{
2061 '; batch not supported'
2062 if imv_batch.is_downgraded
2063 else ''
2064 })"""
2066 if imv_batch.batchnum == 1:
2067 stats += imv_stats
2068 else:
2069 stats = f"insertmanyvalues{imv_stats}"
2071 if not self.engine.hide_parameters:
2072 self._log_info(
2073 "[%s] %r",
2074 stats,
2075 sql_util._repr_params(
2076 sub_params,
2077 batches=10,
2078 ismulti=False,
2079 ),
2080 )
2081 else:
2082 self._log_info(
2083 "[%s] [SQL parameters hidden due to "
2084 "hide_parameters=True]",
2085 stats,
2086 )
2088 for fn in do_execute_dispatch:
2089 if fn(
2090 cursor,
2091 sub_stmt,
2092 sub_params,
2093 context,
2094 ):
2095 break
2096 else:
2097 dialect.do_execute(
2098 cursor,
2099 sub_stmt,
2100 sub_params,
2101 context,
2102 )
2104 if engine_events:
2105 self.dispatch.after_cursor_execute(
2106 self,
2107 cursor,
2108 sub_stmt,
2109 sub_params,
2110 context,
2111 context.executemany,
2112 )
2113 except BaseException as e:
2114 self._handle_dbapi_exception(
2115 e,
2116 sql_util._long_statement(sub_stmt),
2117 sub_params,
2118 cursor,
2119 context,
2120 is_sub_exec=True,
2121 )
2123 if preserve_rowcount:
2124 rowcount += imv_batch.current_batch_size
2126 try:
2127 context.post_exec()
2129 if preserve_rowcount:
2130 context._rowcount = rowcount # type: ignore[attr-defined]
2132 result = context._setup_result_proxy()
2134 except BaseException as e:
2135 self._handle_dbapi_exception(
2136 e, str_statement, effective_parameters, cursor, context
2137 )
2139 return result
2141 def _cursor_execute(
2142 self,
2143 cursor: DBAPICursor,
2144 statement: str,
2145 parameters: _DBAPISingleExecuteParams,
2146 context: Optional[ExecutionContext] = None,
2147 ) -> None:
2148 """Execute a statement + params on the given cursor.
2150 Adds appropriate logging and exception handling.
2152 This method is used by DefaultDialect for special-case
2153 executions, such as for sequences and column defaults.
2154 The path of statement execution in the majority of cases
2155 terminates at _execute_context().
2157 """
2158 try:
2159 if self._has_events or self.engine._has_events:
2160 for fn in self.dispatch.before_cursor_execute:
2161 statement, parameters = fn(
2162 self, cursor, statement, parameters, context, False
2163 )
2165 if self._echo:
2166 self._log_info(statement)
2167 self._log_info("[raw sql] %r", parameters)
2169 for fn in (
2170 ()
2171 if not self.dialect._has_events
2172 else self.dialect.dispatch.do_execute
2173 ):
2174 if fn(cursor, statement, parameters, context):
2175 break
2176 else:
2177 self.dialect.do_execute(cursor, statement, parameters, context)
2179 if self._has_events or self.engine._has_events:
2180 self.dispatch.after_cursor_execute(
2181 self, cursor, statement, parameters, context, False
2182 )
2183 except BaseException as e:
2184 self._handle_dbapi_exception(
2185 e, statement, parameters, cursor, context
2186 )
2188 def _safe_close_cursor(self, cursor: DBAPICursor) -> None:
2189 """Close the given cursor, catching exceptions
2190 and turning into log warnings.
2192 """
2193 try:
2194 cursor.close()
2195 except Exception:
2196 # log the error through the connection pool's logger.
2197 self.engine.pool.logger.error(
2198 "Error closing cursor", exc_info=True
2199 )
2201 _reentrant_error = False
2202 _is_disconnect = False
2204 def _handle_dbapi_exception(
2205 self,
2206 e: BaseException,
2207 statement: Optional[str],
2208 parameters: Optional[_AnyExecuteParams],
2209 cursor: Optional[DBAPICursor],
2210 context: Optional[ExecutionContext],
2211 is_sub_exec: bool = False,
2212 ) -> NoReturn:
2213 exc_info = sys.exc_info()
2215 is_exit_exception = util.is_exit_exception(e)
2217 if not self._is_disconnect:
2218 self._is_disconnect = (
2219 isinstance(e, self.dialect.loaded_dbapi.Error)
2220 and not self.closed
2221 and self.dialect.is_disconnect(
2222 e,
2223 self._dbapi_connection if not self.invalidated else None,
2224 cursor,
2225 )
2226 ) or (is_exit_exception and not self.closed)
2228 invalidate_pool_on_disconnect = not is_exit_exception
2230 ismulti: bool = (
2231 not is_sub_exec and context.executemany
2232 if context is not None
2233 else False
2234 )
2235 if self._reentrant_error:
2236 raise exc.DBAPIError.instance(
2237 statement,
2238 parameters,
2239 e,
2240 self.dialect.loaded_dbapi.Error,
2241 hide_parameters=self.engine.hide_parameters,
2242 dialect=self.dialect,
2243 ismulti=ismulti,
2244 ).with_traceback(exc_info[2]) from e
2245 self._reentrant_error = True
2246 try:
2247 # non-DBAPI error - if we already got a context,
2248 # or there's no string statement, don't wrap it
2249 should_wrap = isinstance(e, self.dialect.loaded_dbapi.Error) or (
2250 not isinstance(e, exc.StatementError)
2251 and statement is not None
2252 and context is None
2253 and not is_exit_exception
2254 )
2256 if should_wrap:
2257 sqlalchemy_exception = exc.DBAPIError.instance(
2258 statement,
2259 parameters,
2260 cast(Exception, e),
2261 self.dialect.loaded_dbapi.Error,
2262 hide_parameters=self.engine.hide_parameters,
2263 connection_invalidated=self._is_disconnect,
2264 dialect=self.dialect,
2265 ismulti=ismulti,
2266 )
2267 else:
2268 sqlalchemy_exception = None
2270 newraise = None
2272 if (self.dialect._has_events) and not self._execution_options.get(
2273 "skip_user_error_events", False
2274 ):
2275 ctx = ExceptionContextImpl(
2276 e,
2277 sqlalchemy_exception,
2278 self.engine,
2279 self.dialect,
2280 self,
2281 cursor,
2282 statement,
2283 parameters,
2284 context,
2285 self._is_disconnect,
2286 invalidate_pool_on_disconnect,
2287 False,
2288 )
2290 for fn in self.dialect.dispatch.handle_error:
2291 try:
2292 # handler returns an exception;
2293 # call next handler in a chain
2294 per_fn = fn(ctx)
2295 if per_fn is not None:
2296 ctx.chained_exception = newraise = per_fn
2297 except Exception as _raised:
2298 # handler raises an exception - stop processing
2299 newraise = _raised
2300 break
2302 if self._is_disconnect != ctx.is_disconnect:
2303 self._is_disconnect = ctx.is_disconnect
2304 if sqlalchemy_exception:
2305 sqlalchemy_exception.connection_invalidated = (
2306 ctx.is_disconnect
2307 )
2309 # set up potentially user-defined value for
2310 # invalidate pool.
2311 invalidate_pool_on_disconnect = (
2312 ctx.invalidate_pool_on_disconnect
2313 )
2315 if should_wrap and context:
2316 context.handle_dbapi_exception(e)
2318 if not self._is_disconnect:
2319 if cursor:
2320 self._safe_close_cursor(cursor)
2321 # "autorollback" was mostly relevant in 1.x series.
2322 # It's very unlikely to reach here, as the connection
2323 # does autobegin so when we are here, we are usually
2324 # in an explicit / semi-explicit transaction.
2325 # however we have a test which manufactures this
2326 # scenario in any case using an event handler.
2327 # test/engine/test_execute.py-> test_actual_autorollback
2328 if not self.in_transaction():
2329 self._rollback_impl()
2331 if newraise:
2332 raise newraise.with_traceback(exc_info[2]) from e
2333 elif should_wrap:
2334 assert sqlalchemy_exception is not None
2335 raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
2336 else:
2337 assert exc_info[1] is not None
2338 raise exc_info[1].with_traceback(exc_info[2])
2339 finally:
2340 del self._reentrant_error
2341 if self._is_disconnect:
2342 del self._is_disconnect
2343 if not self.invalidated:
2344 dbapi_conn_wrapper = self._dbapi_connection
2345 assert dbapi_conn_wrapper is not None
2346 if invalidate_pool_on_disconnect:
2347 self.engine.pool._invalidate(dbapi_conn_wrapper, e)
2348 self.invalidate(e)
2350 @classmethod
2351 def _handle_dbapi_exception_noconnection(
2352 cls,
2353 e: BaseException,
2354 dialect: Dialect,
2355 engine: Optional[Engine] = None,
2356 is_disconnect: Optional[bool] = None,
2357 invalidate_pool_on_disconnect: bool = True,
2358 is_pre_ping: bool = False,
2359 ) -> NoReturn:
2360 exc_info = sys.exc_info()
2362 if is_disconnect is None:
2363 is_disconnect = isinstance(
2364 e, dialect.loaded_dbapi.Error
2365 ) and dialect.is_disconnect(e, None, None)
2367 should_wrap = isinstance(e, dialect.loaded_dbapi.Error)
2369 if should_wrap:
2370 sqlalchemy_exception = exc.DBAPIError.instance(
2371 None,
2372 None,
2373 cast(Exception, e),
2374 dialect.loaded_dbapi.Error,
2375 hide_parameters=(
2376 engine.hide_parameters if engine is not None else False
2377 ),
2378 connection_invalidated=is_disconnect,
2379 dialect=dialect,
2380 )
2381 else:
2382 sqlalchemy_exception = None
2384 newraise = None
2386 if dialect._has_events:
2387 ctx = ExceptionContextImpl(
2388 e,
2389 sqlalchemy_exception,
2390 engine,
2391 dialect,
2392 None,
2393 None,
2394 None,
2395 None,
2396 None,
2397 is_disconnect,
2398 invalidate_pool_on_disconnect,
2399 is_pre_ping,
2400 )
2401 for fn in dialect.dispatch.handle_error:
2402 try:
2403 # handler returns an exception;
2404 # call next handler in a chain
2405 per_fn = fn(ctx)
2406 if per_fn is not None:
2407 ctx.chained_exception = newraise = per_fn
2408 except Exception as _raised:
2409 # handler raises an exception - stop processing
2410 newraise = _raised
2411 break
2413 if sqlalchemy_exception and is_disconnect != ctx.is_disconnect:
2414 sqlalchemy_exception.connection_invalidated = ctx.is_disconnect
2416 if newraise:
2417 raise newraise.with_traceback(exc_info[2]) from e
2418 elif should_wrap:
2419 assert sqlalchemy_exception is not None
2420 raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
2421 else:
2422 assert exc_info[1] is not None
2423 raise exc_info[1].with_traceback(exc_info[2])
2425 def _run_ddl_visitor(
2426 self,
2427 visitorcallable: Type[InvokeDDLBase],
2428 element: SchemaVisitable,
2429 **kwargs: Any,
2430 ) -> None:
2431 """run a DDL visitor.
2433 This method is only here so that the MockConnection can change the
2434 options given to the visitor so that "checkfirst" is skipped.
2436 """
2437 visitorcallable(
2438 dialect=self.dialect, connection=self, **kwargs
2439 ).traverse_single(element)
2442class ExceptionContextImpl(ExceptionContext):
2443 """Implement the :class:`.ExceptionContext` interface."""
2445 __slots__ = (
2446 "connection",
2447 "engine",
2448 "dialect",
2449 "cursor",
2450 "statement",
2451 "parameters",
2452 "original_exception",
2453 "sqlalchemy_exception",
2454 "chained_exception",
2455 "execution_context",
2456 "is_disconnect",
2457 "invalidate_pool_on_disconnect",
2458 "is_pre_ping",
2459 )
2461 def __init__(
2462 self,
2463 exception: BaseException,
2464 sqlalchemy_exception: Optional[exc.StatementError],
2465 engine: Optional[Engine],
2466 dialect: Dialect,
2467 connection: Optional[Connection],
2468 cursor: Optional[DBAPICursor],
2469 statement: Optional[str],
2470 parameters: Optional[_DBAPIAnyExecuteParams],
2471 context: Optional[ExecutionContext],
2472 is_disconnect: bool,
2473 invalidate_pool_on_disconnect: bool,
2474 is_pre_ping: bool,
2475 ):
2476 self.engine = engine
2477 self.dialect = dialect
2478 self.connection = connection
2479 self.sqlalchemy_exception = sqlalchemy_exception
2480 self.original_exception = exception
2481 self.execution_context = context
2482 self.statement = statement
2483 self.parameters = parameters
2484 self.is_disconnect = is_disconnect
2485 self.invalidate_pool_on_disconnect = invalidate_pool_on_disconnect
2486 self.is_pre_ping = is_pre_ping
2489class Transaction(TransactionalContext):
2490 """Represent a database transaction in progress.
2492 The :class:`.Transaction` object is procured by
2493 calling the :meth:`_engine.Connection.begin` method of
2494 :class:`_engine.Connection`::
2496 from sqlalchemy import create_engine
2498 engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
2499 connection = engine.connect()
2500 trans = connection.begin()
2501 connection.execute(text("insert into x (a, b) values (1, 2)"))
2502 trans.commit()
2504 The object provides :meth:`.rollback` and :meth:`.commit`
2505 methods in order to control transaction boundaries. It
2506 also implements a context manager interface so that
2507 the Python ``with`` statement can be used with the
2508 :meth:`_engine.Connection.begin` method::
2510 with connection.begin():
2511 connection.execute(text("insert into x (a, b) values (1, 2)"))
2513 The Transaction object is **not** threadsafe.
2515 .. seealso::
2517 :meth:`_engine.Connection.begin`
2519 :meth:`_engine.Connection.begin_twophase`
2521 :meth:`_engine.Connection.begin_nested`
2523 .. index::
2524 single: thread safety; Transaction
2525 """ # noqa
2527 __slots__ = ()
2529 _is_root: bool = False
2530 is_active: bool
2531 connection: Connection
2533 def __init__(self, connection: Connection):
2534 raise NotImplementedError()
2536 @property
2537 def _deactivated_from_connection(self) -> bool:
2538 """True if this transaction is totally deactivated from the connection
2539 and therefore can no longer affect its state.
2541 """
2542 raise NotImplementedError()
2544 def _do_close(self) -> None:
2545 raise NotImplementedError()
2547 def _do_rollback(self) -> None:
2548 raise NotImplementedError()
2550 def _do_commit(self) -> None:
2551 raise NotImplementedError()
2553 @property
2554 def is_valid(self) -> bool:
2555 return self.is_active and not self.connection.invalidated
2557 def close(self) -> None:
2558 """Close this :class:`.Transaction`.
2560 If this transaction is the base transaction in a begin/commit
2561 nesting, the transaction will rollback(). Otherwise, the
2562 method returns.
2564 This is used to cancel a Transaction without affecting the scope of
2565 an enclosing transaction.
2567 """
2568 try:
2569 self._do_close()
2570 finally:
2571 assert not self.is_active
2573 def rollback(self) -> None:
2574 """Roll back this :class:`.Transaction`.
2576 The implementation of this may vary based on the type of transaction in
2577 use:
2579 * For a simple database transaction (e.g. :class:`.RootTransaction`),
2580 it corresponds to a ROLLBACK.
2582 * For a :class:`.NestedTransaction`, it corresponds to a
2583 "ROLLBACK TO SAVEPOINT" operation.
2585 * For a :class:`.TwoPhaseTransaction`, DBAPI-specific methods for two
2586 phase transactions may be used.
2589 """
2590 try:
2591 self._do_rollback()
2592 finally:
2593 assert not self.is_active
2595 def commit(self) -> None:
2596 """Commit this :class:`.Transaction`.
2598 The implementation of this may vary based on the type of transaction in
2599 use:
2601 * For a simple database transaction (e.g. :class:`.RootTransaction`),
2602 it corresponds to a COMMIT.
2604 * For a :class:`.NestedTransaction`, it corresponds to a
2605 "RELEASE SAVEPOINT" operation.
2607 * For a :class:`.TwoPhaseTransaction`, DBAPI-specific methods for two
2608 phase transactions may be used.
2610 """
2611 try:
2612 self._do_commit()
2613 finally:
2614 assert not self.is_active
2616 def _get_subject(self) -> Connection:
2617 return self.connection
2619 def _transaction_is_active(self) -> bool:
2620 return self.is_active
2622 def _transaction_is_closed(self) -> bool:
2623 return not self._deactivated_from_connection
2625 def _rollback_can_be_called(self) -> bool:
2626 # for RootTransaction / NestedTransaction, it's safe to call
2627 # rollback() even if the transaction is deactive and no warnings
2628 # will be emitted. tested in
2629 # test_transaction.py -> test_no_rollback_in_deactive(?:_savepoint)?
2630 return True
2633class RootTransaction(Transaction):
2634 """Represent the "root" transaction on a :class:`_engine.Connection`.
2636 This corresponds to the current "BEGIN/COMMIT/ROLLBACK" that's occurring
2637 for the :class:`_engine.Connection`. The :class:`_engine.RootTransaction`
2638 is created by calling upon the :meth:`_engine.Connection.begin` method, and
2639 remains associated with the :class:`_engine.Connection` throughout its
2640 active span. The current :class:`_engine.RootTransaction` in use is
2641 accessible via the :attr:`_engine.Connection.get_transaction` method of
2642 :class:`_engine.Connection`.
2644 In :term:`2.0 style` use, the :class:`_engine.Connection` also employs
2645 "autobegin" behavior that will create a new
2646 :class:`_engine.RootTransaction` whenever a connection in a
2647 non-transactional state is used to emit commands on the DBAPI connection.
2648 The scope of the :class:`_engine.RootTransaction` in 2.0 style
2649 use can be controlled using the :meth:`_engine.Connection.commit` and
2650 :meth:`_engine.Connection.rollback` methods.
2653 """
2655 _is_root = True
2657 __slots__ = ("connection", "is_active")
2659 def __init__(self, connection: Connection):
2660 assert connection._transaction is None
2661 if connection._trans_context_manager:
2662 TransactionalContext._trans_ctx_check(connection)
2663 self.connection = connection
2664 self._connection_begin_impl()
2665 connection._transaction = self
2667 self.is_active = True
2669 def _deactivate_from_connection(self) -> None:
2670 if self.is_active:
2671 assert self.connection._transaction is self
2672 self.is_active = False
2674 elif self.connection._transaction is not self:
2675 util.warn("transaction already deassociated from connection")
2677 @property
2678 def _deactivated_from_connection(self) -> bool:
2679 return self.connection._transaction is not self
2681 def _connection_begin_impl(self) -> None:
2682 self.connection._begin_impl(self)
2684 def _connection_rollback_impl(self) -> None:
2685 self.connection._rollback_impl()
2687 def _connection_commit_impl(self) -> None:
2688 self.connection._commit_impl()
2690 def _close_impl(self, try_deactivate: bool = False) -> None:
2691 try:
2692 if self.is_active:
2693 self._connection_rollback_impl()
2695 if self.connection._nested_transaction:
2696 self.connection._nested_transaction._cancel()
2697 finally:
2698 if self.is_active or try_deactivate:
2699 self._deactivate_from_connection()
2700 if self.connection._transaction is self:
2701 self.connection._transaction = None
2703 assert not self.is_active
2704 assert self.connection._transaction is not self
2706 def _do_close(self) -> None:
2707 self._close_impl()
2709 def _do_rollback(self) -> None:
2710 self._close_impl(try_deactivate=True)
2712 def _do_commit(self) -> None:
2713 if self.is_active:
2714 assert self.connection._transaction is self
2716 try:
2717 self._connection_commit_impl()
2718 finally:
2719 # whether or not commit succeeds, cancel any
2720 # nested transactions, make this transaction "inactive"
2721 # and remove it as a reset agent
2722 if self.connection._nested_transaction:
2723 self.connection._nested_transaction._cancel()
2725 self._deactivate_from_connection()
2727 # ...however only remove as the connection's current transaction
2728 # if commit succeeded. otherwise it stays on so that a rollback
2729 # needs to occur.
2730 self.connection._transaction = None
2731 else:
2732 if self.connection._transaction is self:
2733 self.connection._invalid_transaction()
2734 else:
2735 raise exc.InvalidRequestError("This transaction is inactive")
2737 assert not self.is_active
2738 assert self.connection._transaction is not self
2741class NestedTransaction(Transaction):
2742 """Represent a 'nested', or SAVEPOINT transaction.
2744 The :class:`.NestedTransaction` object is created by calling the
2745 :meth:`_engine.Connection.begin_nested` method of
2746 :class:`_engine.Connection`.
2748 When using :class:`.NestedTransaction`, the semantics of "begin" /
2749 "commit" / "rollback" are as follows:
2751 * the "begin" operation corresponds to the "BEGIN SAVEPOINT" command, where
2752 the savepoint is given an explicit name that is part of the state
2753 of this object.
2755 * The :meth:`.NestedTransaction.commit` method corresponds to a
2756 "RELEASE SAVEPOINT" operation, using the savepoint identifier associated
2757 with this :class:`.NestedTransaction`.
2759 * The :meth:`.NestedTransaction.rollback` method corresponds to a
2760 "ROLLBACK TO SAVEPOINT" operation, using the savepoint identifier
2761 associated with this :class:`.NestedTransaction`.
2763 The rationale for mimicking the semantics of an outer transaction in
2764 terms of savepoints so that code may deal with a "savepoint" transaction
2765 and an "outer" transaction in an agnostic way.
2767 .. seealso::
2769 :ref:`session_begin_nested` - ORM version of the SAVEPOINT API.
2771 """
2773 __slots__ = ("connection", "is_active", "_savepoint", "_previous_nested")
2775 _savepoint: str
2777 def __init__(self, connection: Connection):
2778 assert connection._transaction is not None
2779 if connection._trans_context_manager:
2780 TransactionalContext._trans_ctx_check(connection)
2781 self.connection = connection
2782 self._savepoint = self.connection._savepoint_impl()
2783 self.is_active = True
2784 self._previous_nested = connection._nested_transaction
2785 connection._nested_transaction = self
2787 def _deactivate_from_connection(self, warn: bool = True) -> None:
2788 if self.connection._nested_transaction is self:
2789 self.connection._nested_transaction = self._previous_nested
2790 elif warn:
2791 util.warn(
2792 "nested transaction already deassociated from connection"
2793 )
2795 @property
2796 def _deactivated_from_connection(self) -> bool:
2797 return self.connection._nested_transaction is not self
2799 def _cancel(self) -> None:
2800 # called by RootTransaction when the outer transaction is
2801 # committed, rolled back, or closed to cancel all savepoints
2802 # without any action being taken
2803 self.is_active = False
2804 self._deactivate_from_connection()
2805 if self._previous_nested:
2806 self._previous_nested._cancel()
2808 def _close_impl(
2809 self, deactivate_from_connection: bool, warn_already_deactive: bool
2810 ) -> None:
2811 try:
2812 if (
2813 self.is_active
2814 and self.connection._transaction
2815 and self.connection._transaction.is_active
2816 ):
2817 self.connection._rollback_to_savepoint_impl(self._savepoint)
2818 finally:
2819 self.is_active = False
2821 if deactivate_from_connection:
2822 self._deactivate_from_connection(warn=warn_already_deactive)
2824 assert not self.is_active
2825 if deactivate_from_connection:
2826 assert self.connection._nested_transaction is not self
2828 def _do_close(self) -> None:
2829 self._close_impl(True, False)
2831 def _do_rollback(self) -> None:
2832 self._close_impl(True, True)
2834 def _do_commit(self) -> None:
2835 if self.is_active:
2836 try:
2837 self.connection._release_savepoint_impl(self._savepoint)
2838 finally:
2839 # nested trans becomes inactive on failed release
2840 # unconditionally. this prevents it from trying to
2841 # emit SQL when it rolls back.
2842 self.is_active = False
2844 # but only de-associate from connection if it succeeded
2845 self._deactivate_from_connection()
2846 else:
2847 if self.connection._nested_transaction is self:
2848 self.connection._invalid_transaction()
2849 else:
2850 raise exc.InvalidRequestError(
2851 "This nested transaction is inactive"
2852 )
2855class TwoPhaseTransaction(RootTransaction):
2856 """Represent a two-phase transaction.
2858 A new :class:`.TwoPhaseTransaction` object may be procured
2859 using the :meth:`_engine.Connection.begin_twophase` method.
2861 The interface is the same as that of :class:`.Transaction`
2862 with the addition of the :meth:`prepare` method.
2864 """
2866 __slots__ = ("xid", "_is_prepared")
2868 xid: Any
2870 def __init__(self, connection: Connection, xid: Any):
2871 self._is_prepared = False
2872 self.xid = xid
2873 super().__init__(connection)
2875 def prepare(self) -> None:
2876 """Prepare this :class:`.TwoPhaseTransaction`.
2878 After a PREPARE, the transaction can be committed.
2880 """
2881 if not self.is_active:
2882 raise exc.InvalidRequestError("This transaction is inactive")
2883 self.connection._prepare_twophase_impl(self.xid)
2884 self._is_prepared = True
2886 def _connection_begin_impl(self) -> None:
2887 self.connection._begin_twophase_impl(self)
2889 def _connection_rollback_impl(self) -> None:
2890 self.connection._rollback_twophase_impl(self.xid, self._is_prepared)
2892 def _connection_commit_impl(self) -> None:
2893 self.connection._commit_twophase_impl(self.xid, self._is_prepared)
2896class Engine(
2897 ConnectionEventsTarget, log.Identified, inspection.Inspectable["Inspector"]
2898):
2899 """
2900 Connects a :class:`~sqlalchemy.pool.Pool` and
2901 :class:`~sqlalchemy.engine.interfaces.Dialect` together to provide a
2902 source of database connectivity and behavior.
2904 An :class:`_engine.Engine` object is instantiated publicly using the
2905 :func:`~sqlalchemy.create_engine` function.
2907 .. seealso::
2909 :doc:`/core/engines`
2911 :ref:`connections_toplevel`
2913 """
2915 dispatch: dispatcher[ConnectionEventsTarget]
2917 _compiled_cache: Optional[CompiledCacheType]
2919 _execution_options: _ExecuteOptions = _EMPTY_EXECUTION_OPTS
2920 _has_events: bool = False
2921 _connection_cls: Type[Connection] = Connection
2922 _sqla_logger_namespace: str = "sqlalchemy.engine.Engine"
2923 _is_future: bool = False
2925 _schema_translate_map: Optional[SchemaTranslateMapType] = None
2926 _option_cls: Type[OptionEngine]
2928 dialect: Dialect
2929 pool: Pool
2930 url: URL
2931 hide_parameters: bool
2933 def __init__(
2934 self,
2935 pool: Pool,
2936 dialect: Dialect,
2937 url: URL,
2938 logging_name: Optional[str] = None,
2939 echo: Optional[_EchoFlagType] = None,
2940 query_cache_size: int = 500,
2941 execution_options: Optional[Mapping[str, Any]] = None,
2942 hide_parameters: bool = False,
2943 ):
2944 self.pool = pool
2945 self.url = url
2946 self.dialect = dialect
2947 if logging_name:
2948 self.logging_name = logging_name
2949 self.echo = echo
2950 self.hide_parameters = hide_parameters
2951 if query_cache_size != 0:
2952 self._compiled_cache = util.LRUCache(
2953 query_cache_size, size_alert=self._lru_size_alert
2954 )
2955 else:
2956 self._compiled_cache = None
2957 log.instance_logger(self, echoflag=echo)
2958 if execution_options:
2959 self.update_execution_options(**execution_options)
2961 def _lru_size_alert(self, cache: util.LRUCache[Any, Any]) -> None:
2962 if self._should_log_info():
2963 self.logger.info(
2964 "Compiled cache size pruning from %d items to %d. "
2965 "Increase cache size to reduce the frequency of pruning.",
2966 len(cache),
2967 cache.capacity,
2968 )
2970 @property
2971 def engine(self) -> Engine:
2972 """Returns this :class:`.Engine`.
2974 Used for legacy schemes that accept :class:`.Connection` /
2975 :class:`.Engine` objects within the same variable.
2977 """
2978 return self
2980 def clear_compiled_cache(self) -> None:
2981 """Clear the compiled cache associated with the dialect.
2983 This applies **only** to the built-in cache that is established
2984 via the :paramref:`_engine.create_engine.query_cache_size` parameter.
2985 It will not impact any dictionary caches that were passed via the
2986 :paramref:`.Connection.execution_options.compiled_cache` parameter.
2988 .. versionadded:: 1.4
2990 """
2991 if self._compiled_cache:
2992 self._compiled_cache.clear()
2994 def update_execution_options(self, **opt: Any) -> None:
2995 r"""Update the default execution_options dictionary
2996 of this :class:`_engine.Engine`.
2998 The given keys/values in \**opt are added to the
2999 default execution options that will be used for
3000 all connections. The initial contents of this dictionary
3001 can be sent via the ``execution_options`` parameter
3002 to :func:`_sa.create_engine`.
3004 .. seealso::
3006 :meth:`_engine.Connection.execution_options`
3008 :meth:`_engine.Engine.execution_options`
3010 """
3011 self.dispatch.set_engine_execution_options(self, opt)
3012 self._execution_options = self._execution_options.union(opt)
3013 self.dialect.set_engine_execution_options(self, opt)
3015 @overload
3016 def execution_options(
3017 self,
3018 *,
3019 compiled_cache: Optional[CompiledCacheType] = ...,
3020 logging_token: str = ...,
3021 isolation_level: IsolationLevel = ...,
3022 insertmanyvalues_page_size: int = ...,
3023 schema_translate_map: Optional[SchemaTranslateMapType] = ...,
3024 **opt: Any,
3025 ) -> OptionEngine: ...
3027 @overload
3028 def execution_options(self, **opt: Any) -> OptionEngine: ...
3030 def execution_options(self, **opt: Any) -> OptionEngine:
3031 """Return a new :class:`_engine.Engine` that will provide
3032 :class:`_engine.Connection` objects with the given execution options.
3034 The returned :class:`_engine.Engine` remains related to the original
3035 :class:`_engine.Engine` in that it shares the same connection pool and
3036 other state:
3038 * The :class:`_pool.Pool` used by the new :class:`_engine.Engine`
3039 is the
3040 same instance. The :meth:`_engine.Engine.dispose`
3041 method will replace
3042 the connection pool instance for the parent engine as well
3043 as this one.
3044 * Event listeners are "cascaded" - meaning, the new
3045 :class:`_engine.Engine`
3046 inherits the events of the parent, and new events can be associated
3047 with the new :class:`_engine.Engine` individually.
3048 * The logging configuration and logging_name is copied from the parent
3049 :class:`_engine.Engine`.
3051 The intent of the :meth:`_engine.Engine.execution_options` method is
3052 to implement schemes where multiple :class:`_engine.Engine`
3053 objects refer to the same connection pool, but are differentiated
3054 by options that affect some execution-level behavior for each
3055 engine. One such example is breaking into separate "reader" and
3056 "writer" :class:`_engine.Engine` instances, where one
3057 :class:`_engine.Engine`
3058 has a lower :term:`isolation level` setting configured or is even
3059 transaction-disabled using "autocommit". An example of this
3060 configuration is at :ref:`dbapi_autocommit_multiple`.
3062 Another example is one that
3063 uses a custom option ``shard_id`` which is consumed by an event
3064 to change the current schema on a database connection::
3066 from sqlalchemy import event
3067 from sqlalchemy.engine import Engine
3069 primary_engine = create_engine("mysql+mysqldb://")
3070 shard1 = primary_engine.execution_options(shard_id="shard1")
3071 shard2 = primary_engine.execution_options(shard_id="shard2")
3073 shards = {"default": "base", "shard_1": "db1", "shard_2": "db2"}
3076 @event.listens_for(Engine, "before_cursor_execute")
3077 def _switch_shard(conn, cursor, stmt, params, context, executemany):
3078 shard_id = conn.get_execution_options().get("shard_id", "default")
3079 current_shard = conn.info.get("current_shard", None)
3081 if current_shard != shard_id:
3082 cursor.execute("use %s" % shards[shard_id])
3083 conn.info["current_shard"] = shard_id
3085 The above recipe illustrates two :class:`_engine.Engine` objects that
3086 will each serve as factories for :class:`_engine.Connection` objects
3087 that have pre-established "shard_id" execution options present. A
3088 :meth:`_events.ConnectionEvents.before_cursor_execute` event handler
3089 then interprets this execution option to emit a MySQL ``use`` statement
3090 to switch databases before a statement execution, while at the same
3091 time keeping track of which database we've established using the
3092 :attr:`_engine.Connection.info` dictionary.
3094 .. seealso::
3096 :meth:`_engine.Connection.execution_options`
3097 - update execution options
3098 on a :class:`_engine.Connection` object.
3100 :meth:`_engine.Engine.update_execution_options`
3101 - update the execution
3102 options for a given :class:`_engine.Engine` in place.
3104 :meth:`_engine.Engine.get_execution_options`
3107 """ # noqa: E501
3108 return self._option_cls(self, opt)
3110 def get_execution_options(self) -> _ExecuteOptions:
3111 """Get the non-SQL options which will take effect during execution.
3113 .. seealso::
3115 :meth:`_engine.Engine.execution_options`
3116 """
3117 return self._execution_options
3119 @property
3120 def name(self) -> str:
3121 """String name of the :class:`~sqlalchemy.engine.interfaces.Dialect`
3122 in use by this :class:`Engine`.
3124 """
3126 return self.dialect.name
3128 @property
3129 def driver(self) -> str:
3130 """Driver name of the :class:`~sqlalchemy.engine.interfaces.Dialect`
3131 in use by this :class:`Engine`.
3133 """
3135 return self.dialect.driver
3137 echo = log.echo_property()
3139 def __repr__(self) -> str:
3140 return "Engine(%r)" % (self.url,)
3142 def dispose(self, close: bool = True) -> None:
3143 """Dispose of the connection pool used by this
3144 :class:`_engine.Engine`.
3146 A new connection pool is created immediately after the old one has been
3147 disposed. The previous connection pool is disposed either actively, by
3148 closing out all currently checked-in connections in that pool, or
3149 passively, by losing references to it but otherwise not closing any
3150 connections. The latter strategy is more appropriate for an initializer
3151 in a forked Python process.
3153 Event listeners associated with the old pool via :class:`.PoolEvents`
3154 are **transferred to the new pool**; this is to support the pattern
3155 by which :class:`.PoolEvents` are set up in terms of the owning
3156 :class:`.Engine` without the need to refer to the :class:`.Pool`
3157 directly.
3159 :param close: if left at its default of ``True``, has the
3160 effect of fully closing all **currently checked in**
3161 database connections. Connections that are still checked out
3162 will **not** be closed, however they will no longer be associated
3163 with this :class:`_engine.Engine`,
3164 so when they are closed individually, eventually the
3165 :class:`_pool.Pool` which they are associated with will
3166 be garbage collected and they will be closed out fully, if
3167 not already closed on checkin.
3169 If set to ``False``, the previous connection pool is de-referenced,
3170 and otherwise not touched in any way.
3172 .. versionadded:: 1.4.33 Added the :paramref:`.Engine.dispose.close`
3173 parameter to allow the replacement of a connection pool in a child
3174 process without interfering with the connections used by the parent
3175 process.
3178 .. seealso::
3180 :ref:`engine_disposal`
3182 :ref:`pooling_multiprocessing`
3184 :meth:`.ConnectionEvents.engine_disposed`
3186 """
3187 if close:
3188 self.pool.dispose()
3189 self.pool = self.pool.recreate()
3190 self.dispatch.engine_disposed(self)
3192 @contextlib.contextmanager
3193 def _optional_conn_ctx_manager(
3194 self, connection: Optional[Connection] = None
3195 ) -> Iterator[Connection]:
3196 if connection is None:
3197 with self.connect() as conn:
3198 yield conn
3199 else:
3200 yield connection
3202 @contextlib.contextmanager
3203 def begin(self) -> Iterator[Connection]:
3204 """Return a context manager delivering a :class:`_engine.Connection`
3205 with a :class:`.Transaction` established.
3207 E.g.::
3209 with engine.begin() as conn:
3210 conn.execute(text("insert into table (x, y, z) values (1, 2, 3)"))
3211 conn.execute(text("my_special_procedure(5)"))
3213 Upon successful operation, the :class:`.Transaction`
3214 is committed. If an error is raised, the :class:`.Transaction`
3215 is rolled back.
3217 .. seealso::
3219 :meth:`_engine.Engine.connect` - procure a
3220 :class:`_engine.Connection` from
3221 an :class:`_engine.Engine`.
3223 :meth:`_engine.Connection.begin` - start a :class:`.Transaction`
3224 for a particular :class:`_engine.Connection`.
3226 """ # noqa: E501
3227 with self.connect() as conn:
3228 with conn.begin():
3229 yield conn
3231 def _run_ddl_visitor(
3232 self,
3233 visitorcallable: Type[InvokeDDLBase],
3234 element: SchemaVisitable,
3235 **kwargs: Any,
3236 ) -> None:
3237 with self.begin() as conn:
3238 conn._run_ddl_visitor(visitorcallable, element, **kwargs)
3240 def connect(self) -> Connection:
3241 """Return a new :class:`_engine.Connection` object.
3243 The :class:`_engine.Connection` acts as a Python context manager, so
3244 the typical use of this method looks like::
3246 with engine.connect() as connection:
3247 connection.execute(text("insert into table values ('foo')"))
3248 connection.commit()
3250 Where above, after the block is completed, the connection is "closed"
3251 and its underlying DBAPI resources are returned to the connection pool.
3252 This also has the effect of rolling back any transaction that
3253 was explicitly begun or was begun via autobegin, and will
3254 emit the :meth:`_events.ConnectionEvents.rollback` event if one was
3255 started and is still in progress.
3257 .. seealso::
3259 :meth:`_engine.Engine.begin`
3261 """
3263 return self._connection_cls(self)
3265 def raw_connection(self) -> PoolProxiedConnection:
3266 """Return a "raw" DBAPI connection from the connection pool.
3268 The returned object is a proxied version of the DBAPI
3269 connection object used by the underlying driver in use.
3270 The object will have all the same behavior as the real DBAPI
3271 connection, except that its ``close()`` method will result in the
3272 connection being returned to the pool, rather than being closed
3273 for real.
3275 This method provides direct DBAPI connection access for
3276 special situations when the API provided by
3277 :class:`_engine.Connection`
3278 is not needed. When a :class:`_engine.Connection` object is already
3279 present, the DBAPI connection is available using
3280 the :attr:`_engine.Connection.connection` accessor.
3282 .. seealso::
3284 :ref:`dbapi_connections`
3286 """
3287 return self.pool.connect()
3290class OptionEngineMixin(log.Identified):
3291 _sa_propagate_class_events = False
3293 dispatch: dispatcher[ConnectionEventsTarget]
3294 _compiled_cache: Optional[CompiledCacheType]
3295 dialect: Dialect
3296 pool: Pool
3297 url: URL
3298 hide_parameters: bool
3299 echo: log.echo_property
3301 def __init__(
3302 self, proxied: Engine, execution_options: CoreExecuteOptionsParameter
3303 ):
3304 self._proxied = proxied
3305 self.url = proxied.url
3306 self.dialect = proxied.dialect
3307 self.logging_name = proxied.logging_name
3308 self.echo = proxied.echo
3309 self._compiled_cache = proxied._compiled_cache
3310 self.hide_parameters = proxied.hide_parameters
3311 log.instance_logger(self, echoflag=self.echo)
3313 # note: this will propagate events that are assigned to the parent
3314 # engine after this OptionEngine is created. Since we share
3315 # the events of the parent we also disallow class-level events
3316 # to apply to the OptionEngine class directly.
3317 #
3318 # the other way this can work would be to transfer existing
3319 # events only, using:
3320 # self.dispatch._update(proxied.dispatch)
3321 #
3322 # that might be more appropriate however it would be a behavioral
3323 # change for logic that assigns events to the parent engine and
3324 # would like it to take effect for the already-created sub-engine.
3325 self.dispatch = self.dispatch._join(proxied.dispatch)
3327 self._execution_options = proxied._execution_options
3328 self.update_execution_options(**execution_options)
3330 def update_execution_options(self, **opt: Any) -> None:
3331 raise NotImplementedError()
3333 if not typing.TYPE_CHECKING:
3334 # https://github.com/python/typing/discussions/1095
3336 @property
3337 def pool(self) -> Pool:
3338 return self._proxied.pool
3340 @pool.setter
3341 def pool(self, pool: Pool) -> None:
3342 self._proxied.pool = pool
3344 @property
3345 def _has_events(self) -> bool:
3346 return self._proxied._has_events or self.__dict__.get(
3347 "_has_events", False
3348 )
3350 @_has_events.setter
3351 def _has_events(self, value: bool) -> None:
3352 self.__dict__["_has_events"] = value
3355class OptionEngine(OptionEngineMixin, Engine):
3356 def update_execution_options(self, **opt: Any) -> None:
3357 Engine.update_execution_options(self, **opt)
3360Engine._option_cls = OptionEngine