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 # set to True at the class level by ConnectionEvents when listeners
133 # are associated with the Connection class; set at the instance level
134 # when listeners are associated with a specific Connection
135 _has_events: bool = False
137 _transaction: Optional[RootTransaction]
138 _nested_transaction: Optional[NestedTransaction]
140 def __init__(
141 self,
142 engine: Engine,
143 connection: Optional[PoolProxiedConnection] = None,
144 _has_events: Optional[bool] = None,
145 _allow_revalidate: bool = True,
146 _allow_autobegin: bool = True,
147 ):
148 """Construct a new Connection."""
149 self.engine = engine
150 self.dialect = dialect = engine.dialect
152 if connection is None:
153 try:
154 self._dbapi_connection = engine.raw_connection()
155 except dialect.loaded_dbapi.Error as err:
156 Connection._handle_dbapi_exception_noconnection(
157 err, dialect, engine
158 )
159 raise
160 else:
161 self._dbapi_connection = connection
163 self._transaction = self._nested_transaction = None
164 self.__savepoint_seq = 0
165 self.__in_begin = False
167 self.__can_reconnect = _allow_revalidate
168 self._allow_autobegin = _allow_autobegin
169 self._echo = self.engine._should_log_info()
171 if _has_events is None:
172 # if _has_events is sent explicitly as False,
173 # then don't join the dispatch of the engine; we don't
174 # want to handle any of the engine's events in that case.
175 # the engine's _has_events is checked alongside our own at
176 # each event point; leave our own _has_events unset so that
177 # the class-level flag, set by listeners on the Connection
178 # class, takes effect
179 self.dispatch = self.dispatch._join(engine.dispatch)
180 else:
181 self._has_events = _has_events
183 self._execution_options = engine._execution_options
185 if self._has_events or self.engine._has_events:
186 self.dispatch.engine_connect(self)
188 # this can be assigned differently via
189 # characteristics.LoggingTokenCharacteristic
190 _message_formatter: Any = None
192 def _log_parameters(
193 self,
194 parameters: _AnyExecuteParams | None,
195 stats: str,
196 *,
197 ismulti: bool,
198 ) -> None:
199 if not self.engine.hide_parameters:
200 self._log_info(
201 "[%s] %r",
202 stats,
203 sql_util._repr_params(
204 parameters,
205 batches=10,
206 ismulti=ismulti,
207 ),
208 )
209 else:
210 self._log_info(
211 "[%s] [SQL parameters hidden due to hide_parameters=True]",
212 stats,
213 )
215 def _log_info(self, message: str, *arg: Any, **kw: Any) -> None:
216 fmt = self._message_formatter
218 if fmt:
219 message = fmt(message)
221 if log.STACKLEVEL:
222 kw["stacklevel"] = 1 + log.STACKLEVEL_OFFSET
224 self.engine.logger.info(message, *arg, **kw)
226 def _log_debug(self, message: str, *arg: Any, **kw: Any) -> None:
227 fmt = self._message_formatter
229 if fmt:
230 message = fmt(message)
232 if log.STACKLEVEL:
233 kw["stacklevel"] = 1 + log.STACKLEVEL_OFFSET
235 self.engine.logger.debug(message, *arg, **kw)
237 @property
238 def _schema_translate_map(self) -> Optional[SchemaTranslateMapType]:
239 schema_translate_map: Optional[SchemaTranslateMapType] = (
240 self._execution_options.get("schema_translate_map", None)
241 )
243 return schema_translate_map
245 def schema_for_object(self, obj: HasSchemaAttr) -> Optional[str]:
246 """Return the schema name for the given schema item taking into
247 account current schema translate map.
249 """
251 name = obj.schema
252 schema_translate_map: Optional[SchemaTranslateMapType] = (
253 self._execution_options.get("schema_translate_map", None)
254 )
256 if (
257 schema_translate_map
258 and name in schema_translate_map
259 and obj._use_schema_map
260 ):
261 return schema_translate_map[name]
262 else:
263 return name
265 def __enter__(self) -> Connection:
266 return self
268 def __exit__(self, type_: Any, value: Any, traceback: Any) -> None:
269 self.close()
271 @overload
272 def execution_options(
273 self,
274 *,
275 compiled_cache: Optional[CompiledCacheType] = ...,
276 logging_token: str = ...,
277 isolation_level: IsolationLevel = ...,
278 no_parameters: bool = False,
279 stream_results: bool = False,
280 max_row_buffer: int = ...,
281 yield_per: int = ...,
282 insertmanyvalues_page_size: int = ...,
283 schema_translate_map: Optional[SchemaTranslateMapType] = ...,
284 preserve_rowcount: bool = False,
285 driver_column_names: bool = False,
286 **opt: Any,
287 ) -> Connection: ...
289 @overload
290 def execution_options(self, **opt: Any) -> Connection: ...
292 def execution_options(self, **opt: Any) -> Connection:
293 r"""Set non-SQL options for the connection which take effect
294 during execution.
296 This method modifies this :class:`_engine.Connection` **in-place**;
297 the return value is the same :class:`_engine.Connection` object
298 upon which the method is called. Note that this is in contrast
299 to the behavior of the ``execution_options`` methods on other
300 objects such as :meth:`_engine.Engine.execution_options` and
301 :meth:`_sql.Executable.execution_options`. The rationale is that many
302 such execution options necessarily modify the state of the base
303 DBAPI connection in any case so there is no feasible means of
304 keeping the effect of such an option localized to a "sub" connection.
306 .. versionchanged:: 2.0 The :meth:`_engine.Connection.execution_options`
307 method, in contrast to other objects with this method, modifies
308 the connection in-place without creating copy of it.
310 As discussed elsewhere, the :meth:`_engine.Connection.execution_options`
311 method accepts any arbitrary parameters including user defined names.
312 All parameters given are consumable in a number of ways including
313 by using the :meth:`_engine.Connection.get_execution_options` method.
314 See the examples at :meth:`_sql.Executable.execution_options`
315 and :meth:`_engine.Engine.execution_options`.
317 The keywords that are currently recognized by SQLAlchemy itself
318 include all those listed under :meth:`.Executable.execution_options`,
319 as well as others that are specific to :class:`_engine.Connection`.
321 :param compiled_cache: Available on: :class:`_engine.Connection`,
322 :class:`_engine.Engine`.
324 A dictionary where :class:`.Compiled` objects
325 will be cached when the :class:`_engine.Connection`
326 compiles a clause
327 expression into a :class:`.Compiled` object. This dictionary will
328 supersede the statement cache that may be configured on the
329 :class:`_engine.Engine` itself. If set to None, caching
330 is disabled, even if the engine has a configured cache size.
332 Note that the ORM makes use of its own "compiled" caches for
333 some operations, including flush operations. The caching
334 used by the ORM internally supersedes a cache dictionary
335 specified here.
337 :param logging_token: Available on: :class:`_engine.Connection`,
338 :class:`_engine.Engine`, :class:`_sql.Executable`.
340 Adds the specified string token surrounded by brackets in log
341 messages logged by the connection, i.e. the logging that's enabled
342 either via the :paramref:`_sa.create_engine.echo` flag or via the
343 ``logging.getLogger("sqlalchemy.engine")`` logger. This allows a
344 per-connection or per-sub-engine token to be available which is
345 useful for debugging concurrent connection scenarios.
347 .. versionadded:: 1.4.0b2
349 .. seealso::
351 :ref:`dbengine_logging_tokens` - usage example
353 :paramref:`_sa.create_engine.logging_name` - adds a name to the
354 name used by the Python logger object itself.
356 :param isolation_level: Available on: :class:`_engine.Connection`,
357 :class:`_engine.Engine`.
359 Set the transaction isolation level for the lifespan of this
360 :class:`_engine.Connection` object.
361 Valid values include those string
362 values accepted by the :paramref:`_sa.create_engine.isolation_level`
363 parameter passed to :func:`_sa.create_engine`. These levels are
364 semi-database specific; see individual dialect documentation for
365 valid levels.
367 The isolation level option applies the isolation level by emitting
368 statements on the DBAPI connection, and **necessarily affects the
369 original Connection object overall**. The isolation level will remain
370 at the given setting until explicitly changed, or when the DBAPI
371 connection itself is :term:`released` to the connection pool, i.e. the
372 :meth:`_engine.Connection.close` method is called, at which time an
373 event handler will emit additional statements on the DBAPI connection
374 in order to revert the isolation level change.
376 .. note:: The ``isolation_level`` execution option may only be
377 established before the :meth:`_engine.Connection.begin` method is
378 called, as well as before any SQL statements are emitted which
379 would otherwise trigger "autobegin", or directly after a call to
380 :meth:`_engine.Connection.commit` or
381 :meth:`_engine.Connection.rollback`. A database cannot change the
382 isolation level on a transaction in progress.
384 .. note:: The ``isolation_level`` execution option is implicitly
385 reset if the :class:`_engine.Connection` is invalidated, e.g. via
386 the :meth:`_engine.Connection.invalidate` method, or if a
387 disconnection error occurs. The new connection produced after the
388 invalidation will **not** have the selected isolation level
389 re-applied to it automatically.
391 .. seealso::
393 :ref:`dbapi_autocommit`
395 :meth:`_engine.Connection.get_isolation_level`
396 - view current actual level
398 :param no_parameters: Available on: :class:`_engine.Connection`,
399 :class:`_sql.Executable`.
401 When ``True``, if the final parameter
402 list or dictionary is totally empty, will invoke the
403 statement on the cursor as ``cursor.execute(statement)``,
404 not passing the parameter collection at all.
405 Some DBAPIs such as psycopg2 and mysql-python consider
406 percent signs as significant only when parameters are
407 present; this option allows code to generate SQL
408 containing percent signs (and possibly other characters)
409 that is neutral regarding whether it's executed by the DBAPI
410 or piped into a script that's later invoked by
411 command line tools.
413 :param stream_results: Available on: :class:`_engine.Connection`,
414 :class:`_sql.Executable`.
416 Indicate to the dialect that results should be "streamed" and not
417 pre-buffered, if possible. For backends such as PostgreSQL, MySQL
418 and MariaDB, this indicates the use of a "server side cursor" as
419 opposed to a client side cursor. Other backends such as that of
420 Oracle Database may already use server side cursors by default.
422 The usage of
423 :paramref:`_engine.Connection.execution_options.stream_results` is
424 usually combined with setting a fixed number of rows to to be fetched
425 in batches, to allow for efficient iteration of database rows while
426 at the same time not loading all result rows into memory at once;
427 this can be configured on a :class:`_engine.Result` object using the
428 :meth:`_engine.Result.yield_per` method, after execution has
429 returned a new :class:`_engine.Result`. If
430 :meth:`_engine.Result.yield_per` is not used,
431 the :paramref:`_engine.Connection.execution_options.stream_results`
432 mode of operation will instead use a dynamically sized buffer
433 which buffers sets of rows at a time, growing on each batch
434 based on a fixed growth size up until a limit which may
435 be configured using the
436 :paramref:`_engine.Connection.execution_options.max_row_buffer`
437 parameter.
439 When using the ORM to fetch ORM mapped objects from a result,
440 :meth:`_engine.Result.yield_per` should always be used with
441 :paramref:`_engine.Connection.execution_options.stream_results`,
442 so that the ORM does not fetch all rows into new ORM objects at once.
444 For typical use, the
445 :paramref:`_engine.Connection.execution_options.yield_per` execution
446 option should be preferred, which sets up both
447 :paramref:`_engine.Connection.execution_options.stream_results` and
448 :meth:`_engine.Result.yield_per` at once. This option is supported
449 both at a core level by :class:`_engine.Connection` as well as by the
450 ORM :class:`_engine.Session`; the latter is described at
451 :ref:`orm_queryguide_yield_per`.
453 .. seealso::
455 :ref:`engine_stream_results` - background on
456 :paramref:`_engine.Connection.execution_options.stream_results`
458 :paramref:`_engine.Connection.execution_options.max_row_buffer`
460 :paramref:`_engine.Connection.execution_options.yield_per`
462 :ref:`orm_queryguide_yield_per` - in the :ref:`queryguide_toplevel`
463 describing the ORM version of ``yield_per``
465 :param max_row_buffer: Available on: :class:`_engine.Connection`,
466 :class:`_sql.Executable`. Sets a maximum
467 buffer size to use when the
468 :paramref:`_engine.Connection.execution_options.stream_results`
469 execution option is used on a backend that supports server side
470 cursors. The default value if not specified is 1000.
472 .. seealso::
474 :paramref:`_engine.Connection.execution_options.stream_results`
476 :ref:`engine_stream_results`
479 :param yield_per: Available on: :class:`_engine.Connection`,
480 :class:`_sql.Executable`. Integer value applied which will
481 set the :paramref:`_engine.Connection.execution_options.stream_results`
482 execution option and invoke :meth:`_engine.Result.yield_per`
483 automatically at once. Allows equivalent functionality as
484 is present when using this parameter with the ORM.
486 .. versionadded:: 1.4.40
488 .. seealso::
490 :ref:`engine_stream_results` - background and examples
491 on using server side cursors with Core.
493 :ref:`orm_queryguide_yield_per` - in the :ref:`queryguide_toplevel`
494 describing the ORM version of ``yield_per``
496 :param insertmanyvalues_page_size: Available on: :class:`_engine.Connection`,
497 :class:`_engine.Engine`. Number of rows to format into an
498 INSERT statement when the statement uses "insertmanyvalues" mode,
499 which is a paged form of bulk insert that is used for many backends
500 when using :term:`executemany` execution typically in conjunction
501 with RETURNING. Defaults to 1000. May also be modified on a
502 per-engine basis using the
503 :paramref:`_sa.create_engine.insertmanyvalues_page_size` parameter.
505 .. versionadded:: 2.0
507 .. seealso::
509 :ref:`engine_insertmanyvalues`
511 :param schema_translate_map: Available on: :class:`_engine.Connection`,
512 :class:`_engine.Engine`, :class:`_sql.Executable`.
514 A dictionary mapping schema names to schema names, that will be
515 applied to the :paramref:`_schema.Table.schema` element of each
516 :class:`_schema.Table`
517 encountered when SQL or DDL expression elements
518 are compiled into strings; the resulting schema name will be
519 converted based on presence in the map of the original name.
521 .. seealso::
523 :ref:`schema_translating`
525 :param preserve_rowcount: Boolean; when True, the ``cursor.rowcount``
526 attribute will be unconditionally memoized within the result and
527 made available via the :attr:`.CursorResult.rowcount` attribute.
528 Normally, this attribute is only preserved for UPDATE and DELETE
529 statements. Using this option, the DBAPIs rowcount value can
530 be accessed for other kinds of statements such as INSERT and SELECT,
531 to the degree that the DBAPI supports these statements. See
532 :attr:`.CursorResult.rowcount` for notes regarding the behavior
533 of this attribute.
535 .. versionadded:: 2.0.28
537 .. seealso::
539 :meth:`_engine.Engine.execution_options`
541 :meth:`.Executable.execution_options`
543 :meth:`_engine.Connection.get_execution_options`
545 :ref:`orm_queryguide_execution_options` - documentation on all
546 ORM-specific execution options
548 :param driver_column_names: When True, the returned
549 :class:`_engine.CursorResult` will use the column names as written in
550 ``cursor.description`` to set up the keys for the result set,
551 including the names of columns for the :class:`_engine.Row` object as
552 well as the dictionary keys when using :attr:`_engine.Row._mapping`.
553 On backends that use "name normalization" such as Oracle Database to
554 correct for lower case names being converted to all uppercase, this
555 behavior is turned off and the raw UPPERCASE names in
556 cursor.description will be present.
558 .. versionadded:: 2.1
560 """ # noqa
561 if self._has_events or self.engine._has_events:
562 self.dispatch.set_connection_execution_options(self, opt)
563 self._execution_options = self._execution_options.union(opt)
564 self.dialect.set_connection_execution_options(self, opt)
565 return self
567 def get_execution_options(self) -> _ExecuteOptions:
568 """Get the non-SQL options which will take effect during execution.
570 .. seealso::
572 :meth:`_engine.Connection.execution_options`
573 """
574 return self._execution_options
576 @property
577 def _still_open_and_dbapi_connection_is_valid(self) -> bool:
578 pool_proxied_connection = self._dbapi_connection
579 return (
580 pool_proxied_connection is not None
581 and pool_proxied_connection.is_valid
582 )
584 @property
585 def closed(self) -> bool:
586 """Return True if this connection is closed."""
588 return self._dbapi_connection is None and not self.__can_reconnect
590 @property
591 def invalidated(self) -> bool:
592 """Return True if this connection was invalidated.
594 This does not indicate whether or not the connection was
595 invalidated at the pool level, however
597 """
599 # prior to 1.4, "invalid" was stored as a state independent of
600 # "closed", meaning an invalidated connection could be "closed",
601 # the _dbapi_connection would be None and closed=True, yet the
602 # "invalid" flag would stay True. This meant that there were
603 # three separate states (open/valid, closed/valid, closed/invalid)
604 # when there is really no reason for that; a connection that's
605 # "closed" does not need to be "invalid". So the state is now
606 # represented by the two facts alone.
608 pool_proxied_connection = self._dbapi_connection
609 return pool_proxied_connection is None and self.__can_reconnect
611 @property
612 def connection(self) -> PoolProxiedConnection:
613 """The underlying DB-API connection managed by this Connection.
615 This is a SQLAlchemy connection-pool proxied connection
616 which then has the attribute
617 :attr:`_pool._ConnectionFairy.dbapi_connection` that refers to the
618 actual driver connection.
620 .. seealso::
623 :ref:`dbapi_connections`
625 """
627 if self._dbapi_connection is None:
628 try:
629 return self._revalidate_connection()
630 except (exc.PendingRollbackError, exc.ResourceClosedError):
631 raise
632 except BaseException as e:
633 self._handle_dbapi_exception(e, None, None, None, None)
634 else:
635 return self._dbapi_connection
637 def get_isolation_level(self) -> IsolationLevel:
638 """Return the current **actual** isolation level that's present on
639 the database within the scope of this connection.
641 This attribute will perform a live SQL operation against the database
642 in order to procure the current isolation level, so the value returned
643 is the actual level on the underlying DBAPI connection regardless of
644 how this state was set. This will be one of the four actual isolation
645 modes ``READ UNCOMMITTED``, ``READ COMMITTED``, ``REPEATABLE READ``,
646 ``SERIALIZABLE``. It will **not** include the ``AUTOCOMMIT`` isolation
647 level setting. Third party dialects may also feature additional
648 isolation level settings.
650 .. note:: This method **will not report** on the ``AUTOCOMMIT``
651 isolation level, which is a separate :term:`dbapi` setting that's
652 independent of **actual** isolation level. When ``AUTOCOMMIT`` is
653 in use, the database connection still has a "traditional" isolation
654 mode in effect, that is typically one of the four values
655 ``READ UNCOMMITTED``, ``READ COMMITTED``, ``REPEATABLE READ``,
656 ``SERIALIZABLE``.
658 Compare to the :attr:`_engine.Connection.default_isolation_level`
659 accessor which returns the isolation level that is present on the
660 database at initial connection time.
662 .. seealso::
664 :attr:`_engine.Connection.default_isolation_level`
665 - view default level
667 :paramref:`_sa.create_engine.isolation_level`
668 - set per :class:`_engine.Engine` isolation level
670 :paramref:`.Connection.execution_options.isolation_level`
671 - set per :class:`_engine.Connection` isolation level
673 """
674 dbapi_connection = self.connection.dbapi_connection
675 assert dbapi_connection is not None
676 try:
677 return self.dialect.get_isolation_level(dbapi_connection)
678 except BaseException as e:
679 self._handle_dbapi_exception(e, None, None, None, None)
681 @property
682 def default_isolation_level(self) -> Optional[IsolationLevel]:
683 """The initial-connection time isolation level associated with the
684 :class:`_engine.Dialect` in use.
686 This value is independent of the
687 :paramref:`.Connection.execution_options.isolation_level` and
688 :paramref:`.Engine.execution_options.isolation_level` execution
689 options, and is determined by the :class:`_engine.Dialect` when the
690 first connection is created, by performing a SQL query against the
691 database for the current isolation level before any additional commands
692 have been emitted.
694 Calling this accessor does not invoke any new SQL queries.
696 .. seealso::
698 :meth:`_engine.Connection.get_isolation_level`
699 - view current actual isolation level
701 :paramref:`_sa.create_engine.isolation_level`
702 - set per :class:`_engine.Engine` isolation level
704 :paramref:`.Connection.execution_options.isolation_level`
705 - set per :class:`_engine.Connection` isolation level
707 """
708 return self.dialect.default_isolation_level
710 def _invalid_transaction(self) -> NoReturn:
711 raise exc.PendingRollbackError(
712 "Can't reconnect until invalid %stransaction is rolled "
713 "back. Please rollback() fully before proceeding"
714 % ("savepoint " if self._nested_transaction is not None else ""),
715 code="8s2b",
716 )
718 def _revalidate_connection(self) -> PoolProxiedConnection:
719 if self.__can_reconnect and self.invalidated:
720 if self._transaction is not None:
721 self._invalid_transaction()
722 self._dbapi_connection = self.engine.raw_connection()
723 return self._dbapi_connection
724 raise exc.ResourceClosedError("This Connection is closed")
726 @property
727 def info(self) -> _InfoType:
728 """Info dictionary associated with the underlying DBAPI connection
729 referred to by this :class:`_engine.Connection`, allowing user-defined
730 data to be associated with the connection.
732 The data here will follow along with the DBAPI connection including
733 after it is returned to the connection pool and used again
734 in subsequent instances of :class:`_engine.Connection`.
736 """
738 return self.connection.info
740 def invalidate(self, exception: Optional[BaseException] = None) -> None:
741 """Invalidate the underlying DBAPI connection associated with
742 this :class:`_engine.Connection`.
744 An attempt will be made to close the underlying DBAPI connection
745 immediately; however if this operation fails, the error is logged
746 but not raised. The connection is then discarded whether or not
747 close() succeeded.
749 Upon the next use (where "use" typically means using the
750 :meth:`_engine.Connection.execute` method or similar),
751 this :class:`_engine.Connection` will attempt to
752 procure a new DBAPI connection using the services of the
753 :class:`_pool.Pool` as a source of connectivity (e.g.
754 a "reconnection").
756 If a transaction was in progress (e.g. the
757 :meth:`_engine.Connection.begin` method has been called) when
758 :meth:`_engine.Connection.invalidate` method is called, at the DBAPI
759 level all state associated with this transaction is lost, as
760 the DBAPI connection is closed. The :class:`_engine.Connection`
761 will not allow a reconnection to proceed until the
762 :class:`.Transaction` object is ended, by calling the
763 :meth:`.Transaction.rollback` method; until that point, any attempt at
764 continuing to use the :class:`_engine.Connection` will raise an
765 :class:`~sqlalchemy.exc.InvalidRequestError`.
766 This is to prevent applications from accidentally
767 continuing an ongoing transactional operations despite the
768 fact that the transaction has been lost due to an
769 invalidation.
771 The :meth:`_engine.Connection.invalidate` method,
772 just like auto-invalidation,
773 will at the connection pool level invoke the
774 :meth:`_events.PoolEvents.invalidate` event.
776 :param exception: an optional ``Exception`` instance that's the
777 reason for the invalidation. is passed along to event handlers
778 and logging functions.
780 .. seealso::
782 :ref:`pool_connection_invalidation`
784 """
786 if self.invalidated:
787 return
789 if self.closed:
790 raise exc.ResourceClosedError("This Connection is closed")
792 try:
793 if self._still_open_and_dbapi_connection_is_valid:
794 pool_proxied_connection = self._dbapi_connection
795 assert pool_proxied_connection is not None
796 pool_proxied_connection.invalidate(exception)
797 finally:
798 # this is always a hard invalidation, which closes the DBAPI
799 # connection; for an asyncio driver that close is an awaited
800 # terminate() which a cancellation can interrupt. this
801 # Connection is invalidated regardless, so that nothing
802 # further, such as a rollback, is emitted on the DBAPI
803 # connection
804 self._dbapi_connection = None
806 def detach(self) -> None:
807 """Detach the underlying DB-API connection from its connection pool.
809 E.g.::
811 with engine.connect() as conn:
812 conn.detach()
813 conn.execute(text("SET search_path TO schema1, schema2"))
815 # work with connection
817 # connection is fully closed (since we used "with:", can
818 # also call .close())
820 This :class:`_engine.Connection` instance will remain usable.
821 When closed
822 (or exited from a context manager context as above),
823 the DB-API connection will be literally closed and not
824 returned to its originating pool.
826 This method can be used to insulate the rest of an application
827 from a modified state on a connection (such as a transaction
828 isolation level or similar).
830 """
832 if self.closed:
833 raise exc.ResourceClosedError("This Connection is closed")
835 pool_proxied_connection = self._dbapi_connection
836 if pool_proxied_connection is None:
837 raise exc.InvalidRequestError(
838 "Can't detach an invalidated Connection"
839 )
840 pool_proxied_connection.detach()
842 def _autobegin(self) -> None:
843 if self._allow_autobegin and not self.__in_begin:
844 self.begin()
846 def begin(self) -> RootTransaction:
847 """Begin a transaction prior to autobegin occurring.
849 E.g.::
851 with engine.connect() as conn:
852 with conn.begin() as trans:
853 conn.execute(table.insert(), {"username": "sandy"})
855 The returned object is an instance of :class:`_engine.RootTransaction`.
856 This object represents the "scope" of the transaction,
857 which completes when either the :meth:`_engine.Transaction.rollback`
858 or :meth:`_engine.Transaction.commit` method is called; the object
859 also works as a context manager as illustrated above.
861 The :meth:`_engine.Connection.begin` method begins a
862 transaction that normally will be begun in any case when the connection
863 is first used to execute a statement. The reason this method might be
864 used would be to invoke the :meth:`_events.ConnectionEvents.begin`
865 event at a specific time, or to organize code within the scope of a
866 connection checkout in terms of context managed blocks, such as::
868 with engine.connect() as conn:
869 with conn.begin():
870 conn.execute(...)
871 conn.execute(...)
873 with conn.begin():
874 conn.execute(...)
875 conn.execute(...)
877 The above code is not fundamentally any different in its behavior than
878 the following code which does not use
879 :meth:`_engine.Connection.begin`; the below style is known
880 as "commit as you go" style::
882 with engine.connect() as conn:
883 conn.execute(...)
884 conn.execute(...)
885 conn.commit()
887 conn.execute(...)
888 conn.execute(...)
889 conn.commit()
891 From a database point of view, the :meth:`_engine.Connection.begin`
892 method does not emit any SQL or change the state of the underlying
893 DBAPI connection in any way; the Python DBAPI does not have any
894 concept of explicit transaction begin.
896 .. seealso::
898 :ref:`tutorial_working_with_transactions` - in the
899 :ref:`unified_tutorial`
901 :meth:`_engine.Connection.begin_nested` - use a SAVEPOINT
903 :meth:`_engine.Connection.begin_twophase` -
904 use a two phase /XID transaction
906 :meth:`_engine.Engine.begin` - context manager available from
907 :class:`_engine.Engine`
909 """
910 if self._transaction is None:
911 self._transaction = RootTransaction(self)
912 return self._transaction
913 else:
914 raise exc.InvalidRequestError(
915 "This connection has already initialized a SQLAlchemy "
916 "Transaction() object via begin() or autobegin; can't "
917 "call begin() here unless rollback() or commit() "
918 "is called first."
919 )
921 def begin_nested(self) -> NestedTransaction:
922 """Begin a nested transaction (i.e. SAVEPOINT) and return a transaction
923 handle that controls the scope of the SAVEPOINT.
925 E.g.::
927 with engine.begin() as connection:
928 with connection.begin_nested():
929 connection.execute(table.insert(), {"username": "sandy"})
931 The returned object is an instance of
932 :class:`_engine.NestedTransaction`, which includes transactional
933 methods :meth:`_engine.NestedTransaction.commit` and
934 :meth:`_engine.NestedTransaction.rollback`; for a nested transaction,
935 these methods correspond to the operations "RELEASE SAVEPOINT <name>"
936 and "ROLLBACK TO SAVEPOINT <name>". The name of the savepoint is local
937 to the :class:`_engine.NestedTransaction` object and is generated
938 automatically. Like any other :class:`_engine.Transaction`, the
939 :class:`_engine.NestedTransaction` may be used as a context manager as
940 illustrated above which will "release" or "rollback" corresponding to
941 if the operation within the block were successful or raised an
942 exception.
944 Nested transactions require SAVEPOINT support in the underlying
945 database, else the behavior is undefined. SAVEPOINT is commonly used to
946 run operations within a transaction that may fail, while continuing the
947 outer transaction. E.g.::
949 from sqlalchemy import exc
951 with engine.begin() as connection:
952 trans = connection.begin_nested()
953 try:
954 connection.execute(table.insert(), {"username": "sandy"})
955 trans.commit()
956 except exc.IntegrityError: # catch for duplicate username
957 trans.rollback() # rollback to savepoint
959 # outer transaction continues
960 connection.execute(...)
962 If :meth:`_engine.Connection.begin_nested` is called without first
963 calling :meth:`_engine.Connection.begin` or
964 :meth:`_engine.Engine.begin`, the :class:`_engine.Connection` object
965 will "autobegin" the outer transaction first. This outer transaction
966 may be committed using "commit-as-you-go" style, e.g.::
968 with engine.connect() as connection: # begin() wasn't called
970 with connection.begin_nested(): # will auto-"begin()" first
971 connection.execute(...)
972 # savepoint is released
974 connection.execute(...)
976 # explicitly commit outer transaction
977 connection.commit()
979 # can continue working with connection here
981 .. versionchanged:: 2.0
983 :meth:`_engine.Connection.begin_nested` will now participate
984 in the connection "autobegin" behavior that is new as of
985 2.0 / "future" style connections in 1.4.
987 .. seealso::
989 :meth:`_engine.Connection.begin`
991 :ref:`session_begin_nested` - ORM support for SAVEPOINT
993 """
994 if self._transaction is None:
995 self._autobegin()
997 return NestedTransaction(self)
999 def begin_twophase(self, xid: Optional[Any] = None) -> TwoPhaseTransaction:
1000 """Begin a two-phase or XA transaction and return a transaction
1001 handle.
1003 The returned object is an instance of :class:`.TwoPhaseTransaction`,
1004 which in addition to the methods provided by
1005 :class:`.Transaction`, also provides a
1006 :meth:`~.TwoPhaseTransaction.prepare` method.
1008 :param xid: the two phase transaction id. If not supplied, a
1009 random id will be generated. The accepted type and value depends on
1010 the driver in use.
1012 .. seealso::
1014 :meth:`_engine.Connection.begin`
1016 :meth:`_engine.Connection.begin_twophase`
1018 """
1020 if self._transaction is not None:
1021 raise exc.InvalidRequestError(
1022 "Cannot start a two phase transaction when a transaction "
1023 "is already in progress."
1024 )
1025 if xid is None:
1026 xid = self.engine.dialect.create_xid()
1027 return TwoPhaseTransaction(self, xid)
1029 def commit(self) -> None:
1030 """Commit the transaction that is currently in progress.
1032 This method commits the current transaction if one has been started.
1033 If no transaction was started, the method has no effect, assuming
1034 the connection is in a non-invalidated state.
1036 A transaction is begun on a :class:`_engine.Connection` automatically
1037 whenever a statement is first executed, or when the
1038 :meth:`_engine.Connection.begin` method is called.
1040 .. note:: The :meth:`_engine.Connection.commit` method only acts upon
1041 the primary database transaction that is linked to the
1042 :class:`_engine.Connection` object. It does not operate upon a
1043 SAVEPOINT that would have been invoked from the
1044 :meth:`_engine.Connection.begin_nested` method; for control of a
1045 SAVEPOINT, call :meth:`_engine.NestedTransaction.commit` on the
1046 :class:`_engine.NestedTransaction` that is returned by the
1047 :meth:`_engine.Connection.begin_nested` method itself.
1050 """
1051 if self._transaction:
1052 self._transaction.commit()
1054 def rollback(self) -> None:
1055 """Roll back the transaction that is currently in progress.
1057 This method rolls back the current transaction if one has been started.
1058 If no transaction was started, the method has no effect. If a
1059 transaction was started and the connection is in an invalidated state,
1060 the transaction is cleared using this method.
1062 A transaction is begun on a :class:`_engine.Connection` automatically
1063 whenever a statement is first executed, or when the
1064 :meth:`_engine.Connection.begin` method is called.
1066 .. note:: The :meth:`_engine.Connection.rollback` method only acts
1067 upon the primary database transaction that is linked to the
1068 :class:`_engine.Connection` object. It does not operate upon a
1069 SAVEPOINT that would have been invoked from the
1070 :meth:`_engine.Connection.begin_nested` method; for control of a
1071 SAVEPOINT, call :meth:`_engine.NestedTransaction.rollback` on the
1072 :class:`_engine.NestedTransaction` that is returned by the
1073 :meth:`_engine.Connection.begin_nested` method itself.
1076 """
1077 if self._transaction:
1078 self._transaction.rollback()
1080 def recover_twophase(self) -> List[Any]:
1081 return self.engine.dialect.do_recover_twophase(self)
1083 def rollback_prepared(self, xid: Any, recover: bool = False) -> None:
1084 self.engine.dialect.do_rollback_twophase(self, xid, recover=recover)
1086 def commit_prepared(self, xid: Any, recover: bool = False) -> None:
1087 self.engine.dialect.do_commit_twophase(self, xid, recover=recover)
1089 def in_transaction(self) -> bool:
1090 """Return True if a transaction is in progress."""
1091 return self._transaction is not None and self._transaction.is_active
1093 def in_nested_transaction(self) -> bool:
1094 """Return True if a transaction is in progress."""
1095 return (
1096 self._nested_transaction is not None
1097 and self._nested_transaction.is_active
1098 )
1100 def _is_autocommit_isolation(self) -> bool:
1101 opt_iso = self._execution_options.get("isolation_level", None)
1102 return bool(
1103 opt_iso == "AUTOCOMMIT"
1104 or (
1105 opt_iso is None
1106 and self.engine.dialect._on_connect_isolation_level
1107 == "AUTOCOMMIT"
1108 )
1109 )
1111 def _get_required_transaction(self) -> RootTransaction:
1112 trans = self._transaction
1113 if trans is None:
1114 raise exc.InvalidRequestError("connection is not in a transaction")
1115 return trans
1117 def _get_required_nested_transaction(self) -> NestedTransaction:
1118 trans = self._nested_transaction
1119 if trans is None:
1120 raise exc.InvalidRequestError(
1121 "connection is not in a nested transaction"
1122 )
1123 return trans
1125 def get_transaction(self) -> Optional[RootTransaction]:
1126 """Return the current root transaction in progress, if any.
1128 .. versionadded:: 1.4
1130 """
1132 return self._transaction
1134 def get_nested_transaction(self) -> Optional[NestedTransaction]:
1135 """Return the current nested transaction in progress, if any.
1137 .. versionadded:: 1.4
1139 """
1140 return self._nested_transaction
1142 def _begin_impl(self, transaction: RootTransaction) -> None:
1143 if self._echo:
1144 if self._is_autocommit_isolation():
1145 self._log_info(
1146 "BEGIN (implicit; DBAPI should not BEGIN due to "
1147 "autocommit mode)"
1148 )
1149 else:
1150 self._log_info("BEGIN (implicit)")
1152 self.__in_begin = True
1154 if self._has_events or self.engine._has_events:
1155 self.dispatch.begin(self)
1157 try:
1158 self.engine.dialect.do_begin(self.connection)
1159 except BaseException as e:
1160 self._handle_dbapi_exception(e, None, None, None, None)
1161 finally:
1162 self.__in_begin = False
1164 def _rollback_impl(self) -> None:
1165 if self._has_events or self.engine._has_events:
1166 self.dispatch.rollback(self)
1168 if self._still_open_and_dbapi_connection_is_valid:
1169 if self._echo:
1170 if self._is_autocommit_isolation():
1171 if self.dialect.skip_autocommit_rollback:
1172 self._log_info(
1173 "ROLLBACK will be skipped by "
1174 "skip_autocommit_rollback"
1175 )
1176 else:
1177 self._log_info(
1178 "ROLLBACK using DBAPI connection.rollback(); "
1179 "set skip_autocommit_rollback to prevent fully"
1180 )
1181 else:
1182 self._log_info("ROLLBACK")
1183 try:
1184 self.engine.dialect.do_rollback(self.connection)
1185 except BaseException as e:
1186 self._handle_dbapi_exception(e, None, None, None, None)
1188 def _commit_impl(self) -> None:
1189 if self._has_events or self.engine._has_events:
1190 self.dispatch.commit(self)
1192 if self._echo:
1193 if self._is_autocommit_isolation():
1194 self._log_info(
1195 "COMMIT using DBAPI connection.commit(), "
1196 "has no effect due to autocommit mode"
1197 )
1198 else:
1199 self._log_info("COMMIT")
1200 try:
1201 self.engine.dialect.do_commit(self.connection)
1202 except BaseException as e:
1203 self._handle_dbapi_exception(e, None, None, None, None)
1205 def _savepoint_impl(self, name: Optional[str] = None) -> str:
1206 if self._has_events or self.engine._has_events:
1207 self.dispatch.savepoint(self, name)
1209 if name is None:
1210 self.__savepoint_seq += 1
1211 name = "sa_savepoint_%s" % self.__savepoint_seq
1212 self.engine.dialect.do_savepoint(self, name)
1213 return name
1215 def _rollback_to_savepoint_impl(self, name: str) -> None:
1216 if self._has_events or self.engine._has_events:
1217 self.dispatch.rollback_savepoint(self, name, None)
1219 if self._still_open_and_dbapi_connection_is_valid:
1220 self.engine.dialect.do_rollback_to_savepoint(self, name)
1222 def _release_savepoint_impl(self, name: str) -> None:
1223 if self._has_events or self.engine._has_events:
1224 self.dispatch.release_savepoint(self, name, None)
1226 self.engine.dialect.do_release_savepoint(self, name)
1228 def _begin_twophase_impl(self, transaction: TwoPhaseTransaction) -> None:
1229 if self._echo:
1230 self._log_info("BEGIN TWOPHASE (implicit)")
1231 if self._has_events or self.engine._has_events:
1232 self.dispatch.begin_twophase(self, transaction.xid)
1234 self.__in_begin = True
1235 try:
1236 self.engine.dialect.do_begin_twophase(self, transaction.xid)
1237 except BaseException as e:
1238 self._handle_dbapi_exception(e, None, None, None, None)
1239 finally:
1240 self.__in_begin = False
1242 def _prepare_twophase_impl(self, xid: Any) -> None:
1243 if self._has_events or self.engine._has_events:
1244 self.dispatch.prepare_twophase(self, xid)
1246 assert isinstance(self._transaction, TwoPhaseTransaction)
1247 try:
1248 self.engine.dialect.do_prepare_twophase(self, xid)
1249 except BaseException as e:
1250 self._handle_dbapi_exception(e, None, None, None, None)
1252 def _rollback_twophase_impl(self, xid: Any, is_prepared: bool) -> None:
1253 if self._has_events or self.engine._has_events:
1254 self.dispatch.rollback_twophase(self, xid, is_prepared)
1256 if self._still_open_and_dbapi_connection_is_valid:
1257 assert isinstance(self._transaction, TwoPhaseTransaction)
1258 try:
1259 self.engine.dialect.do_rollback_twophase(
1260 self, xid, is_prepared
1261 )
1262 except BaseException as e:
1263 self._handle_dbapi_exception(e, None, None, None, None)
1265 def _commit_twophase_impl(self, xid: Any, is_prepared: bool) -> None:
1266 if self._has_events or self.engine._has_events:
1267 self.dispatch.commit_twophase(self, xid, is_prepared)
1269 assert isinstance(self._transaction, TwoPhaseTransaction)
1270 try:
1271 self.engine.dialect.do_commit_twophase(self, xid, is_prepared)
1272 except BaseException as e:
1273 self._handle_dbapi_exception(e, None, None, None, None)
1275 def close(self) -> None:
1276 """Close this :class:`_engine.Connection`.
1278 This results in a release of the underlying database
1279 resources, that is, the DBAPI connection referenced
1280 internally. The DBAPI connection is typically restored
1281 back to the connection-holding :class:`_pool.Pool` referenced
1282 by the :class:`_engine.Engine` that produced this
1283 :class:`_engine.Connection`. Any transactional state present on
1284 the DBAPI connection is also unconditionally released via
1285 the DBAPI connection's ``rollback()`` method, regardless
1286 of any :class:`.Transaction` object that may be
1287 outstanding with regards to this :class:`_engine.Connection`.
1289 This has the effect of also calling :meth:`_engine.Connection.rollback`
1290 if any transaction is in place.
1292 After :meth:`_engine.Connection.close` is called, the
1293 :class:`_engine.Connection` is permanently in a closed state,
1294 and will allow no further operations.
1296 """
1298 # if the rollback here raises, skip_reset stays False and the pool
1299 # does its own reset of the connection, invalidating it if that
1300 # fails also. see #13657
1301 skip_reset = False
1302 try:
1303 if self._transaction:
1304 self._transaction.close()
1305 skip_reset = True
1306 finally:
1307 try:
1308 if self._dbapi_connection is not None:
1309 conn = self._dbapi_connection
1311 # as we just closed the transaction, close the connection
1312 # pool connection without doing an additional reset
1313 if skip_reset:
1314 cast("_ConnectionFairy", conn)._close_special(
1315 transaction_reset=True
1316 )
1317 else:
1318 conn.close()
1319 finally:
1320 # There is a slight chance that conn.close() may have
1321 # triggered an invalidation here in which case
1322 # _dbapi_connection would already be None, however usually
1323 # it will be non-None here and in a "closed" state.
1324 #
1325 # there is also a chance that the conn.close() here could
1326 # itself be interrupted, such as by a cancellation; hence a
1327 # second try/finally is needed as well (see #13699)
1328 self._dbapi_connection = None
1329 self.__can_reconnect = False
1331 # special case to handle mypy issue:
1332 # https://github.com/python/mypy/issues/20651
1333 @overload
1334 def scalar(
1335 self,
1336 statement: TypedReturnsRows[Never],
1337 parameters: Optional[_CoreSingleExecuteParams] = None,
1338 *,
1339 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1340 ) -> Optional[Any]: ...
1342 @overload
1343 def scalar(
1344 self,
1345 statement: TypedReturnsRows[_T],
1346 parameters: Optional[_CoreSingleExecuteParams] = None,
1347 *,
1348 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1349 ) -> Optional[_T]: ...
1351 @overload
1352 def scalar(
1353 self,
1354 statement: Executable,
1355 parameters: Optional[_CoreSingleExecuteParams] = None,
1356 *,
1357 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1358 ) -> Any: ...
1360 def scalar(
1361 self,
1362 statement: Executable,
1363 parameters: Optional[_CoreSingleExecuteParams] = None,
1364 *,
1365 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1366 ) -> Any:
1367 r"""Executes a SQL statement construct and returns a scalar object.
1369 This method is shorthand for invoking the
1370 :meth:`_engine.Result.scalar` method after invoking the
1371 :meth:`_engine.Connection.execute` method. Parameters are equivalent.
1373 :return: a scalar Python value representing the first column of the
1374 first row returned.
1376 """
1377 distilled_parameters = _distill_params_20(parameters)
1378 try:
1379 meth = statement._execute_on_scalar
1380 except AttributeError as err:
1381 raise exc.ObjectNotExecutableError(statement) from err
1382 else:
1383 return meth(
1384 self,
1385 distilled_parameters,
1386 execution_options or NO_OPTIONS,
1387 )
1389 @overload
1390 def scalars(
1391 self,
1392 statement: TypedReturnsRows[_T],
1393 parameters: Optional[_CoreAnyExecuteParams] = None,
1394 *,
1395 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1396 ) -> ScalarResult[_T]: ...
1398 @overload
1399 def scalars(
1400 self,
1401 statement: Executable,
1402 parameters: Optional[_CoreAnyExecuteParams] = None,
1403 *,
1404 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1405 ) -> ScalarResult[Any]: ...
1407 def scalars(
1408 self,
1409 statement: Executable,
1410 parameters: Optional[_CoreAnyExecuteParams] = None,
1411 *,
1412 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1413 ) -> ScalarResult[Any]:
1414 """Executes and returns a scalar result set, which yields scalar values
1415 from the first column of each row.
1417 This method is equivalent to calling :meth:`_engine.Connection.execute`
1418 to receive a :class:`_result.Result` object, then invoking the
1419 :meth:`_result.Result.scalars` method to produce a
1420 :class:`_result.ScalarResult` instance.
1422 :return: a :class:`_result.ScalarResult`
1424 .. versionadded:: 1.4.24
1426 """
1428 return self.execute(
1429 statement, parameters, execution_options=execution_options
1430 ).scalars()
1432 @overload
1433 def execute(
1434 self,
1435 statement: TypedReturnsRows[Unpack[_Ts]],
1436 parameters: Optional[_CoreAnyExecuteParams] = None,
1437 *,
1438 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1439 ) -> CursorResult[Unpack[_Ts]]: ...
1441 @overload
1442 def execute(
1443 self,
1444 statement: Executable,
1445 parameters: Optional[_CoreAnyExecuteParams] = None,
1446 *,
1447 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1448 ) -> CursorResult[Unpack[TupleAny]]: ...
1450 def execute(
1451 self,
1452 statement: Executable,
1453 parameters: Optional[_CoreAnyExecuteParams] = None,
1454 *,
1455 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1456 ) -> CursorResult[Unpack[TupleAny]]:
1457 r"""Executes a SQL statement construct and returns a
1458 :class:`_engine.CursorResult`.
1460 :param statement: The statement to be executed. This is always
1461 an object that is in both the :class:`_expression.ClauseElement` and
1462 :class:`_expression.Executable` hierarchies, including:
1464 * :class:`_expression.Select`
1465 * :class:`_expression.Insert`, :class:`_expression.Update`,
1466 :class:`_expression.Delete`
1467 * :class:`_expression.TextClause` and
1468 :class:`_expression.TextualSelect`
1469 * :class:`_schema.DDL` and objects which inherit from
1470 :class:`_schema.ExecutableDDLElement`
1472 :param parameters: parameters which will be bound into the statement.
1473 This may be either a dictionary of parameter names to values,
1474 or a mutable sequence (e.g. a list) of dictionaries. When a
1475 list of dictionaries is passed, the underlying statement execution
1476 will make use of the DBAPI ``cursor.executemany()`` method.
1477 When a single dictionary is passed, the DBAPI ``cursor.execute()``
1478 method will be used.
1480 :param execution_options: optional dictionary of execution options,
1481 which will be associated with the statement execution. This
1482 dictionary can provide a subset of the options that are accepted
1483 by :meth:`_engine.Connection.execution_options`.
1485 :return: a :class:`_engine.Result` object.
1487 """
1488 distilled_parameters = _distill_params_20(parameters)
1489 try:
1490 meth = statement._execute_on_connection
1491 except AttributeError as err:
1492 raise exc.ObjectNotExecutableError(statement) from err
1493 else:
1494 return meth(
1495 self,
1496 distilled_parameters,
1497 execution_options or NO_OPTIONS,
1498 )
1500 def _execute_function(
1501 self,
1502 func: FunctionElement[Any],
1503 distilled_parameters: _CoreMultiExecuteParams,
1504 execution_options: CoreExecuteOptionsParameter,
1505 ) -> CursorResult[Unpack[TupleAny]]:
1506 """Execute a sql.FunctionElement object."""
1508 return self._execute_clauseelement(
1509 func.select(), distilled_parameters, execution_options
1510 )
1512 def _execute_default(
1513 self,
1514 default: DefaultGenerator,
1515 distilled_parameters: _CoreMultiExecuteParams,
1516 execution_options: CoreExecuteOptionsParameter,
1517 ) -> Any:
1518 """Execute a schema.ColumnDefault object."""
1520 exec_opts = self._execution_options.merge_with(execution_options)
1522 event_multiparams: Optional[_CoreMultiExecuteParams]
1523 event_params: Optional[_CoreAnyExecuteParams]
1525 # note for event handlers, the "distilled parameters" which is always
1526 # a list of dicts is broken out into separate "multiparams" and
1527 # "params" collections, which allows the handler to distinguish
1528 # between an executemany and execute style set of parameters.
1529 if self._has_events or self.engine._has_events:
1530 (
1531 default,
1532 distilled_parameters,
1533 event_multiparams,
1534 event_params,
1535 ) = self._invoke_before_exec_event(
1536 default, distilled_parameters, exec_opts
1537 )
1538 else:
1539 event_multiparams = event_params = None
1541 try:
1542 conn = self._dbapi_connection
1543 if conn is None:
1544 conn = self._revalidate_connection()
1546 dialect = self.dialect
1547 ctx = dialect.execution_ctx_cls._init_default(
1548 dialect, self, conn, exec_opts
1549 )
1550 except (exc.PendingRollbackError, exc.ResourceClosedError):
1551 raise
1552 except BaseException as e:
1553 self._handle_dbapi_exception(e, None, None, None, None)
1555 ret = ctx._exec_default(None, default, None)
1557 if self._has_events or self.engine._has_events:
1558 self.dispatch.after_execute(
1559 self,
1560 default,
1561 event_multiparams,
1562 event_params,
1563 exec_opts,
1564 ret,
1565 )
1567 return ret
1569 def _execute_ddl(
1570 self,
1571 ddl: ExecutableDDLElement,
1572 distilled_parameters: _CoreMultiExecuteParams,
1573 execution_options: CoreExecuteOptionsParameter,
1574 ) -> CursorResult[Unpack[TupleAny]]:
1575 """Execute a schema.DDL object."""
1577 exec_opts = ddl._execution_options.merge_with(
1578 self._execution_options, execution_options
1579 )
1581 event_multiparams: Optional[_CoreMultiExecuteParams]
1582 event_params: Optional[_CoreSingleExecuteParams]
1584 if self._has_events or self.engine._has_events:
1585 (
1586 ddl,
1587 distilled_parameters,
1588 event_multiparams,
1589 event_params,
1590 ) = self._invoke_before_exec_event(
1591 ddl, distilled_parameters, exec_opts
1592 )
1593 else:
1594 event_multiparams = event_params = None
1596 schema_translate_map = exec_opts.get("schema_translate_map", None)
1598 dialect = self.dialect
1600 compiled = ddl.compile(
1601 dialect=dialect, schema_translate_map=schema_translate_map
1602 )
1603 ret = self._execute_context(
1604 dialect,
1605 dialect.execution_ctx_cls._init_ddl,
1606 compiled,
1607 None,
1608 exec_opts,
1609 compiled,
1610 )
1611 if self._has_events or self.engine._has_events:
1612 self.dispatch.after_execute(
1613 self,
1614 ddl,
1615 event_multiparams,
1616 event_params,
1617 exec_opts,
1618 ret,
1619 )
1620 return ret
1622 def _invoke_before_exec_event(
1623 self,
1624 elem: Any,
1625 distilled_params: _CoreMultiExecuteParams,
1626 execution_options: _ExecuteOptions,
1627 ) -> Tuple[
1628 Any,
1629 _CoreMultiExecuteParams,
1630 _CoreMultiExecuteParams,
1631 _CoreSingleExecuteParams,
1632 ]:
1633 event_multiparams: _CoreMultiExecuteParams
1634 event_params: _CoreSingleExecuteParams
1636 if len(distilled_params) == 1:
1637 event_multiparams, event_params = [], distilled_params[0]
1638 else:
1639 event_multiparams, event_params = distilled_params, {}
1641 for fn in self.dispatch.before_execute:
1642 elem, event_multiparams, event_params = fn(
1643 self,
1644 elem,
1645 event_multiparams,
1646 event_params,
1647 execution_options,
1648 )
1650 if event_multiparams:
1651 distilled_params = list(event_multiparams)
1652 if event_params:
1653 raise exc.InvalidRequestError(
1654 "Event handler can't return non-empty multiparams "
1655 "and params at the same time"
1656 )
1657 elif event_params:
1658 distilled_params = [event_params]
1659 else:
1660 distilled_params = []
1662 return elem, distilled_params, event_multiparams, event_params
1664 def _execute_clauseelement(
1665 self,
1666 elem: Executable,
1667 distilled_parameters: _CoreMultiExecuteParams,
1668 execution_options: CoreExecuteOptionsParameter,
1669 ) -> CursorResult[Unpack[TupleAny]]:
1670 """Execute a sql.ClauseElement object."""
1672 exec_opts = elem._execution_options.merge_with(
1673 self._execution_options, execution_options
1674 )
1676 has_events = self._has_events or self.engine._has_events
1677 if has_events:
1678 (
1679 elem,
1680 distilled_parameters,
1681 event_multiparams,
1682 event_params,
1683 ) = self._invoke_before_exec_event(
1684 elem, distilled_parameters, exec_opts
1685 )
1687 if distilled_parameters:
1688 # ensure we don't retain a link to the view object for keys()
1689 # which links to the values, which we don't want to cache
1690 keys = sorted(distilled_parameters[0])
1691 for_executemany = len(distilled_parameters) > 1
1692 else:
1693 keys = []
1694 for_executemany = False
1696 dialect = self.dialect
1698 schema_translate_map = exec_opts.get("schema_translate_map", None)
1700 compiled_cache: Optional[CompiledCacheType] = exec_opts.get(
1701 "compiled_cache", self.engine._compiled_cache
1702 )
1704 compiled_sql, extracted_params, param_dict, cache_hit = (
1705 elem._compile_w_cache(
1706 dialect=dialect,
1707 compiled_cache=compiled_cache,
1708 column_keys=keys,
1709 for_executemany=for_executemany,
1710 schema_translate_map=schema_translate_map,
1711 linting=self.dialect.compiler_linting | compiler.WARN_LINTING,
1712 )
1713 )
1714 ret = self._execute_context(
1715 dialect,
1716 dialect.execution_ctx_cls._init_compiled,
1717 compiled_sql,
1718 distilled_parameters,
1719 exec_opts,
1720 compiled_sql,
1721 distilled_parameters,
1722 elem,
1723 extracted_params,
1724 cache_hit=cache_hit,
1725 param_dict=param_dict,
1726 )
1727 if has_events:
1728 self.dispatch.after_execute(
1729 self,
1730 elem,
1731 event_multiparams,
1732 event_params,
1733 exec_opts,
1734 ret,
1735 )
1736 return ret
1738 def exec_driver_sql(
1739 self,
1740 statement: str,
1741 parameters: Optional[_DBAPIAnyExecuteParams] = None,
1742 execution_options: Optional[CoreExecuteOptionsParameter] = None,
1743 ) -> CursorResult[Unpack[TupleAny]]:
1744 r"""Executes a string SQL statement on the DBAPI cursor directly,
1745 without any SQL compilation steps.
1747 This can be used to pass any string directly to the
1748 ``cursor.execute()`` method of the DBAPI in use.
1750 :param statement: The statement str to be executed. Bound parameters
1751 must use the underlying DBAPI's paramstyle, such as "qmark",
1752 "pyformat", "format", etc.
1754 :param parameters: represent bound parameter values to be used in the
1755 execution. The format is one of: a dictionary of named parameters,
1756 a tuple of positional parameters, or a list containing either
1757 dictionaries or tuples for multiple-execute support.
1759 :return: a :class:`_engine.CursorResult`.
1761 E.g. multiple dictionaries::
1764 conn.exec_driver_sql(
1765 "INSERT INTO table (id, value) VALUES (%(id)s, %(value)s)",
1766 [{"id": 1, "value": "v1"}, {"id": 2, "value": "v2"}],
1767 )
1769 Single dictionary::
1771 conn.exec_driver_sql(
1772 "INSERT INTO table (id, value) VALUES (%(id)s, %(value)s)",
1773 dict(id=1, value="v1"),
1774 )
1776 Single tuple::
1778 conn.exec_driver_sql(
1779 "INSERT INTO table (id, value) VALUES (?, ?)", (1, "v1")
1780 )
1782 .. note:: The :meth:`_engine.Connection.exec_driver_sql` method does
1783 not participate in the
1784 :meth:`_events.ConnectionEvents.before_execute` and
1785 :meth:`_events.ConnectionEvents.after_execute` events. To
1786 intercept calls to :meth:`_engine.Connection.exec_driver_sql`, use
1787 :meth:`_events.ConnectionEvents.before_cursor_execute` and
1788 :meth:`_events.ConnectionEvents.after_cursor_execute`.
1790 .. seealso::
1792 :pep:`249`
1794 """
1796 distilled_parameters = _distill_raw_params(parameters)
1798 exec_opts = self._execution_options.merge_with(execution_options)
1800 dialect = self.dialect
1801 ret = self._execute_context(
1802 dialect,
1803 dialect.execution_ctx_cls._init_statement,
1804 statement,
1805 None,
1806 exec_opts,
1807 statement,
1808 distilled_parameters,
1809 )
1811 return ret
1813 def _execute_context(
1814 self,
1815 dialect: Dialect,
1816 constructor: Callable[..., ExecutionContext],
1817 statement: Union[str, Compiled],
1818 parameters: Optional[_AnyMultiExecuteParams],
1819 execution_options: _ExecuteOptions,
1820 *args: Any,
1821 **kw: Any,
1822 ) -> CursorResult[Unpack[TupleAny]]:
1823 """Create an :class:`.ExecutionContext` and execute, returning
1824 a :class:`_engine.CursorResult`."""
1826 if execution_options:
1827 yp = execution_options.get("yield_per", None)
1828 if yp:
1829 execution_options = execution_options.union(
1830 {"stream_results": True, "max_row_buffer": yp}
1831 )
1832 try:
1833 conn = self._dbapi_connection
1834 if conn is None:
1835 conn = self._revalidate_connection()
1837 context = constructor(
1838 dialect, self, conn, execution_options, *args, **kw
1839 )
1840 except (exc.PendingRollbackError, exc.ResourceClosedError):
1841 raise
1842 except BaseException as e:
1843 self._handle_dbapi_exception(
1844 e, str(statement), parameters, None, None
1845 )
1847 if (
1848 self._transaction
1849 and not self._transaction.is_active
1850 or (
1851 self._nested_transaction
1852 and not self._nested_transaction.is_active
1853 )
1854 ):
1855 self._invalid_transaction()
1857 elif self._trans_context_manager:
1858 TransactionalContext._trans_ctx_check(self)
1860 if self._transaction is None:
1861 self._autobegin()
1863 context.pre_exec()
1865 if context.execute_style is ExecuteStyle.INSERTMANYVALUES:
1866 return self._exec_insertmany_context(dialect, context)
1867 else:
1868 return self._exec_single_context(
1869 dialect, context, statement, parameters
1870 )
1872 def _exec_single_context(
1873 self,
1874 dialect: Dialect,
1875 context: ExecutionContext,
1876 statement: Union[str, Compiled],
1877 parameters: Optional[_AnyMultiExecuteParams],
1878 ) -> CursorResult[Unpack[TupleAny]]:
1879 """continue the _execute_context() method for a single DBAPI
1880 cursor.execute() or cursor.executemany() call.
1882 """
1883 if dialect.bind_typing is BindTyping.SETINPUTSIZES:
1884 generic_setinputsizes = context._prepare_set_input_sizes()
1886 if generic_setinputsizes:
1887 try:
1888 dialect.do_set_input_sizes(
1889 context.cursor, generic_setinputsizes, context
1890 )
1891 except BaseException as e:
1892 self._handle_dbapi_exception(
1893 e, str(statement), parameters, None, context
1894 )
1896 cursor, str_statement, parameters = (
1897 context.cursor,
1898 context.statement,
1899 context.parameters,
1900 )
1902 effective_parameters: Optional[_AnyExecuteParams]
1904 if not context.executemany:
1905 effective_parameters = parameters[0]
1906 else:
1907 effective_parameters = parameters
1909 evt_handled: bool = False
1910 try:
1911 if self._has_events or self.engine._has_events:
1912 for fn in self.dispatch.before_cursor_execute:
1913 str_statement, effective_parameters = fn(
1914 self,
1915 cursor,
1916 str_statement,
1917 effective_parameters,
1918 context,
1919 context.executemany,
1920 )
1922 if self._echo:
1923 self._log_info(str_statement)
1925 stats = context._get_cache_stats()
1927 self._log_parameters(
1928 effective_parameters, stats, ismulti=context.executemany
1929 )
1931 if context.execute_style is ExecuteStyle.EXECUTEMANY:
1932 effective_parameters = cast(
1933 "_CoreMultiExecuteParams", effective_parameters
1934 )
1935 if self.dialect._has_events:
1936 for fn in self.dialect.dispatch.do_executemany:
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_executemany(
1947 cursor,
1948 str_statement,
1949 effective_parameters,
1950 context,
1951 )
1952 elif not effective_parameters and context.no_parameters:
1953 if self.dialect._has_events:
1954 for fn in self.dialect.dispatch.do_execute_no_params:
1955 if fn(cursor, str_statement, context):
1956 evt_handled = True
1957 break
1958 if not evt_handled:
1959 self.dialect.do_execute_no_params(
1960 cursor, str_statement, context
1961 )
1962 else:
1963 effective_parameters = cast(
1964 "_CoreSingleExecuteParams", effective_parameters
1965 )
1966 if self.dialect._has_events:
1967 for fn in self.dialect.dispatch.do_execute:
1968 if fn(
1969 cursor,
1970 str_statement,
1971 effective_parameters,
1972 context,
1973 ):
1974 evt_handled = True
1975 break
1976 if not evt_handled:
1977 self.dialect.do_execute(
1978 cursor, str_statement, effective_parameters, context
1979 )
1981 if self._has_events or self.engine._has_events:
1982 self.dispatch.after_cursor_execute(
1983 self,
1984 cursor,
1985 str_statement,
1986 effective_parameters,
1987 context,
1988 context.executemany,
1989 )
1991 context.post_exec()
1993 result = context._setup_result_proxy()
1995 except BaseException as e:
1996 self._handle_dbapi_exception(
1997 e, str_statement, effective_parameters, cursor, context
1998 )
2000 return result
2002 def _exec_insertmany_context(
2003 self,
2004 dialect: Dialect,
2005 context: ExecutionContext,
2006 ) -> CursorResult[Unpack[TupleAny]]:
2007 """continue the _execute_context() method for an "insertmanyvalues"
2008 operation, which will invoke DBAPI
2009 cursor.execute() one or more times with individual log and
2010 event hook calls.
2012 """
2014 if dialect.bind_typing is BindTyping.SETINPUTSIZES:
2015 generic_setinputsizes = context._prepare_set_input_sizes()
2016 else:
2017 generic_setinputsizes = None
2019 cursor, str_statement, parameters = (
2020 context.cursor,
2021 context.statement,
2022 context.parameters,
2023 )
2025 effective_parameters = parameters
2027 engine_events = self._has_events or self.engine._has_events
2028 if self.dialect._has_events:
2029 do_execute_dispatch: Iterable[Any] = (
2030 self.dialect.dispatch.do_execute
2031 )
2032 else:
2033 do_execute_dispatch = ()
2035 if self._echo:
2036 stats = context._get_cache_stats() + " (insertmanyvalues)"
2038 preserve_rowcount = context.execution_options.get(
2039 "preserve_rowcount", False
2040 )
2041 rowcount = 0
2043 for imv_batch in dialect._deliver_insertmanyvalues_batches(
2044 self,
2045 cursor,
2046 str_statement,
2047 effective_parameters,
2048 generic_setinputsizes,
2049 context,
2050 ):
2051 if imv_batch.processed_setinputsizes:
2052 try:
2053 dialect.do_set_input_sizes(
2054 context.cursor,
2055 imv_batch.processed_setinputsizes,
2056 context,
2057 )
2058 except BaseException as e:
2059 self._handle_dbapi_exception(
2060 e,
2061 sql_util._long_statement(imv_batch.replaced_statement),
2062 imv_batch.replaced_parameters,
2063 None,
2064 context,
2065 is_sub_exec=True,
2066 )
2068 sub_stmt = imv_batch.replaced_statement
2069 sub_params = imv_batch.replaced_parameters
2071 try:
2072 if engine_events:
2073 for fn in self.dispatch.before_cursor_execute:
2074 sub_stmt, sub_params = fn(
2075 self,
2076 cursor,
2077 sub_stmt,
2078 sub_params,
2079 context,
2080 True,
2081 )
2083 if self._echo:
2084 self._log_info(sql_util._long_statement(sub_stmt))
2086 imv_stats = f""" {imv_batch.batchnum}/{
2087 imv_batch.total_batches
2088 } ({
2089 'ordered'
2090 if imv_batch.rows_sorted else 'unordered'
2091 }{
2092 '; batch not supported'
2093 if imv_batch.is_downgraded
2094 else ''
2095 })"""
2097 if imv_batch.batchnum == 1:
2098 stats += imv_stats
2099 else:
2100 stats = f"insertmanyvalues{imv_stats}"
2102 self._log_parameters(sub_params, stats, ismulti=False)
2104 for fn in do_execute_dispatch:
2105 if fn(
2106 cursor,
2107 sub_stmt,
2108 sub_params,
2109 context,
2110 ):
2111 break
2112 else:
2113 dialect.do_execute(
2114 cursor,
2115 sub_stmt,
2116 sub_params,
2117 context,
2118 )
2120 if engine_events:
2121 self.dispatch.after_cursor_execute(
2122 self,
2123 cursor,
2124 sub_stmt,
2125 sub_params,
2126 context,
2127 context.executemany,
2128 )
2129 except BaseException as e:
2130 self._handle_dbapi_exception(
2131 e,
2132 sql_util._long_statement(sub_stmt),
2133 sub_params,
2134 cursor,
2135 context,
2136 is_sub_exec=True,
2137 )
2139 if preserve_rowcount:
2140 rowcount += imv_batch.current_batch_size
2142 try:
2143 context.post_exec()
2145 if preserve_rowcount:
2146 context._rowcount = rowcount # type: ignore[attr-defined]
2148 result = context._setup_result_proxy()
2150 except BaseException as e:
2151 self._handle_dbapi_exception(
2152 e, str_statement, effective_parameters, cursor, context
2153 )
2155 return result
2157 def _cursor_execute(
2158 self,
2159 cursor: DBAPICursor,
2160 statement: str,
2161 parameters: _DBAPISingleExecuteParams,
2162 context: Optional[ExecutionContext] = None,
2163 ) -> None:
2164 """Execute a statement + params on the given cursor.
2166 Adds appropriate logging and exception handling.
2168 This method is used by DefaultDialect for special-case
2169 executions, such as for sequences and column defaults.
2170 The path of statement execution in the majority of cases
2171 terminates at _execute_context().
2173 """
2174 try:
2175 if self._has_events or self.engine._has_events:
2176 for fn in self.dispatch.before_cursor_execute:
2177 statement, parameters = fn(
2178 self, cursor, statement, parameters, context, False
2179 )
2181 if self._echo:
2182 self._log_info(statement)
2183 self._log_parameters(parameters, "raw sql", ismulti=False)
2185 for fn in (
2186 ()
2187 if not self.dialect._has_events
2188 else self.dialect.dispatch.do_execute
2189 ):
2190 if fn(cursor, statement, parameters, context):
2191 break
2192 else:
2193 self.dialect.do_execute(cursor, statement, parameters, context)
2195 if self._has_events or self.engine._has_events:
2196 self.dispatch.after_cursor_execute(
2197 self, cursor, statement, parameters, context, False
2198 )
2199 except BaseException as e:
2200 self._handle_dbapi_exception(
2201 e, statement, parameters, cursor, context
2202 )
2204 def _safe_close_cursor(self, cursor: DBAPICursor) -> None:
2205 """Close the given cursor, catching exceptions
2206 and turning into log warnings.
2208 """
2209 try:
2210 cursor.close()
2211 except Exception:
2212 # log the error through the connection pool's logger.
2213 self.engine.pool.logger.error(
2214 "Error closing cursor", exc_info=True
2215 )
2217 _reentrant_error = False
2218 _is_disconnect = False
2220 def _handle_dbapi_exception(
2221 self,
2222 e: BaseException,
2223 statement: Optional[str],
2224 parameters: Optional[_AnyExecuteParams],
2225 cursor: Optional[DBAPICursor],
2226 context: Optional[ExecutionContext],
2227 is_sub_exec: bool = False,
2228 ) -> NoReturn:
2229 exc_info = sys.exc_info()
2231 is_exit_exception = util.is_exit_exception(e)
2233 if not self._is_disconnect:
2234 self._is_disconnect = (
2235 isinstance(e, self.dialect.loaded_dbapi.Error)
2236 and not self.closed
2237 and self.dialect.is_disconnect(
2238 e,
2239 self._dbapi_connection if not self.invalidated else None,
2240 cursor,
2241 )
2242 ) or (is_exit_exception and not self.closed)
2244 invalidate_pool_on_disconnect = not is_exit_exception
2246 ismulti: bool = (
2247 not is_sub_exec and context.executemany
2248 if context is not None
2249 else False
2250 )
2251 if self._reentrant_error:
2252 raise exc.DBAPIError.instance(
2253 statement,
2254 parameters,
2255 e,
2256 self.dialect.loaded_dbapi.Error,
2257 hide_parameters=self.engine.hide_parameters,
2258 dialect=self.dialect,
2259 ismulti=ismulti,
2260 ).with_traceback(exc_info[2]) from e
2261 self._reentrant_error = True
2262 try:
2263 # non-DBAPI error - if we already got a context,
2264 # or there's no string statement, don't wrap it
2265 should_wrap = isinstance(e, self.dialect.loaded_dbapi.Error) or (
2266 not isinstance(e, exc.StatementError)
2267 and statement is not None
2268 and context is None
2269 and not is_exit_exception
2270 )
2272 if should_wrap:
2273 sqlalchemy_exception = exc.DBAPIError.instance(
2274 statement,
2275 parameters,
2276 cast(Exception, e),
2277 self.dialect.loaded_dbapi.Error,
2278 hide_parameters=self.engine.hide_parameters,
2279 connection_invalidated=self._is_disconnect,
2280 dialect=self.dialect,
2281 ismulti=ismulti,
2282 )
2283 else:
2284 sqlalchemy_exception = None
2286 newraise = None
2288 if (self.dialect._has_events) and not self._execution_options.get(
2289 "skip_user_error_events", False
2290 ):
2291 ctx = ExceptionContextImpl(
2292 e,
2293 sqlalchemy_exception,
2294 self.engine,
2295 self.dialect,
2296 self,
2297 cursor,
2298 statement,
2299 parameters,
2300 context,
2301 self._is_disconnect,
2302 invalidate_pool_on_disconnect,
2303 False,
2304 )
2306 for fn in self.dialect.dispatch.handle_error:
2307 try:
2308 # handler returns an exception;
2309 # call next handler in a chain
2310 per_fn = fn(ctx)
2311 if per_fn is not None:
2312 ctx.chained_exception = newraise = per_fn
2313 except Exception as _raised:
2314 # handler raises an exception - stop processing
2315 newraise = _raised
2316 break
2318 if self._is_disconnect != ctx.is_disconnect:
2319 self._is_disconnect = ctx.is_disconnect
2320 if sqlalchemy_exception:
2321 sqlalchemy_exception.connection_invalidated = (
2322 ctx.is_disconnect
2323 )
2325 # set up potentially user-defined value for
2326 # invalidate pool.
2327 invalidate_pool_on_disconnect = (
2328 ctx.invalidate_pool_on_disconnect
2329 )
2331 if should_wrap and context:
2332 context.handle_dbapi_exception(e)
2334 if not self._is_disconnect:
2335 if cursor:
2336 self._safe_close_cursor(cursor)
2337 # "autorollback" was mostly relevant in 1.x series.
2338 # It's very unlikely to reach here, as the connection
2339 # does autobegin so when we are here, we are usually
2340 # in an explicit / semi-explicit transaction.
2341 # however we have a test which manufactures this
2342 # scenario in any case using an event handler.
2343 # test/engine/test_execute.py-> test_actual_autorollback
2344 if not self.in_transaction():
2345 self._rollback_impl()
2347 if newraise:
2348 raise newraise.with_traceback(exc_info[2]) from e
2349 elif should_wrap:
2350 assert sqlalchemy_exception is not None
2351 raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
2352 else:
2353 assert exc_info[1] is not None
2354 raise exc_info[1].with_traceback(exc_info[2])
2355 finally:
2356 del self._reentrant_error
2357 if self._is_disconnect:
2358 del self._is_disconnect
2359 if not self.invalidated:
2360 dbapi_conn_wrapper = self._dbapi_connection
2361 assert dbapi_conn_wrapper is not None
2362 if invalidate_pool_on_disconnect:
2363 self.engine.pool._invalidate(dbapi_conn_wrapper, e)
2364 self.invalidate(e)
2366 @classmethod
2367 def _handle_dbapi_exception_noconnection(
2368 cls,
2369 e: BaseException,
2370 dialect: Dialect,
2371 engine: Optional[Engine] = None,
2372 is_disconnect: Optional[bool] = None,
2373 invalidate_pool_on_disconnect: bool = True,
2374 is_pre_ping: bool = False,
2375 ) -> NoReturn:
2376 exc_info = sys.exc_info()
2378 if is_disconnect is None:
2379 is_disconnect = isinstance(
2380 e, dialect.loaded_dbapi.Error
2381 ) and dialect.is_disconnect(e, None, None)
2383 should_wrap = isinstance(e, dialect.loaded_dbapi.Error)
2385 if should_wrap:
2386 sqlalchemy_exception = exc.DBAPIError.instance(
2387 None,
2388 None,
2389 cast(Exception, e),
2390 dialect.loaded_dbapi.Error,
2391 hide_parameters=(
2392 engine.hide_parameters if engine is not None else False
2393 ),
2394 connection_invalidated=is_disconnect,
2395 dialect=dialect,
2396 )
2397 else:
2398 sqlalchemy_exception = None
2400 newraise = None
2402 if dialect._has_events:
2403 ctx = ExceptionContextImpl(
2404 e,
2405 sqlalchemy_exception,
2406 engine,
2407 dialect,
2408 None,
2409 None,
2410 None,
2411 None,
2412 None,
2413 is_disconnect,
2414 invalidate_pool_on_disconnect,
2415 is_pre_ping,
2416 )
2417 for fn in dialect.dispatch.handle_error:
2418 try:
2419 # handler returns an exception;
2420 # call next handler in a chain
2421 per_fn = fn(ctx)
2422 if per_fn is not None:
2423 ctx.chained_exception = newraise = per_fn
2424 except Exception as _raised:
2425 # handler raises an exception - stop processing
2426 newraise = _raised
2427 break
2429 if sqlalchemy_exception and is_disconnect != ctx.is_disconnect:
2430 sqlalchemy_exception.connection_invalidated = ctx.is_disconnect
2432 if newraise:
2433 raise newraise.with_traceback(exc_info[2]) from e
2434 elif should_wrap:
2435 assert sqlalchemy_exception is not None
2436 raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
2437 else:
2438 assert exc_info[1] is not None
2439 raise exc_info[1].with_traceback(exc_info[2])
2441 def _run_ddl_visitor(
2442 self,
2443 visitorcallable: Type[InvokeDDLBase],
2444 element: SchemaVisitable,
2445 **kwargs: Any,
2446 ) -> None:
2447 """run a DDL visitor.
2449 This method is only here so that the MockConnection can change the
2450 options given to the visitor so that "checkfirst" is skipped.
2452 """
2453 visitorcallable(
2454 dialect=self.dialect, connection=self, **kwargs
2455 ).traverse_single(element)
2458class ExceptionContextImpl(ExceptionContext):
2459 """Implement the :class:`.ExceptionContext` interface."""
2461 __slots__ = (
2462 "connection",
2463 "engine",
2464 "dialect",
2465 "cursor",
2466 "statement",
2467 "parameters",
2468 "original_exception",
2469 "sqlalchemy_exception",
2470 "chained_exception",
2471 "execution_context",
2472 "is_disconnect",
2473 "invalidate_pool_on_disconnect",
2474 "is_pre_ping",
2475 )
2477 def __init__(
2478 self,
2479 exception: BaseException,
2480 sqlalchemy_exception: Optional[exc.StatementError],
2481 engine: Optional[Engine],
2482 dialect: Dialect,
2483 connection: Optional[Connection],
2484 cursor: Optional[DBAPICursor],
2485 statement: Optional[str],
2486 parameters: Optional[_DBAPIAnyExecuteParams],
2487 context: Optional[ExecutionContext],
2488 is_disconnect: bool,
2489 invalidate_pool_on_disconnect: bool,
2490 is_pre_ping: bool,
2491 ):
2492 self.engine = engine
2493 self.dialect = dialect
2494 self.connection = connection
2495 self.sqlalchemy_exception = sqlalchemy_exception
2496 self.original_exception = exception
2497 self.execution_context = context
2498 self.statement = statement
2499 self.parameters = parameters
2500 self.is_disconnect = is_disconnect
2501 self.invalidate_pool_on_disconnect = invalidate_pool_on_disconnect
2502 self.is_pre_ping = is_pre_ping
2505class Transaction(TransactionalContext):
2506 """Represent a database transaction in progress.
2508 The :class:`.Transaction` object is procured by
2509 calling the :meth:`_engine.Connection.begin` method of
2510 :class:`_engine.Connection`::
2512 from sqlalchemy import create_engine
2514 engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
2515 connection = engine.connect()
2516 trans = connection.begin()
2517 connection.execute(text("insert into x (a, b) values (1, 2)"))
2518 trans.commit()
2520 The object provides :meth:`.rollback` and :meth:`.commit`
2521 methods in order to control transaction boundaries. It
2522 also implements a context manager interface so that
2523 the Python ``with`` statement can be used with the
2524 :meth:`_engine.Connection.begin` method::
2526 with connection.begin():
2527 connection.execute(text("insert into x (a, b) values (1, 2)"))
2529 The Transaction object is **not** threadsafe.
2531 .. seealso::
2533 :meth:`_engine.Connection.begin`
2535 :meth:`_engine.Connection.begin_twophase`
2537 :meth:`_engine.Connection.begin_nested`
2539 .. index::
2540 single: thread safety; Transaction
2541 """ # noqa
2543 __slots__ = ()
2545 _is_root: bool = False
2546 is_active: bool
2547 connection: Connection
2549 def __init__(self, connection: Connection):
2550 raise NotImplementedError()
2552 @property
2553 def _deactivated_from_connection(self) -> bool:
2554 """True if this transaction is totally deactivated from the connection
2555 and therefore can no longer affect its state.
2557 """
2558 raise NotImplementedError()
2560 def _do_close(self) -> None:
2561 raise NotImplementedError()
2563 def _do_rollback(self) -> None:
2564 raise NotImplementedError()
2566 def _do_commit(self) -> None:
2567 raise NotImplementedError()
2569 @property
2570 def is_valid(self) -> bool:
2571 return self.is_active and not self.connection.invalidated
2573 def close(self) -> None:
2574 """Close this :class:`.Transaction`.
2576 If this transaction is the base transaction in a begin/commit
2577 nesting, the transaction will rollback(). Otherwise, the
2578 method returns.
2580 This is used to cancel a Transaction without affecting the scope of
2581 an enclosing transaction.
2583 """
2584 try:
2585 self._do_close()
2586 finally:
2587 assert not self.is_active
2589 def rollback(self) -> None:
2590 """Roll back this :class:`.Transaction`.
2592 The implementation of this may vary based on the type of transaction in
2593 use:
2595 * For a simple database transaction (e.g. :class:`.RootTransaction`),
2596 it corresponds to a ROLLBACK.
2598 * For a :class:`.NestedTransaction`, it corresponds to a
2599 "ROLLBACK TO SAVEPOINT" operation.
2601 * For a :class:`.TwoPhaseTransaction`, DBAPI-specific methods for two
2602 phase transactions may be used.
2605 """
2606 try:
2607 self._do_rollback()
2608 finally:
2609 assert not self.is_active
2611 def commit(self) -> None:
2612 """Commit this :class:`.Transaction`.
2614 The implementation of this may vary based on the type of transaction in
2615 use:
2617 * For a simple database transaction (e.g. :class:`.RootTransaction`),
2618 it corresponds to a COMMIT.
2620 * For a :class:`.NestedTransaction`, it corresponds to a
2621 "RELEASE SAVEPOINT" operation.
2623 * For a :class:`.TwoPhaseTransaction`, DBAPI-specific methods for two
2624 phase transactions may be used.
2626 """
2627 try:
2628 self._do_commit()
2629 finally:
2630 assert not self.is_active
2632 def _get_subject(self) -> Connection:
2633 return self.connection
2635 def _transaction_is_active(self) -> bool:
2636 return self.is_active
2638 def _transaction_is_closed(self) -> bool:
2639 return not self._deactivated_from_connection
2641 def _rollback_can_be_called(self) -> bool:
2642 # for RootTransaction / NestedTransaction, it's safe to call
2643 # rollback() even if the transaction is deactive and no warnings
2644 # will be emitted. tested in
2645 # test_transaction.py -> test_no_rollback_in_deactive(?:_savepoint)?
2646 return True
2649class RootTransaction(Transaction):
2650 """Represent the "root" transaction on a :class:`_engine.Connection`.
2652 This corresponds to the current "BEGIN/COMMIT/ROLLBACK" that's occurring
2653 for the :class:`_engine.Connection`. The :class:`_engine.RootTransaction`
2654 is created by calling upon the :meth:`_engine.Connection.begin` method, and
2655 remains associated with the :class:`_engine.Connection` throughout its
2656 active span. The current :class:`_engine.RootTransaction` in use is
2657 accessible via the :attr:`_engine.Connection.get_transaction` method of
2658 :class:`_engine.Connection`.
2660 In :term:`2.0 style` use, the :class:`_engine.Connection` also employs
2661 "autobegin" behavior that will create a new
2662 :class:`_engine.RootTransaction` whenever a connection in a
2663 non-transactional state is used to emit commands on the DBAPI connection.
2664 The scope of the :class:`_engine.RootTransaction` in 2.0 style
2665 use can be controlled using the :meth:`_engine.Connection.commit` and
2666 :meth:`_engine.Connection.rollback` methods.
2669 """
2671 _is_root = True
2673 __slots__ = ("connection", "is_active")
2675 def __init__(self, connection: Connection):
2676 assert connection._transaction is None
2677 if connection._trans_context_manager:
2678 TransactionalContext._trans_ctx_check(connection)
2679 self.connection = connection
2680 self._connection_begin_impl()
2681 connection._transaction = self
2683 self.is_active = True
2685 def _deactivate_from_connection(self) -> None:
2686 if self.is_active:
2687 assert self.connection._transaction is self
2688 self.is_active = False
2690 elif self.connection._transaction is not self:
2691 util.warn("transaction already deassociated from connection")
2693 @property
2694 def _deactivated_from_connection(self) -> bool:
2695 return self.connection._transaction is not self
2697 def _connection_begin_impl(self) -> None:
2698 self.connection._begin_impl(self)
2700 def _connection_rollback_impl(self) -> None:
2701 self.connection._rollback_impl()
2703 def _connection_commit_impl(self) -> None:
2704 self.connection._commit_impl()
2706 def _close_impl(self, try_deactivate: bool = False) -> None:
2707 try:
2708 if self.is_active:
2709 self._connection_rollback_impl()
2711 if self.connection._nested_transaction:
2712 self.connection._nested_transaction._cancel()
2713 finally:
2714 if self.is_active or try_deactivate:
2715 self._deactivate_from_connection()
2716 if self.connection._transaction is self:
2717 self.connection._transaction = None
2719 assert not self.is_active
2720 assert self.connection._transaction is not self
2722 def _do_close(self) -> None:
2723 self._close_impl()
2725 def _do_rollback(self) -> None:
2726 self._close_impl(try_deactivate=True)
2728 def _do_commit(self) -> None:
2729 if self.is_active:
2730 assert self.connection._transaction is self
2732 try:
2733 self._connection_commit_impl()
2734 finally:
2735 # whether or not commit succeeds, cancel any
2736 # nested transactions, make this transaction "inactive"
2737 # and remove it as a reset agent
2738 if self.connection._nested_transaction:
2739 self.connection._nested_transaction._cancel()
2741 self._deactivate_from_connection()
2743 # ...however only remove as the connection's current transaction
2744 # if commit succeeded. otherwise it stays on so that a rollback
2745 # needs to occur.
2746 self.connection._transaction = None
2747 else:
2748 if self.connection._transaction is self:
2749 self.connection._invalid_transaction()
2750 else:
2751 raise exc.InvalidRequestError("This transaction is inactive")
2753 assert not self.is_active
2754 assert self.connection._transaction is not self
2757class NestedTransaction(Transaction):
2758 """Represent a 'nested', or SAVEPOINT transaction.
2760 The :class:`.NestedTransaction` object is created by calling the
2761 :meth:`_engine.Connection.begin_nested` method of
2762 :class:`_engine.Connection`.
2764 When using :class:`.NestedTransaction`, the semantics of "begin" /
2765 "commit" / "rollback" are as follows:
2767 * the "begin" operation corresponds to the "BEGIN SAVEPOINT" command, where
2768 the savepoint is given an explicit name that is part of the state
2769 of this object.
2771 * The :meth:`.NestedTransaction.commit` method corresponds to a
2772 "RELEASE SAVEPOINT" operation, using the savepoint identifier associated
2773 with this :class:`.NestedTransaction`.
2775 * The :meth:`.NestedTransaction.rollback` method corresponds to a
2776 "ROLLBACK TO SAVEPOINT" operation, using the savepoint identifier
2777 associated with this :class:`.NestedTransaction`.
2779 The rationale for mimicking the semantics of an outer transaction in
2780 terms of savepoints so that code may deal with a "savepoint" transaction
2781 and an "outer" transaction in an agnostic way.
2783 .. seealso::
2785 :ref:`session_begin_nested` - ORM version of the SAVEPOINT API.
2787 """
2789 __slots__ = ("connection", "is_active", "_savepoint", "_previous_nested")
2791 _savepoint: str
2793 def __init__(self, connection: Connection):
2794 assert connection._transaction is not None
2795 if connection._trans_context_manager:
2796 TransactionalContext._trans_ctx_check(connection)
2797 self.connection = connection
2798 self._savepoint = self.connection._savepoint_impl()
2799 self.is_active = True
2800 self._previous_nested = connection._nested_transaction
2801 connection._nested_transaction = self
2803 def _deactivate_from_connection(self, warn: bool = True) -> None:
2804 if self.connection._nested_transaction is self:
2805 self.connection._nested_transaction = self._previous_nested
2806 elif warn:
2807 util.warn(
2808 "nested transaction already deassociated from connection"
2809 )
2811 @property
2812 def _deactivated_from_connection(self) -> bool:
2813 return self.connection._nested_transaction is not self
2815 def _cancel(self) -> None:
2816 # called by RootTransaction when the outer transaction is
2817 # committed, rolled back, or closed to cancel all savepoints
2818 # without any action being taken
2819 self.is_active = False
2820 self._deactivate_from_connection()
2821 if self._previous_nested:
2822 self._previous_nested._cancel()
2824 def _close_impl(
2825 self, deactivate_from_connection: bool, warn_already_deactive: bool
2826 ) -> None:
2827 try:
2828 if (
2829 self.is_active
2830 and self.connection._transaction
2831 and self.connection._transaction.is_active
2832 ):
2833 self.connection._rollback_to_savepoint_impl(self._savepoint)
2834 finally:
2835 self.is_active = False
2837 if deactivate_from_connection:
2838 self._deactivate_from_connection(warn=warn_already_deactive)
2840 assert not self.is_active
2841 if deactivate_from_connection:
2842 assert self.connection._nested_transaction is not self
2844 def _do_close(self) -> None:
2845 self._close_impl(True, False)
2847 def _do_rollback(self) -> None:
2848 self._close_impl(True, True)
2850 def _do_commit(self) -> None:
2851 if self.is_active:
2852 try:
2853 self.connection._release_savepoint_impl(self._savepoint)
2854 finally:
2855 # nested trans becomes inactive on failed release
2856 # unconditionally. this prevents it from trying to
2857 # emit SQL when it rolls back.
2858 self.is_active = False
2860 # but only de-associate from connection if it succeeded
2861 self._deactivate_from_connection()
2862 else:
2863 if self.connection._nested_transaction is self:
2864 self.connection._invalid_transaction()
2865 else:
2866 raise exc.InvalidRequestError(
2867 "This nested transaction is inactive"
2868 )
2871class TwoPhaseTransaction(RootTransaction):
2872 """Represent a two-phase transaction.
2874 A new :class:`.TwoPhaseTransaction` object may be procured
2875 using the :meth:`_engine.Connection.begin_twophase` method.
2877 The interface is the same as that of :class:`.Transaction`
2878 with the addition of the :meth:`prepare` method.
2880 """
2882 __slots__ = ("xid", "_is_prepared")
2884 xid: Any
2886 def __init__(self, connection: Connection, xid: Any):
2887 self._is_prepared = False
2888 self.xid = xid
2889 super().__init__(connection)
2891 def prepare(self) -> None:
2892 """Prepare this :class:`.TwoPhaseTransaction`.
2894 After a PREPARE, the transaction can be committed.
2896 """
2897 if not self.is_active:
2898 raise exc.InvalidRequestError("This transaction is inactive")
2899 self.connection._prepare_twophase_impl(self.xid)
2900 self._is_prepared = True
2902 def _connection_begin_impl(self) -> None:
2903 self.connection._begin_twophase_impl(self)
2905 def _connection_rollback_impl(self) -> None:
2906 self.connection._rollback_twophase_impl(self.xid, self._is_prepared)
2908 def _connection_commit_impl(self) -> None:
2909 self.connection._commit_twophase_impl(self.xid, self._is_prepared)
2912class Engine(
2913 ConnectionEventsTarget, log.Identified, inspection.Inspectable["Inspector"]
2914):
2915 """
2916 Connects a :class:`~sqlalchemy.pool.Pool` and
2917 :class:`~sqlalchemy.engine.interfaces.Dialect` together to provide a
2918 source of database connectivity and behavior.
2920 An :class:`_engine.Engine` object is instantiated publicly using the
2921 :func:`~sqlalchemy.create_engine` function.
2923 .. seealso::
2925 :doc:`/core/engines`
2927 :ref:`connections_toplevel`
2929 """
2931 dispatch: dispatcher[ConnectionEventsTarget]
2933 _compiled_cache: Optional[CompiledCacheType]
2935 _execution_options: _ExecuteOptions = _EMPTY_EXECUTION_OPTS
2936 _has_events: bool = False
2937 _connection_cls: Type[Connection] = Connection
2938 _sqla_logger_namespace: str = "sqlalchemy.engine.Engine"
2939 _is_future: bool = False
2941 _schema_translate_map: Optional[SchemaTranslateMapType] = None
2942 _option_cls: Type[OptionEngine]
2944 dialect: Dialect
2945 pool: Pool
2946 url: URL
2947 hide_parameters: bool
2949 def __init__(
2950 self,
2951 pool: Pool,
2952 dialect: Dialect,
2953 url: URL,
2954 logging_name: Optional[str] = None,
2955 echo: Optional[_EchoFlagType] = None,
2956 query_cache_size: int = 500,
2957 execution_options: Optional[Mapping[str, Any]] = None,
2958 hide_parameters: bool = False,
2959 ):
2960 self.pool = pool
2961 self.url = url
2962 self.dialect = dialect
2963 if logging_name:
2964 self.logging_name = logging_name
2965 self.echo = echo
2966 self.hide_parameters = hide_parameters
2967 if query_cache_size != 0:
2968 self._compiled_cache = util.LRUCache(
2969 query_cache_size, size_alert=self._lru_size_alert
2970 )
2971 else:
2972 self._compiled_cache = None
2973 log.instance_logger(self, echoflag=echo)
2974 if execution_options:
2975 self.update_execution_options(**execution_options)
2977 def _lru_size_alert(self, cache: util.LRUCache[Any, Any]) -> None:
2978 if self._should_log_info():
2979 self.logger.info(
2980 "Compiled cache size pruning from %d items to %d. "
2981 "Increase cache size to reduce the frequency of pruning.",
2982 len(cache),
2983 cache.capacity,
2984 )
2986 @property
2987 def engine(self) -> Engine:
2988 """Returns this :class:`.Engine`.
2990 Used for legacy schemes that accept :class:`.Connection` /
2991 :class:`.Engine` objects within the same variable.
2993 """
2994 return self
2996 def clear_compiled_cache(self) -> None:
2997 """Clear the compiled cache associated with the dialect.
2999 This applies **only** to the built-in cache that is established
3000 via the :paramref:`_engine.create_engine.query_cache_size` parameter.
3001 It will not impact any dictionary caches that were passed via the
3002 :paramref:`.Connection.execution_options.compiled_cache` parameter.
3004 .. versionadded:: 1.4
3006 """
3007 if self._compiled_cache:
3008 self._compiled_cache.clear()
3010 def update_execution_options(self, **opt: Any) -> None:
3011 r"""Update the default execution_options dictionary
3012 of this :class:`_engine.Engine`.
3014 The given keys/values in \**opt are added to the
3015 default execution options that will be used for
3016 all connections. The initial contents of this dictionary
3017 can be sent via the ``execution_options`` parameter
3018 to :func:`_sa.create_engine`.
3020 .. seealso::
3022 :meth:`_engine.Connection.execution_options`
3024 :meth:`_engine.Engine.execution_options`
3026 """
3027 self.dispatch.set_engine_execution_options(self, opt)
3028 self._execution_options = self._execution_options.union(opt)
3029 self.dialect.set_engine_execution_options(self, opt)
3031 @overload
3032 def execution_options(
3033 self,
3034 *,
3035 compiled_cache: Optional[CompiledCacheType] = ...,
3036 logging_token: str = ...,
3037 isolation_level: IsolationLevel = ...,
3038 insertmanyvalues_page_size: int = ...,
3039 schema_translate_map: Optional[SchemaTranslateMapType] = ...,
3040 **opt: Any,
3041 ) -> OptionEngine: ...
3043 @overload
3044 def execution_options(self, **opt: Any) -> OptionEngine: ...
3046 def execution_options(self, **opt: Any) -> OptionEngine:
3047 """Return a new :class:`_engine.Engine` that will provide
3048 :class:`_engine.Connection` objects with the given execution options.
3050 The returned :class:`_engine.Engine` remains related to the original
3051 :class:`_engine.Engine` in that it shares the same connection pool and
3052 other state:
3054 * The :class:`_pool.Pool` used by the new :class:`_engine.Engine`
3055 is the
3056 same instance. The :meth:`_engine.Engine.dispose`
3057 method will replace
3058 the connection pool instance for the parent engine as well
3059 as this one.
3060 * Event listeners are "cascaded" - meaning, the new
3061 :class:`_engine.Engine`
3062 inherits the events of the parent, and new events can be associated
3063 with the new :class:`_engine.Engine` individually.
3064 * The logging configuration and logging_name is copied from the parent
3065 :class:`_engine.Engine`.
3067 The intent of the :meth:`_engine.Engine.execution_options` method is
3068 to implement schemes where multiple :class:`_engine.Engine`
3069 objects refer to the same connection pool, but are differentiated
3070 by options that affect some execution-level behavior for each
3071 engine. One such example is breaking into separate "reader" and
3072 "writer" :class:`_engine.Engine` instances, where one
3073 :class:`_engine.Engine`
3074 has a lower :term:`isolation level` setting configured or is even
3075 transaction-disabled using "autocommit". An example of this
3076 configuration is at :ref:`dbapi_autocommit_multiple`.
3078 Another example is one that
3079 uses a custom option ``shard_id`` which is consumed by an event
3080 to change the current schema on a database connection::
3082 from sqlalchemy import event
3083 from sqlalchemy.engine import Engine
3085 primary_engine = create_engine("mysql+mysqldb://")
3086 shard1 = primary_engine.execution_options(shard_id="shard1")
3087 shard2 = primary_engine.execution_options(shard_id="shard2")
3089 shards = {"default": "base", "shard_1": "db1", "shard_2": "db2"}
3092 @event.listens_for(Engine, "before_cursor_execute")
3093 def _switch_shard(conn, cursor, stmt, params, context, executemany):
3094 shard_id = conn.get_execution_options().get("shard_id", "default")
3095 current_shard = conn.info.get("current_shard", None)
3097 if current_shard != shard_id:
3098 cursor.execute("use %s" % shards[shard_id])
3099 conn.info["current_shard"] = shard_id
3101 The above recipe illustrates two :class:`_engine.Engine` objects that
3102 will each serve as factories for :class:`_engine.Connection` objects
3103 that have pre-established "shard_id" execution options present. A
3104 :meth:`_events.ConnectionEvents.before_cursor_execute` event handler
3105 then interprets this execution option to emit a MySQL ``use`` statement
3106 to switch databases before a statement execution, while at the same
3107 time keeping track of which database we've established using the
3108 :attr:`_engine.Connection.info` dictionary.
3110 .. seealso::
3112 :meth:`_engine.Connection.execution_options`
3113 - update execution options
3114 on a :class:`_engine.Connection` object.
3116 :meth:`_engine.Engine.update_execution_options`
3117 - update the execution
3118 options for a given :class:`_engine.Engine` in place.
3120 :meth:`_engine.Engine.get_execution_options`
3123 """ # noqa: E501
3124 return self._option_cls(self, opt)
3126 def get_execution_options(self) -> _ExecuteOptions:
3127 """Get the non-SQL options which will take effect during execution.
3129 .. seealso::
3131 :meth:`_engine.Engine.execution_options`
3132 """
3133 return self._execution_options
3135 @property
3136 def name(self) -> str:
3137 """String name of the :class:`~sqlalchemy.engine.interfaces.Dialect`
3138 in use by this :class:`Engine`.
3140 """
3142 return self.dialect.name
3144 @property
3145 def driver(self) -> str:
3146 """Driver name of the :class:`~sqlalchemy.engine.interfaces.Dialect`
3147 in use by this :class:`Engine`.
3149 """
3151 return self.dialect.driver
3153 echo = log.echo_property()
3155 def __repr__(self) -> str:
3156 return "Engine(%r)" % (self.url,)
3158 def dispose(self, close: bool = True) -> None:
3159 """Dispose of the connection pool used by this
3160 :class:`_engine.Engine`.
3162 A new connection pool is created immediately after the old one has been
3163 disposed. The previous connection pool is disposed either actively, by
3164 closing out all currently checked-in connections in that pool, or
3165 passively, by losing references to it but otherwise not closing any
3166 connections. The latter strategy is more appropriate for an initializer
3167 in a forked Python process.
3169 Event listeners associated with the old pool via :class:`.PoolEvents`
3170 are **transferred to the new pool**; this is to support the pattern
3171 by which :class:`.PoolEvents` are set up in terms of the owning
3172 :class:`.Engine` without the need to refer to the :class:`.Pool`
3173 directly.
3175 :param close: if left at its default of ``True``, has the
3176 effect of fully closing all **currently checked in**
3177 database connections. Connections that are still checked out
3178 will **not** be closed, however they will no longer be associated
3179 with this :class:`_engine.Engine`,
3180 so when they are closed individually, eventually the
3181 :class:`_pool.Pool` which they are associated with will
3182 be garbage collected and they will be closed out fully, if
3183 not already closed on checkin.
3185 If set to ``False``, the previous connection pool is de-referenced,
3186 and otherwise not touched in any way.
3188 .. versionadded:: 1.4.33 Added the :paramref:`.Engine.dispose.close`
3189 parameter to allow the replacement of a connection pool in a child
3190 process without interfering with the connections used by the parent
3191 process.
3194 .. seealso::
3196 :ref:`engine_disposal`
3198 :ref:`pooling_multiprocessing`
3200 :meth:`.ConnectionEvents.engine_disposed`
3202 """
3203 if close:
3204 self.pool.dispose()
3205 self.pool = self.pool.recreate()
3206 self.dispatch.engine_disposed(self)
3208 @contextlib.contextmanager
3209 def _optional_conn_ctx_manager(
3210 self, connection: Optional[Connection] = None
3211 ) -> Iterator[Connection]:
3212 if connection is None:
3213 with self.connect() as conn:
3214 yield conn
3215 else:
3216 yield connection
3218 @contextlib.contextmanager
3219 def begin(self) -> Iterator[Connection]:
3220 """Return a context manager delivering a :class:`_engine.Connection`
3221 with a :class:`.Transaction` established.
3223 E.g.::
3225 with engine.begin() as conn:
3226 conn.execute(text("insert into table (x, y, z) values (1, 2, 3)"))
3227 conn.execute(text("my_special_procedure(5)"))
3229 Upon successful operation, the :class:`.Transaction`
3230 is committed. If an error is raised, the :class:`.Transaction`
3231 is rolled back.
3233 .. seealso::
3235 :meth:`_engine.Engine.connect` - procure a
3236 :class:`_engine.Connection` from
3237 an :class:`_engine.Engine`.
3239 :meth:`_engine.Connection.begin` - start a :class:`.Transaction`
3240 for a particular :class:`_engine.Connection`.
3242 """ # noqa: E501
3243 with self.connect() as conn:
3244 with conn.begin():
3245 yield conn
3247 def _run_ddl_visitor(
3248 self,
3249 visitorcallable: Type[InvokeDDLBase],
3250 element: SchemaVisitable,
3251 **kwargs: Any,
3252 ) -> None:
3253 with self.begin() as conn:
3254 conn._run_ddl_visitor(visitorcallable, element, **kwargs)
3256 def connect(self) -> Connection:
3257 """Return a new :class:`_engine.Connection` object.
3259 The :class:`_engine.Connection` acts as a Python context manager, so
3260 the typical use of this method looks like::
3262 with engine.connect() as connection:
3263 connection.execute(text("insert into table values ('foo')"))
3264 connection.commit()
3266 Where above, after the block is completed, the connection is "closed"
3267 and its underlying DBAPI resources are returned to the connection pool.
3268 This also has the effect of rolling back any transaction that
3269 was explicitly begun or was begun via autobegin, and will
3270 emit the :meth:`_events.ConnectionEvents.rollback` event if one was
3271 started and is still in progress.
3273 .. seealso::
3275 :meth:`_engine.Engine.begin`
3277 """
3279 return self._connection_cls(self)
3281 def raw_connection(self) -> PoolProxiedConnection:
3282 """Return a "raw" DBAPI connection from the connection pool.
3284 The returned object is a proxied version of the DBAPI
3285 connection object used by the underlying driver in use.
3286 The object will have all the same behavior as the real DBAPI
3287 connection, except that its ``close()`` method will result in the
3288 connection being returned to the pool, rather than being closed
3289 for real.
3291 This method provides direct DBAPI connection access for
3292 special situations when the API provided by
3293 :class:`_engine.Connection`
3294 is not needed. When a :class:`_engine.Connection` object is already
3295 present, the DBAPI connection is available using
3296 the :attr:`_engine.Connection.connection` accessor.
3298 .. seealso::
3300 :ref:`dbapi_connections`
3302 """
3303 return self.pool.connect()
3306class OptionEngineMixin(log.Identified):
3307 _sa_propagate_class_events = False
3309 dispatch: dispatcher[ConnectionEventsTarget]
3310 _compiled_cache: Optional[CompiledCacheType]
3311 dialect: Dialect
3312 pool: Pool
3313 url: URL
3314 hide_parameters: bool
3315 echo: log.echo_property
3317 def __init__(
3318 self, proxied: Engine, execution_options: CoreExecuteOptionsParameter
3319 ):
3320 self._proxied = proxied
3321 self.url = proxied.url
3322 self.dialect = proxied.dialect
3323 self.logging_name = proxied.logging_name
3324 self.echo = proxied.echo
3325 self._compiled_cache = proxied._compiled_cache
3326 self.hide_parameters = proxied.hide_parameters
3327 log.instance_logger(self, echoflag=self.echo)
3329 # note: this will propagate events that are assigned to the parent
3330 # engine after this OptionEngine is created. Since we share
3331 # the events of the parent we also disallow class-level events
3332 # to apply to the OptionEngine class directly.
3333 #
3334 # the other way this can work would be to transfer existing
3335 # events only, using:
3336 # self.dispatch._update(proxied.dispatch)
3337 #
3338 # that might be more appropriate however it would be a behavioral
3339 # change for logic that assigns events to the parent engine and
3340 # would like it to take effect for the already-created sub-engine.
3341 self.dispatch = self.dispatch._join(proxied.dispatch)
3343 self._execution_options = proxied._execution_options
3344 self.update_execution_options(**execution_options)
3346 def update_execution_options(self, **opt: Any) -> None:
3347 raise NotImplementedError()
3349 if not typing.TYPE_CHECKING:
3350 # https://github.com/python/typing/discussions/1095
3352 @property
3353 def pool(self) -> Pool:
3354 return self._proxied.pool
3356 @pool.setter
3357 def pool(self, pool: Pool) -> None:
3358 self._proxied.pool = pool
3360 @property
3361 def _has_events(self) -> bool:
3362 return self._proxied._has_events or self.__dict__.get(
3363 "_has_events", False
3364 )
3366 @_has_events.setter
3367 def _has_events(self, value: bool) -> None:
3368 self.__dict__["_has_events"] = value
3371class OptionEngine(OptionEngineMixin, Engine):
3372 def update_execution_options(self, **opt: Any) -> None:
3373 Engine.update_execution_options(self, **opt)
3376Engine._option_cls = OptionEngine