Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py: 24%
1027 statements
« prev ^ index » next coverage.py v7.2.7, created at 2023-06-07 06:35 +0000
1# engine/base.py
2# Copyright (C) 2005-2023 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
7from __future__ import with_statement
9import contextlib
10import sys
12from .interfaces import Connectable
13from .interfaces import ExceptionContext
14from .util import _distill_params
15from .util import _distill_params_20
16from .util import TransactionalContext
17from .. import exc
18from .. import inspection
19from .. import log
20from .. import util
21from ..sql import compiler
22from ..sql import util as sql_util
25"""Defines :class:`_engine.Connection` and :class:`_engine.Engine`.
27"""
29_EMPTY_EXECUTION_OPTS = util.immutabledict()
32class Connection(Connectable):
33 """Provides high-level functionality for a wrapped DB-API connection.
35 **This is the SQLAlchemy 1.x.x version** of the :class:`_engine.Connection`
36 class. For the :term:`2.0 style` version, which features some API
37 differences, see :class:`_future.Connection`.
39 The :class:`_engine.Connection` object is procured by calling
40 the :meth:`_engine.Engine.connect` method of the :class:`_engine.Engine`
41 object, and provides services for execution of SQL statements as well
42 as transaction control.
44 The Connection object is **not** thread-safe. While a Connection can be
45 shared among threads using properly synchronized access, it is still
46 possible that the underlying DBAPI connection may not support shared
47 access between threads. Check the DBAPI documentation for details.
49 The Connection object represents a single DBAPI connection checked out
50 from the connection pool. In this state, the connection pool has no affect
51 upon the connection, including its expiration or timeout state. For the
52 connection pool to properly manage connections, connections should be
53 returned to the connection pool (i.e. ``connection.close()``) whenever the
54 connection is not in use.
56 .. index::
57 single: thread safety; Connection
59 """
61 _is_future = False
62 _sqla_logger_namespace = "sqlalchemy.engine.Connection"
64 # used by sqlalchemy.engine.util.TransactionalContext
65 _trans_context_manager = None
67 def __init__(
68 self,
69 engine,
70 connection=None,
71 close_with_result=False,
72 _branch_from=None,
73 _execution_options=None,
74 _dispatch=None,
75 _has_events=None,
76 _allow_revalidate=True,
77 ):
78 """Construct a new Connection."""
79 self.engine = engine
80 self.dialect = engine.dialect
81 self.__branch_from = _branch_from
83 if _branch_from:
84 # branching is always "from" the root connection
85 assert _branch_from.__branch_from is None
86 self._dbapi_connection = connection
87 self._execution_options = _execution_options
88 self._echo = _branch_from._echo
89 self.should_close_with_result = False
90 self.dispatch = _dispatch
91 self._has_events = _branch_from._has_events
92 else:
93 self._dbapi_connection = (
94 connection
95 if connection is not None
96 else engine.raw_connection()
97 )
99 self._transaction = self._nested_transaction = None
100 self.__savepoint_seq = 0
101 self.__in_begin = False
102 self.should_close_with_result = close_with_result
104 self.__can_reconnect = _allow_revalidate
105 self._echo = self.engine._should_log_info()
107 if _has_events is None:
108 # if _has_events is sent explicitly as False,
109 # then don't join the dispatch of the engine; we don't
110 # want to handle any of the engine's events in that case.
111 self.dispatch = self.dispatch._join(engine.dispatch)
112 self._has_events = _has_events or (
113 _has_events is None and engine._has_events
114 )
116 assert not _execution_options
117 self._execution_options = engine._execution_options
119 if self._has_events or self.engine._has_events:
120 self.dispatch.engine_connect(self, _branch_from is not None)
122 @util.memoized_property
123 def _message_formatter(self):
124 if "logging_token" in self._execution_options:
125 token = self._execution_options["logging_token"]
126 return lambda msg: "[%s] %s" % (token, msg)
127 else:
128 return None
130 def _log_info(self, message, *arg, **kw):
131 fmt = self._message_formatter
133 if fmt:
134 message = fmt(message)
136 if log.STACKLEVEL:
137 kw["stacklevel"] = 1 + log.STACKLEVEL_OFFSET
139 self.engine.logger.info(message, *arg, **kw)
141 def _log_debug(self, message, *arg, **kw):
142 fmt = self._message_formatter
144 if fmt:
145 message = fmt(message)
147 if log.STACKLEVEL:
148 kw["stacklevel"] = 1 + log.STACKLEVEL_OFFSET
150 self.engine.logger.debug(message, *arg, **kw)
152 @property
153 def _schema_translate_map(self):
154 return self._execution_options.get("schema_translate_map", None)
156 def schema_for_object(self, obj):
157 """Return the schema name for the given schema item taking into
158 account current schema translate map.
160 """
162 name = obj.schema
163 schema_translate_map = self._execution_options.get(
164 "schema_translate_map", None
165 )
167 if (
168 schema_translate_map
169 and name in schema_translate_map
170 and obj._use_schema_map
171 ):
172 return schema_translate_map[name]
173 else:
174 return name
176 def _branch(self):
177 """Return a new Connection which references this Connection's
178 engine and connection; but does not have close_with_result enabled,
179 and also whose close() method does nothing.
181 .. deprecated:: 1.4 the "branching" concept will be removed in
182 SQLAlchemy 2.0 as well as the "Connection.connect()" method which
183 is the only consumer for this.
185 The Core uses this very sparingly, only in the case of
186 custom SQL default functions that are to be INSERTed as the
187 primary key of a row where we need to get the value back, so we have
188 to invoke it distinctly - this is a very uncommon case.
190 Userland code accesses _branch() when the connect()
191 method is called. The branched connection
192 acts as much as possible like the parent, except that it stays
193 connected when a close() event occurs.
195 """
196 return self.engine._connection_cls(
197 self.engine,
198 self._dbapi_connection,
199 _branch_from=self.__branch_from if self.__branch_from else self,
200 _execution_options=self._execution_options,
201 _has_events=self._has_events,
202 _dispatch=self.dispatch,
203 )
205 def _generate_for_options(self):
206 """define connection method chaining behavior for execution_options"""
208 if self._is_future:
209 return self
210 else:
211 c = self.__class__.__new__(self.__class__)
212 c.__dict__ = self.__dict__.copy()
213 return c
215 def __enter__(self):
216 return self
218 def __exit__(self, type_, value, traceback):
219 self.close()
221 def execution_options(self, **opt):
222 r""" Set non-SQL options for the connection which take effect
223 during execution.
225 For a "future" style connection, this method returns this same
226 :class:`_future.Connection` object with the new options added.
228 For a legacy connection, this method returns a copy of this
229 :class:`_engine.Connection` which references the same underlying DBAPI
230 connection, but also defines the given execution options which will
231 take effect for a call to
232 :meth:`execute`. As the new :class:`_engine.Connection` references the
233 same underlying resource, it's usually a good idea to ensure that
234 the copies will be discarded immediately, which is implicit if used
235 as in::
237 result = connection.execution_options(stream_results=True).\
238 execute(stmt)
240 Note that any key/value can be passed to
241 :meth:`_engine.Connection.execution_options`,
242 and it will be stored in the
243 ``_execution_options`` dictionary of the :class:`_engine.Connection`.
244 It
245 is suitable for usage by end-user schemes to communicate with
246 event listeners, for example.
248 The keywords that are currently recognized by SQLAlchemy itself
249 include all those listed under :meth:`.Executable.execution_options`,
250 as well as others that are specific to :class:`_engine.Connection`.
252 :param autocommit: Available on: Connection, statement.
253 When True, a COMMIT will be invoked after execution
254 when executed in 'autocommit' mode, i.e. when an explicit
255 transaction is not begun on the connection. Note that this
256 is **library level, not DBAPI level autocommit**. The DBAPI
257 connection will remain in a real transaction unless the
258 "AUTOCOMMIT" isolation level is used.
260 .. deprecated:: 1.4 The "autocommit" execution option is deprecated
261 and will be removed in SQLAlchemy 2.0. See
262 :ref:`migration_20_autocommit` for discussion.
264 :param compiled_cache: Available on: Connection.
265 A dictionary where :class:`.Compiled` objects
266 will be cached when the :class:`_engine.Connection`
267 compiles a clause
268 expression into a :class:`.Compiled` object. This dictionary will
269 supersede the statement cache that may be configured on the
270 :class:`_engine.Engine` itself. If set to None, caching
271 is disabled, even if the engine has a configured cache size.
273 Note that the ORM makes use of its own "compiled" caches for
274 some operations, including flush operations. The caching
275 used by the ORM internally supersedes a cache dictionary
276 specified here.
278 :param logging_token: Available on: :class:`_engine.Connection`,
279 :class:`_engine.Engine`.
281 Adds the specified string token surrounded by brackets in log
282 messages logged by the connection, i.e. the logging that's enabled
283 either via the :paramref:`_sa.create_engine.echo` flag or via the
284 ``logging.getLogger("sqlalchemy.engine")`` logger. This allows a
285 per-connection or per-sub-engine token to be available which is
286 useful for debugging concurrent connection scenarios.
288 .. versionadded:: 1.4.0b2
290 .. seealso::
292 :ref:`dbengine_logging_tokens` - usage example
294 :paramref:`_sa.create_engine.logging_name` - adds a name to the
295 name used by the Python logger object itself.
297 :param isolation_level: Available on: :class:`_engine.Connection`.
299 Set the transaction isolation level for the lifespan of this
300 :class:`_engine.Connection` object.
301 Valid values include those string
302 values accepted by the :paramref:`_sa.create_engine.isolation_level`
303 parameter passed to :func:`_sa.create_engine`. These levels are
304 semi-database specific; see individual dialect documentation for
305 valid levels.
307 The isolation level option applies the isolation level by emitting
308 statements on the DBAPI connection, and **necessarily affects the
309 original Connection object overall**, not just the copy that is
310 returned by the call to :meth:`_engine.Connection.execution_options`
311 method. The isolation level will remain at the given setting until
312 the DBAPI connection itself is returned to the connection pool, i.e.
313 the :meth:`_engine.Connection.close` method on the original
314 :class:`_engine.Connection` is called,
315 where an event handler will emit
316 additional statements on the DBAPI connection in order to revert the
317 isolation level change.
319 .. warning:: The ``isolation_level`` execution option should
320 **not** be used when a transaction is already established, that
321 is, the :meth:`_engine.Connection.begin`
322 method or similar has been
323 called. A database cannot change the isolation level on a
324 transaction in progress, and different DBAPIs and/or
325 SQLAlchemy dialects may implicitly roll back or commit
326 the transaction, or not affect the connection at all.
328 .. note:: The ``isolation_level`` execution option is implicitly
329 reset if the :class:`_engine.Connection` is invalidated, e.g. via
330 the :meth:`_engine.Connection.invalidate` method, or if a
331 disconnection error occurs. The new connection produced after
332 the invalidation will not have the isolation level re-applied
333 to it automatically.
335 .. seealso::
337 :paramref:`_sa.create_engine.isolation_level`
338 - set per :class:`_engine.Engine` isolation level
340 :meth:`_engine.Connection.get_isolation_level`
341 - view current actual level
343 :ref:`SQLite Transaction Isolation <sqlite_isolation_level>`
345 :ref:`PostgreSQL Transaction Isolation <postgresql_isolation_level>`
347 :ref:`MySQL Transaction Isolation <mysql_isolation_level>`
349 :ref:`SQL Server Transaction Isolation <mssql_isolation_level>`
351 :ref:`session_transaction_isolation` - for the ORM
353 :param no_parameters: When ``True``, if the final parameter
354 list or dictionary is totally empty, will invoke the
355 statement on the cursor as ``cursor.execute(statement)``,
356 not passing the parameter collection at all.
357 Some DBAPIs such as psycopg2 and mysql-python consider
358 percent signs as significant only when parameters are
359 present; this option allows code to generate SQL
360 containing percent signs (and possibly other characters)
361 that is neutral regarding whether it's executed by the DBAPI
362 or piped into a script that's later invoked by
363 command line tools.
365 :param stream_results: Available on: Connection, statement.
366 Indicate to the dialect that results should be
367 "streamed" and not pre-buffered, if possible. For backends
368 such as PostgreSQL, MySQL and MariaDB, this indicates the use of
369 a "server side cursor" as opposed to a client side cursor.
370 Other backends such as that of Oracle may already use server
371 side cursors by default.
373 The usage of
374 :paramref:`_engine.Connection.execution_options.stream_results` is
375 usually combined with setting a fixed number of rows to to be fetched
376 in batches, to allow for efficient iteration of database rows while
377 at the same time not loading all result rows into memory at once;
378 this can be configured on a :class:`_engine.Result` object using the
379 :meth:`_engine.Result.yield_per` method, after execution has
380 returned a new :class:`_engine.Result`. If
381 :meth:`_engine.Result.yield_per` is not used,
382 the :paramref:`_engine.Connection.execution_options.stream_results`
383 mode of operation will instead use a dynamically sized buffer
384 which buffers sets of rows at a time, growing on each batch
385 based on a fixed growth size up until a limit which may
386 be configured using the
387 :paramref:`_engine.Connection.execution_options.max_row_buffer`
388 parameter.
390 When using the ORM to fetch ORM mapped objects from a result,
391 :meth:`_engine.Result.yield_per` should always be used with
392 :paramref:`_engine.Connection.execution_options.stream_results`,
393 so that the ORM does not fetch all rows into new ORM objects at once.
395 For typical use, the
396 :paramref:`_engine.Connection.execution_options.yield_per` execution
397 option should be preferred, which sets up both
398 :paramref:`_engine.Connection.execution_options.stream_results` and
399 :meth:`_engine.Result.yield_per` at once. This option is supported
400 both at a core level by :class:`_engine.Connection` as well as by the
401 ORM :class:`_engine.Session`; the latter is described at
402 :ref:`orm_queryguide_yield_per`.
404 .. seealso::
406 :ref:`engine_stream_results` - background on
407 :paramref:`_engine.Connection.execution_options.stream_results`
409 :paramref:`_engine.Connection.execution_options.max_row_buffer`
411 :paramref:`_engine.Connection.execution_options.yield_per`
413 :ref:`orm_queryguide_yield_per` - in the :ref:`queryguide_toplevel`
414 describing the ORM version of ``yield_per``
416 :param max_row_buffer: Available on: :class:`_engine.Connection`,
417 :class:`_sql.Executable`. Sets a maximum
418 buffer size to use when the
419 :paramref:`_engine.Connection.execution_options.stream_results`
420 execution option is used on a backend that supports server side
421 cursors. The default value if not specified is 1000.
423 .. seealso::
425 :paramref:`_engine.Connection.execution_options.stream_results`
427 :ref:`engine_stream_results`
430 :param yield_per: Available on: :class:`_engine.Connection`,
431 :class:`_sql.Executable`. Integer value applied which will
432 set the :paramref:`_engine.Connection.execution_options.stream_results`
433 execution option and invoke :meth:`_engine.Result.yield_per`
434 automatically at once. Allows equivalent functionality as
435 is present when using this parameter with the ORM.
437 .. versionadded:: 1.4.40
439 .. seealso::
441 :ref:`engine_stream_results` - background and examples
442 on using server side cursors with Core.
444 :ref:`orm_queryguide_yield_per` - in the :ref:`queryguide_toplevel`
445 describing the ORM version of ``yield_per``
447 :param schema_translate_map: Available on: :class:`_engine.Connection`,
448 :class:`_engine.Engine`, :class:`_sql.Executable`.
450 :param schema_translate_map: Available on: Connection, Engine.
451 A dictionary mapping schema names to schema names, that will be
452 applied to the :paramref:`_schema.Table.schema` element of each
453 :class:`_schema.Table`
454 encountered when SQL or DDL expression elements
455 are compiled into strings; the resulting schema name will be
456 converted based on presence in the map of the original name.
458 .. versionadded:: 1.1
460 .. seealso::
462 :ref:`schema_translating`
464 .. seealso::
466 :meth:`_engine.Engine.execution_options`
468 :meth:`.Executable.execution_options`
470 :meth:`_engine.Connection.get_execution_options`
473 """ # noqa
474 c = self._generate_for_options()
475 c._execution_options = c._execution_options.union(opt)
476 if self._has_events or self.engine._has_events:
477 self.dispatch.set_connection_execution_options(c, opt)
478 self.dialect.set_connection_execution_options(c, opt)
479 return c
481 def get_execution_options(self):
482 """Get the non-SQL options which will take effect during execution.
484 .. versionadded:: 1.3
486 .. seealso::
488 :meth:`_engine.Connection.execution_options`
489 """
490 return self._execution_options
492 @property
493 def closed(self):
494 """Return True if this connection is closed."""
496 # note this is independent for a "branched" connection vs.
497 # the base
499 return self._dbapi_connection is None and not self.__can_reconnect
501 @property
502 def invalidated(self):
503 """Return True if this connection was invalidated."""
505 # prior to 1.4, "invalid" was stored as a state independent of
506 # "closed", meaning an invalidated connection could be "closed",
507 # the _dbapi_connection would be None and closed=True, yet the
508 # "invalid" flag would stay True. This meant that there were
509 # three separate states (open/valid, closed/valid, closed/invalid)
510 # when there is really no reason for that; a connection that's
511 # "closed" does not need to be "invalid". So the state is now
512 # represented by the two facts alone.
514 if self.__branch_from:
515 return self.__branch_from.invalidated
517 return self._dbapi_connection is None and not self.closed
519 @property
520 def connection(self):
521 """The underlying DB-API connection managed by this Connection.
523 This is a SQLAlchemy connection-pool proxied connection
524 which then has the attribute
525 :attr:`_pool._ConnectionFairy.dbapi_connection` that refers to the
526 actual driver connection.
528 .. seealso::
531 :ref:`dbapi_connections`
533 """
535 if self._dbapi_connection is None:
536 try:
537 return self._revalidate_connection()
538 except (exc.PendingRollbackError, exc.ResourceClosedError):
539 raise
540 except BaseException as e:
541 self._handle_dbapi_exception(e, None, None, None, None)
542 else:
543 return self._dbapi_connection
545 def get_isolation_level(self):
546 """Return the current **actual** isolation level that's present on
547 the database within the scope of this connection.
549 This attribute will perform a live SQL operation against the database
550 in order to procure the current isolation level, so the value returned
551 is the actual level on the underlying DBAPI connection regardless of
552 how this state was set. This will be one of the four actual isolation
553 modes ``READ UNCOMMITTED``, ``READ COMMITTED``, ``REPEATABLE READ``,
554 ``SERIALIZABLE``. It will **not** include the ``AUTOCOMMIT`` isolation
555 level setting. Third party dialects may also feature additional
556 isolation level settings.
558 .. note:: This method **will not report** on the ``AUTOCOMMIT``
559 isolation level, which is a separate :term:`dbapi` setting that's
560 independent of **actual** isolation level. When ``AUTOCOMMIT`` is
561 in use, the database connection still has a "traditional" isolation
562 mode in effect, that is typically one of the four values
563 ``READ UNCOMMITTED``, ``READ COMMITTED``, ``REPEATABLE READ``,
564 ``SERIALIZABLE``.
566 Compare to the :attr:`_engine.Connection.default_isolation_level`
567 accessor which returns the isolation level that is present on the
568 database at initial connection time.
570 .. versionadded:: 0.9.9
572 .. seealso::
574 :attr:`_engine.Connection.default_isolation_level`
575 - view default level
577 :paramref:`_sa.create_engine.isolation_level`
578 - set per :class:`_engine.Engine` isolation level
580 :paramref:`.Connection.execution_options.isolation_level`
581 - set per :class:`_engine.Connection` isolation level
583 """
584 try:
585 return self.dialect.get_isolation_level(self.connection)
586 except BaseException as e:
587 self._handle_dbapi_exception(e, None, None, None, None)
589 @property
590 def default_isolation_level(self):
591 """The initial-connection time isolation level associated with the
592 :class:`_engine.Dialect` in use.
594 This value is independent of the
595 :paramref:`.Connection.execution_options.isolation_level` and
596 :paramref:`.Engine.execution_options.isolation_level` execution
597 options, and is determined by the :class:`_engine.Dialect` when the
598 first connection is created, by performing a SQL query against the
599 database for the current isolation level before any additional commands
600 have been emitted.
602 Calling this accessor does not invoke any new SQL queries.
604 .. versionadded:: 0.9.9
606 .. seealso::
608 :meth:`_engine.Connection.get_isolation_level`
609 - view current actual isolation level
611 :paramref:`_sa.create_engine.isolation_level`
612 - set per :class:`_engine.Engine` isolation level
614 :paramref:`.Connection.execution_options.isolation_level`
615 - set per :class:`_engine.Connection` isolation level
617 """
618 return self.dialect.default_isolation_level
620 def _invalid_transaction(self):
621 if self.invalidated:
622 raise exc.PendingRollbackError(
623 "Can't reconnect until invalid %stransaction is rolled "
624 "back."
625 % (
626 "savepoint "
627 if self._nested_transaction is not None
628 else ""
629 ),
630 code="8s2b",
631 )
632 else:
633 assert not self._is_future
634 raise exc.PendingRollbackError(
635 "This connection is on an inactive %stransaction. "
636 "Please rollback() fully before proceeding."
637 % (
638 "savepoint "
639 if self._nested_transaction is not None
640 else ""
641 ),
642 code="8s2a",
643 )
645 def _revalidate_connection(self):
646 if self.__branch_from:
647 return self.__branch_from._revalidate_connection()
648 if self.__can_reconnect and self.invalidated:
649 if self._transaction is not None:
650 self._invalid_transaction()
651 self._dbapi_connection = self.engine.raw_connection(
652 _connection=self
653 )
654 return self._dbapi_connection
655 raise exc.ResourceClosedError("This Connection is closed")
657 @property
658 def _still_open_and_dbapi_connection_is_valid(self):
659 return self._dbapi_connection is not None and getattr(
660 self._dbapi_connection, "is_valid", False
661 )
663 @property
664 def info(self):
665 """Info dictionary associated with the underlying DBAPI connection
666 referred to by this :class:`_engine.Connection`, allowing user-defined
667 data to be associated with the connection.
669 The data here will follow along with the DBAPI connection including
670 after it is returned to the connection pool and used again
671 in subsequent instances of :class:`_engine.Connection`.
673 """
675 return self.connection.info
677 @util.deprecated_20(":meth:`.Connection.connect`")
678 def connect(self, close_with_result=False):
679 """Returns a branched version of this :class:`_engine.Connection`.
681 The :meth:`_engine.Connection.close` method on the returned
682 :class:`_engine.Connection` can be called and this
683 :class:`_engine.Connection` will remain open.
685 This method provides usage symmetry with
686 :meth:`_engine.Engine.connect`, including for usage
687 with context managers.
689 """
691 return self._branch()
693 def invalidate(self, exception=None):
694 """Invalidate the underlying DBAPI connection associated with
695 this :class:`_engine.Connection`.
697 An attempt will be made to close the underlying DBAPI connection
698 immediately; however if this operation fails, the error is logged
699 but not raised. The connection is then discarded whether or not
700 close() succeeded.
702 Upon the next use (where "use" typically means using the
703 :meth:`_engine.Connection.execute` method or similar),
704 this :class:`_engine.Connection` will attempt to
705 procure a new DBAPI connection using the services of the
706 :class:`_pool.Pool` as a source of connectivity (e.g.
707 a "reconnection").
709 If a transaction was in progress (e.g. the
710 :meth:`_engine.Connection.begin` method has been called) when
711 :meth:`_engine.Connection.invalidate` method is called, at the DBAPI
712 level all state associated with this transaction is lost, as
713 the DBAPI connection is closed. The :class:`_engine.Connection`
714 will not allow a reconnection to proceed until the
715 :class:`.Transaction` object is ended, by calling the
716 :meth:`.Transaction.rollback` method; until that point, any attempt at
717 continuing to use the :class:`_engine.Connection` will raise an
718 :class:`~sqlalchemy.exc.InvalidRequestError`.
719 This is to prevent applications from accidentally
720 continuing an ongoing transactional operations despite the
721 fact that the transaction has been lost due to an
722 invalidation.
724 The :meth:`_engine.Connection.invalidate` method,
725 just like auto-invalidation,
726 will at the connection pool level invoke the
727 :meth:`_events.PoolEvents.invalidate` event.
729 :param exception: an optional ``Exception`` instance that's the
730 reason for the invalidation. is passed along to event handlers
731 and logging functions.
733 .. seealso::
735 :ref:`pool_connection_invalidation`
737 """
739 if self.__branch_from:
740 return self.__branch_from.invalidate(exception=exception)
742 if self.invalidated:
743 return
745 if self.closed:
746 raise exc.ResourceClosedError("This Connection is closed")
748 if self._still_open_and_dbapi_connection_is_valid:
749 self._dbapi_connection.invalidate(exception)
750 self._dbapi_connection = None
752 def detach(self):
753 """Detach the underlying DB-API connection from its connection pool.
755 E.g.::
757 with engine.connect() as conn:
758 conn.detach()
759 conn.execute(text("SET search_path TO schema1, schema2"))
761 # work with connection
763 # connection is fully closed (since we used "with:", can
764 # also call .close())
766 This :class:`_engine.Connection` instance will remain usable.
767 When closed
768 (or exited from a context manager context as above),
769 the DB-API connection will be literally closed and not
770 returned to its originating pool.
772 This method can be used to insulate the rest of an application
773 from a modified state on a connection (such as a transaction
774 isolation level or similar).
776 """
778 self._dbapi_connection.detach()
780 def _autobegin(self):
781 self.begin()
783 def begin(self):
784 """Begin a transaction and return a transaction handle.
786 The returned object is an instance of :class:`.Transaction`.
787 This object represents the "scope" of the transaction,
788 which completes when either the :meth:`.Transaction.rollback`
789 or :meth:`.Transaction.commit` method is called.
791 .. tip::
793 The :meth:`_engine.Connection.begin` method is invoked when using
794 the :meth:`_engine.Engine.begin` context manager method as well.
795 All documentation that refers to behaviors specific to the
796 :meth:`_engine.Connection.begin` method also apply to use of the
797 :meth:`_engine.Engine.begin` method.
799 Legacy use: nested calls to :meth:`.begin` on the same
800 :class:`_engine.Connection` will return new :class:`.Transaction`
801 objects that represent an emulated transaction within the scope of the
802 enclosing transaction, that is::
804 trans = conn.begin() # outermost transaction
805 trans2 = conn.begin() # "nested"
806 trans2.commit() # does nothing
807 trans.commit() # actually commits
809 Calls to :meth:`.Transaction.commit` only have an effect
810 when invoked via the outermost :class:`.Transaction` object, though the
811 :meth:`.Transaction.rollback` method of any of the
812 :class:`.Transaction` objects will roll back the
813 transaction.
815 .. tip::
817 The above "nesting" behavior is a legacy behavior specific to
818 :term:`1.x style` use and will be removed in SQLAlchemy 2.0. For
819 notes on :term:`2.0 style` use, see
820 :meth:`_future.Connection.begin`.
823 .. seealso::
825 :meth:`_engine.Connection.begin_nested` - use a SAVEPOINT
827 :meth:`_engine.Connection.begin_twophase` -
828 use a two phase /XID transaction
830 :meth:`_engine.Engine.begin` - context manager available from
831 :class:`_engine.Engine`
833 """
834 if self._is_future:
835 assert not self.__branch_from
836 elif self.__branch_from:
837 return self.__branch_from.begin()
839 if self.__in_begin:
840 # for dialects that emit SQL within the process of
841 # dialect.do_begin() or dialect.do_begin_twophase(), this
842 # flag prevents "autobegin" from being emitted within that
843 # process, while allowing self._transaction to remain at None
844 # until it's complete.
845 return
846 elif self._transaction is None:
847 self._transaction = RootTransaction(self)
848 return self._transaction
849 else:
850 if self._is_future:
851 raise exc.InvalidRequestError(
852 "This connection has already initialized a SQLAlchemy "
853 "Transaction() object via begin() or autobegin; can't "
854 "call begin() here unless rollback() or commit() "
855 "is called first."
856 )
857 else:
858 return MarkerTransaction(self)
860 def begin_nested(self):
861 """Begin a nested transaction (i.e. SAVEPOINT) and return a
862 transaction handle, assuming an outer transaction is already
863 established.
865 Nested transactions require SAVEPOINT support in the
866 underlying database. Any transaction in the hierarchy may
867 ``commit`` and ``rollback``, however the outermost transaction
868 still controls the overall ``commit`` or ``rollback`` of the
869 transaction of a whole.
871 The legacy form of :meth:`_engine.Connection.begin_nested` method has
872 alternate behaviors based on whether or not the
873 :meth:`_engine.Connection.begin` method was called previously. If
874 :meth:`_engine.Connection.begin` was not called, then this method will
875 behave the same as the :meth:`_engine.Connection.begin` method and
876 return a :class:`.RootTransaction` object that begins and commits a
877 real transaction - **no savepoint is invoked**. If
878 :meth:`_engine.Connection.begin` **has** been called, and a
879 :class:`.RootTransaction` is already established, then this method
880 returns an instance of :class:`.NestedTransaction` which will invoke
881 and manage the scope of a SAVEPOINT.
883 .. tip::
885 The above mentioned behavior of
886 :meth:`_engine.Connection.begin_nested` is a legacy behavior
887 specific to :term:`1.x style` use. In :term:`2.0 style` use, the
888 :meth:`_future.Connection.begin_nested` method instead autobegins
889 the outer transaction that can be committed using
890 "commit-as-you-go" style; see
891 :meth:`_future.Connection.begin_nested` for migration details.
893 .. versionchanged:: 1.4.13 The behavior of
894 :meth:`_engine.Connection.begin_nested`
895 as returning a :class:`.RootTransaction` if
896 :meth:`_engine.Connection.begin` were not called has been restored
897 as was the case in 1.3.x versions; in previous 1.4.x versions, an
898 outer transaction would be "autobegun" but would not be committed.
901 .. seealso::
903 :meth:`_engine.Connection.begin`
905 :ref:`session_begin_nested` - ORM support for SAVEPOINT
907 """
908 if self._is_future:
909 assert not self.__branch_from
910 elif self.__branch_from:
911 return self.__branch_from.begin_nested()
913 if self._transaction is None:
914 if not self._is_future:
915 util.warn_deprecated_20(
916 "Calling Connection.begin_nested() in 2.0 style use will "
917 "return a NestedTransaction (SAVEPOINT) in all cases, "
918 "that will not commit the outer transaction. For code "
919 "that is cross-compatible between 1.x and 2.0 style use, "
920 "ensure Connection.begin() is called before calling "
921 "Connection.begin_nested()."
922 )
923 return self.begin()
924 else:
925 self._autobegin()
927 return NestedTransaction(self)
929 def begin_twophase(self, xid=None):
930 """Begin a two-phase or XA transaction and return a transaction
931 handle.
933 The returned object is an instance of :class:`.TwoPhaseTransaction`,
934 which in addition to the methods provided by
935 :class:`.Transaction`, also provides a
936 :meth:`~.TwoPhaseTransaction.prepare` method.
938 :param xid: the two phase transaction id. If not supplied, a
939 random id will be generated.
941 .. seealso::
943 :meth:`_engine.Connection.begin`
945 :meth:`_engine.Connection.begin_twophase`
947 """
949 if self.__branch_from:
950 return self.__branch_from.begin_twophase(xid=xid)
952 if self._transaction is not None:
953 raise exc.InvalidRequestError(
954 "Cannot start a two phase transaction when a transaction "
955 "is already in progress."
956 )
957 if xid is None:
958 xid = self.engine.dialect.create_xid()
959 return TwoPhaseTransaction(self, xid)
961 def recover_twophase(self):
962 return self.engine.dialect.do_recover_twophase(self)
964 def rollback_prepared(self, xid, recover=False):
965 self.engine.dialect.do_rollback_twophase(self, xid, recover=recover)
967 def commit_prepared(self, xid, recover=False):
968 self.engine.dialect.do_commit_twophase(self, xid, recover=recover)
970 def in_transaction(self):
971 """Return True if a transaction is in progress."""
972 if self.__branch_from is not None:
973 return self.__branch_from.in_transaction()
975 return self._transaction is not None and self._transaction.is_active
977 def in_nested_transaction(self):
978 """Return True if a transaction is in progress."""
979 if self.__branch_from is not None:
980 return self.__branch_from.in_nested_transaction()
982 return (
983 self._nested_transaction is not None
984 and self._nested_transaction.is_active
985 )
987 def _is_autocommit_isolation(self):
988 opt_iso = self._execution_options.get("isolation_level", None)
989 return bool(
990 opt_iso == "AUTOCOMMIT"
991 or (
992 opt_iso is None
993 and getattr(self.engine.dialect, "isolation_level", None)
994 == "AUTOCOMMIT"
995 )
996 )
998 def get_transaction(self):
999 """Return the current root transaction in progress, if any.
1001 .. versionadded:: 1.4
1003 """
1005 if self.__branch_from is not None:
1006 return self.__branch_from.get_transaction()
1008 return self._transaction
1010 def get_nested_transaction(self):
1011 """Return the current nested transaction in progress, if any.
1013 .. versionadded:: 1.4
1015 """
1016 if self.__branch_from is not None:
1018 return self.__branch_from.get_nested_transaction()
1020 return self._nested_transaction
1022 def _begin_impl(self, transaction):
1023 assert not self.__branch_from
1025 if self._echo:
1026 if self._is_autocommit_isolation():
1027 self._log_info(
1028 "BEGIN (implicit; DBAPI should not BEGIN due to "
1029 "autocommit mode)"
1030 )
1031 else:
1032 self._log_info("BEGIN (implicit)")
1034 self.__in_begin = True
1036 if self._has_events or self.engine._has_events:
1037 self.dispatch.begin(self)
1039 try:
1040 self.engine.dialect.do_begin(self.connection)
1041 except BaseException as e:
1042 self._handle_dbapi_exception(e, None, None, None, None)
1043 finally:
1044 self.__in_begin = False
1046 def _rollback_impl(self):
1047 assert not self.__branch_from
1049 if self._has_events or self.engine._has_events:
1050 self.dispatch.rollback(self)
1052 if self._still_open_and_dbapi_connection_is_valid:
1053 if self._echo:
1054 if self._is_autocommit_isolation():
1055 self._log_info(
1056 "ROLLBACK using DBAPI connection.rollback(), "
1057 "DBAPI should ignore due to autocommit mode"
1058 )
1059 else:
1060 self._log_info("ROLLBACK")
1061 try:
1062 self.engine.dialect.do_rollback(self.connection)
1063 except BaseException as e:
1064 self._handle_dbapi_exception(e, None, None, None, None)
1066 def _commit_impl(self, autocommit=False):
1067 assert not self.__branch_from
1069 # AUTOCOMMIT isolation-level is a dialect-specific concept, however
1070 # if a connection has this set as the isolation level, we can skip
1071 # the "autocommit" warning as the operation will do "autocommit"
1072 # in any case
1073 if autocommit and not self._is_autocommit_isolation():
1074 util.warn_deprecated_20(
1075 "The current statement is being autocommitted using "
1076 "implicit autocommit, which will be removed in "
1077 "SQLAlchemy 2.0. "
1078 "Use the .begin() method of Engine or Connection in order to "
1079 "use an explicit transaction for DML and DDL statements."
1080 )
1082 if self._has_events or self.engine._has_events:
1083 self.dispatch.commit(self)
1085 if self._echo:
1086 if self._is_autocommit_isolation():
1087 self._log_info(
1088 "COMMIT using DBAPI connection.commit(), "
1089 "DBAPI should ignore due to autocommit mode"
1090 )
1091 else:
1092 self._log_info("COMMIT")
1093 try:
1094 self.engine.dialect.do_commit(self.connection)
1095 except BaseException as e:
1096 self._handle_dbapi_exception(e, None, None, None, None)
1098 def _savepoint_impl(self, name=None):
1099 assert not self.__branch_from
1101 if self._has_events or self.engine._has_events:
1102 self.dispatch.savepoint(self, name)
1104 if name is None:
1105 self.__savepoint_seq += 1
1106 name = "sa_savepoint_%s" % self.__savepoint_seq
1107 if self._still_open_and_dbapi_connection_is_valid:
1108 self.engine.dialect.do_savepoint(self, name)
1109 return name
1111 def _rollback_to_savepoint_impl(self, name):
1112 assert not self.__branch_from
1114 if self._has_events or self.engine._has_events:
1115 self.dispatch.rollback_savepoint(self, name, None)
1117 if self._still_open_and_dbapi_connection_is_valid:
1118 self.engine.dialect.do_rollback_to_savepoint(self, name)
1120 def _release_savepoint_impl(self, name):
1121 assert not self.__branch_from
1123 if self._has_events or self.engine._has_events:
1124 self.dispatch.release_savepoint(self, name, None)
1126 if self._still_open_and_dbapi_connection_is_valid:
1127 self.engine.dialect.do_release_savepoint(self, name)
1129 def _begin_twophase_impl(self, transaction):
1130 assert not self.__branch_from
1132 if self._echo:
1133 self._log_info("BEGIN TWOPHASE (implicit)")
1134 if self._has_events or self.engine._has_events:
1135 self.dispatch.begin_twophase(self, transaction.xid)
1137 if self._still_open_and_dbapi_connection_is_valid:
1138 self.__in_begin = True
1139 try:
1140 self.engine.dialect.do_begin_twophase(self, transaction.xid)
1141 except BaseException as e:
1142 self._handle_dbapi_exception(e, None, None, None, None)
1143 finally:
1144 self.__in_begin = False
1146 def _prepare_twophase_impl(self, xid):
1147 assert not self.__branch_from
1149 if self._has_events or self.engine._has_events:
1150 self.dispatch.prepare_twophase(self, xid)
1152 if self._still_open_and_dbapi_connection_is_valid:
1153 assert isinstance(self._transaction, TwoPhaseTransaction)
1154 try:
1155 self.engine.dialect.do_prepare_twophase(self, xid)
1156 except BaseException as e:
1157 self._handle_dbapi_exception(e, None, None, None, None)
1159 def _rollback_twophase_impl(self, xid, is_prepared):
1160 assert not self.__branch_from
1162 if self._has_events or self.engine._has_events:
1163 self.dispatch.rollback_twophase(self, xid, is_prepared)
1165 if self._still_open_and_dbapi_connection_is_valid:
1166 assert isinstance(self._transaction, TwoPhaseTransaction)
1167 try:
1168 self.engine.dialect.do_rollback_twophase(
1169 self, xid, is_prepared
1170 )
1171 except BaseException as e:
1172 self._handle_dbapi_exception(e, None, None, None, None)
1174 def _commit_twophase_impl(self, xid, is_prepared):
1175 assert not self.__branch_from
1177 if self._has_events or self.engine._has_events:
1178 self.dispatch.commit_twophase(self, xid, is_prepared)
1180 if self._still_open_and_dbapi_connection_is_valid:
1181 assert isinstance(self._transaction, TwoPhaseTransaction)
1182 try:
1183 self.engine.dialect.do_commit_twophase(self, xid, is_prepared)
1184 except BaseException as e:
1185 self._handle_dbapi_exception(e, None, None, None, None)
1187 def _autorollback(self):
1188 if self.__branch_from:
1189 self.__branch_from._autorollback()
1191 if not self.in_transaction():
1192 self._rollback_impl()
1194 def _warn_for_legacy_exec_format(self):
1195 util.warn_deprecated_20(
1196 "The connection.execute() method in "
1197 "SQLAlchemy 2.0 will accept parameters as a single "
1198 "dictionary or a "
1199 "single sequence of dictionaries only. "
1200 "Parameters passed as keyword arguments, tuples or positionally "
1201 "oriented dictionaries and/or tuples "
1202 "will no longer be accepted."
1203 )
1205 def close(self):
1206 """Close this :class:`_engine.Connection`.
1208 This results in a release of the underlying database
1209 resources, that is, the DBAPI connection referenced
1210 internally. The DBAPI connection is typically restored
1211 back to the connection-holding :class:`_pool.Pool` referenced
1212 by the :class:`_engine.Engine` that produced this
1213 :class:`_engine.Connection`. Any transactional state present on
1214 the DBAPI connection is also unconditionally released via
1215 the DBAPI connection's ``rollback()`` method, regardless
1216 of any :class:`.Transaction` object that may be
1217 outstanding with regards to this :class:`_engine.Connection`.
1219 After :meth:`_engine.Connection.close` is called, the
1220 :class:`_engine.Connection` is permanently in a closed state,
1221 and will allow no further operations.
1223 """
1225 if self.__branch_from:
1226 assert not self._is_future
1227 util.warn_deprecated_20(
1228 "The .close() method on a so-called 'branched' connection is "
1229 "deprecated as of 1.4, as are 'branched' connections overall, "
1230 "and will be removed in a future release. If this is a "
1231 "default-handling function, don't close the connection."
1232 )
1233 self._dbapi_connection = None
1234 self.__can_reconnect = False
1235 return
1237 if self._transaction:
1238 self._transaction.close()
1239 skip_reset = True
1240 else:
1241 skip_reset = False
1243 if self._dbapi_connection is not None:
1244 conn = self._dbapi_connection
1246 # as we just closed the transaction, close the connection
1247 # pool connection without doing an additional reset
1248 if skip_reset:
1249 conn._close_special(transaction_reset=True)
1250 else:
1251 conn.close()
1253 # There is a slight chance that conn.close() may have
1254 # triggered an invalidation here in which case
1255 # _dbapi_connection would already be None, however usually
1256 # it will be non-None here and in a "closed" state.
1257 self._dbapi_connection = None
1258 self.__can_reconnect = False
1260 def scalar(self, object_, *multiparams, **params):
1261 """Executes and returns the first column of the first row.
1263 The underlying result/cursor is closed after execution.
1265 """
1267 return self.execute(object_, *multiparams, **params).scalar()
1269 def scalars(self, object_, *multiparams, **params):
1270 """Executes and returns a scalar result set, which yields scalar values
1271 from the first column of each row.
1273 This method is equivalent to calling :meth:`_engine.Connection.execute`
1274 to receive a :class:`_result.Result` object, then invoking the
1275 :meth:`_result.Result.scalars` method to produce a
1276 :class:`_result.ScalarResult` instance.
1278 :return: a :class:`_result.ScalarResult`
1280 .. versionadded:: 1.4.24
1282 """
1284 return self.execute(object_, *multiparams, **params).scalars()
1286 def execute(self, statement, *multiparams, **params):
1287 r"""Executes a SQL statement construct and returns a
1288 :class:`_engine.CursorResult`.
1290 :param statement: The statement to be executed. May be
1291 one of:
1293 * a plain string (deprecated)
1294 * any :class:`_expression.ClauseElement` construct that is also
1295 a subclass of :class:`.Executable`, such as a
1296 :func:`_expression.select` construct
1297 * a :class:`.FunctionElement`, such as that generated
1298 by :data:`.func`, will be automatically wrapped in
1299 a SELECT statement, which is then executed.
1300 * a :class:`.DDLElement` object
1301 * a :class:`.DefaultGenerator` object
1302 * a :class:`.Compiled` object
1304 .. deprecated:: 2.0 passing a string to
1305 :meth:`_engine.Connection.execute` is
1306 deprecated and will be removed in version 2.0. Use the
1307 :func:`_expression.text` construct with
1308 :meth:`_engine.Connection.execute`, or the
1309 :meth:`_engine.Connection.exec_driver_sql`
1310 method to invoke a driver-level
1311 SQL string.
1313 :param \*multiparams/\**params: represent bound parameter
1314 values to be used in the execution. Typically,
1315 the format is either a collection of one or more
1316 dictionaries passed to \*multiparams::
1318 conn.execute(
1319 table.insert(),
1320 {"id":1, "value":"v1"},
1321 {"id":2, "value":"v2"}
1322 )
1324 ...or individual key/values interpreted by \**params::
1326 conn.execute(
1327 table.insert(), id=1, value="v1"
1328 )
1330 In the case that a plain SQL string is passed, and the underlying
1331 DBAPI accepts positional bind parameters, a collection of tuples
1332 or individual values in \*multiparams may be passed::
1334 conn.execute(
1335 "INSERT INTO table (id, value) VALUES (?, ?)",
1336 (1, "v1"), (2, "v2")
1337 )
1339 conn.execute(
1340 "INSERT INTO table (id, value) VALUES (?, ?)",
1341 1, "v1"
1342 )
1344 Note above, the usage of a question mark "?" or other
1345 symbol is contingent upon the "paramstyle" accepted by the DBAPI
1346 in use, which may be any of "qmark", "named", "pyformat", "format",
1347 "numeric". See `pep-249
1348 <https://www.python.org/dev/peps/pep-0249/>`_ for details on
1349 paramstyle.
1351 To execute a textual SQL statement which uses bound parameters in a
1352 DBAPI-agnostic way, use the :func:`_expression.text` construct.
1354 .. deprecated:: 2.0 use of tuple or scalar positional parameters
1355 is deprecated. All params should be dicts or sequences of dicts.
1356 Use :meth:`.exec_driver_sql` to execute a plain string with
1357 tuple or scalar positional parameters.
1359 """
1361 if isinstance(statement, util.string_types):
1362 util.warn_deprecated_20(
1363 "Passing a string to Connection.execute() is "
1364 "deprecated and will be removed in version 2.0. Use the "
1365 "text() construct, "
1366 "or the Connection.exec_driver_sql() method to invoke a "
1367 "driver-level SQL string."
1368 )
1370 return self._exec_driver_sql(
1371 statement,
1372 multiparams,
1373 params,
1374 _EMPTY_EXECUTION_OPTS,
1375 future=False,
1376 )
1378 try:
1379 meth = statement._execute_on_connection
1380 except AttributeError as err:
1381 util.raise_(
1382 exc.ObjectNotExecutableError(statement), replace_context=err
1383 )
1384 else:
1385 return meth(self, multiparams, params, _EMPTY_EXECUTION_OPTS)
1387 def _execute_function(self, func, multiparams, params, execution_options):
1388 """Execute a sql.FunctionElement object."""
1390 return self._execute_clauseelement(
1391 func.select(), multiparams, params, execution_options
1392 )
1394 def _execute_default(
1395 self,
1396 default,
1397 multiparams,
1398 params,
1399 # migrate is calling this directly :(
1400 execution_options=_EMPTY_EXECUTION_OPTS,
1401 ):
1402 """Execute a schema.ColumnDefault object."""
1404 execution_options = self._execution_options.merge_with(
1405 execution_options
1406 )
1408 distilled_parameters = _distill_params(self, multiparams, params)
1410 if self._has_events or self.engine._has_events:
1411 (
1412 default,
1413 distilled_params,
1414 event_multiparams,
1415 event_params,
1416 ) = self._invoke_before_exec_event(
1417 default, distilled_parameters, execution_options
1418 )
1420 try:
1421 conn = self._dbapi_connection
1422 if conn is None:
1423 conn = self._revalidate_connection()
1425 dialect = self.dialect
1426 ctx = dialect.execution_ctx_cls._init_default(
1427 dialect, self, conn, execution_options
1428 )
1429 except (exc.PendingRollbackError, exc.ResourceClosedError):
1430 raise
1431 except BaseException as e:
1432 self._handle_dbapi_exception(e, None, None, None, None)
1434 ret = ctx._exec_default(None, default, None)
1435 if self.should_close_with_result:
1436 self.close()
1438 if self._has_events or self.engine._has_events:
1439 self.dispatch.after_execute(
1440 self,
1441 default,
1442 event_multiparams,
1443 event_params,
1444 execution_options,
1445 ret,
1446 )
1448 return ret
1450 def _execute_ddl(self, ddl, multiparams, params, execution_options):
1451 """Execute a schema.DDL object."""
1453 execution_options = ddl._execution_options.merge_with(
1454 self._execution_options, execution_options
1455 )
1457 distilled_parameters = _distill_params(self, multiparams, params)
1459 if self._has_events or self.engine._has_events:
1460 (
1461 ddl,
1462 distilled_params,
1463 event_multiparams,
1464 event_params,
1465 ) = self._invoke_before_exec_event(
1466 ddl, distilled_parameters, execution_options
1467 )
1469 exec_opts = self._execution_options.merge_with(execution_options)
1470 schema_translate_map = exec_opts.get("schema_translate_map", None)
1472 dialect = self.dialect
1474 compiled = ddl.compile(
1475 dialect=dialect, schema_translate_map=schema_translate_map
1476 )
1477 ret = self._execute_context(
1478 dialect,
1479 dialect.execution_ctx_cls._init_ddl,
1480 compiled,
1481 None,
1482 execution_options,
1483 compiled,
1484 )
1485 if self._has_events or self.engine._has_events:
1486 self.dispatch.after_execute(
1487 self,
1488 ddl,
1489 event_multiparams,
1490 event_params,
1491 execution_options,
1492 ret,
1493 )
1494 return ret
1496 def _invoke_before_exec_event(
1497 self, elem, distilled_params, execution_options
1498 ):
1500 if len(distilled_params) == 1:
1501 event_multiparams, event_params = [], distilled_params[0]
1502 else:
1503 event_multiparams, event_params = distilled_params, {}
1505 for fn in self.dispatch.before_execute:
1506 elem, event_multiparams, event_params = fn(
1507 self,
1508 elem,
1509 event_multiparams,
1510 event_params,
1511 execution_options,
1512 )
1514 if event_multiparams:
1515 distilled_params = list(event_multiparams)
1516 if event_params:
1517 raise exc.InvalidRequestError(
1518 "Event handler can't return non-empty multiparams "
1519 "and params at the same time"
1520 )
1521 elif event_params:
1522 distilled_params = [event_params]
1523 else:
1524 distilled_params = []
1526 return elem, distilled_params, event_multiparams, event_params
1528 def _execute_clauseelement(
1529 self, elem, multiparams, params, execution_options
1530 ):
1531 """Execute a sql.ClauseElement object."""
1533 execution_options = elem._execution_options.merge_with(
1534 self._execution_options, execution_options
1535 )
1537 distilled_params = _distill_params(self, multiparams, params)
1539 has_events = self._has_events or self.engine._has_events
1540 if has_events:
1541 (
1542 elem,
1543 distilled_params,
1544 event_multiparams,
1545 event_params,
1546 ) = self._invoke_before_exec_event(
1547 elem, distilled_params, execution_options
1548 )
1550 if distilled_params:
1551 # ensure we don't retain a link to the view object for keys()
1552 # which links to the values, which we don't want to cache
1553 keys = sorted(distilled_params[0])
1554 for_executemany = len(distilled_params) > 1
1555 else:
1556 keys = []
1557 for_executemany = False
1559 dialect = self.dialect
1561 schema_translate_map = execution_options.get(
1562 "schema_translate_map", None
1563 )
1565 compiled_cache = execution_options.get(
1566 "compiled_cache", self.engine._compiled_cache
1567 )
1569 compiled_sql, extracted_params, cache_hit = elem._compile_w_cache(
1570 dialect=dialect,
1571 compiled_cache=compiled_cache,
1572 column_keys=keys,
1573 for_executemany=for_executemany,
1574 schema_translate_map=schema_translate_map,
1575 linting=self.dialect.compiler_linting | compiler.WARN_LINTING,
1576 )
1577 ret = self._execute_context(
1578 dialect,
1579 dialect.execution_ctx_cls._init_compiled,
1580 compiled_sql,
1581 distilled_params,
1582 execution_options,
1583 compiled_sql,
1584 distilled_params,
1585 elem,
1586 extracted_params,
1587 cache_hit=cache_hit,
1588 )
1589 if has_events:
1590 self.dispatch.after_execute(
1591 self,
1592 elem,
1593 event_multiparams,
1594 event_params,
1595 execution_options,
1596 ret,
1597 )
1598 return ret
1600 def _execute_compiled(
1601 self,
1602 compiled,
1603 multiparams,
1604 params,
1605 execution_options=_EMPTY_EXECUTION_OPTS,
1606 ):
1607 """Execute a sql.Compiled object.
1609 TODO: why do we have this? likely deprecate or remove
1611 """
1613 execution_options = compiled.execution_options.merge_with(
1614 self._execution_options, execution_options
1615 )
1616 distilled_parameters = _distill_params(self, multiparams, params)
1618 if self._has_events or self.engine._has_events:
1619 (
1620 compiled,
1621 distilled_params,
1622 event_multiparams,
1623 event_params,
1624 ) = self._invoke_before_exec_event(
1625 compiled, distilled_parameters, execution_options
1626 )
1628 dialect = self.dialect
1630 ret = self._execute_context(
1631 dialect,
1632 dialect.execution_ctx_cls._init_compiled,
1633 compiled,
1634 distilled_parameters,
1635 execution_options,
1636 compiled,
1637 distilled_parameters,
1638 None,
1639 None,
1640 )
1641 if self._has_events or self.engine._has_events:
1642 self.dispatch.after_execute(
1643 self,
1644 compiled,
1645 event_multiparams,
1646 event_params,
1647 execution_options,
1648 ret,
1649 )
1650 return ret
1652 def _exec_driver_sql(
1653 self, statement, multiparams, params, execution_options, future
1654 ):
1656 execution_options = self._execution_options.merge_with(
1657 execution_options
1658 )
1660 distilled_parameters = _distill_params(self, multiparams, params)
1662 if not future:
1663 if self._has_events or self.engine._has_events:
1664 (
1665 statement,
1666 distilled_params,
1667 event_multiparams,
1668 event_params,
1669 ) = self._invoke_before_exec_event(
1670 statement, distilled_parameters, execution_options
1671 )
1673 dialect = self.dialect
1674 ret = self._execute_context(
1675 dialect,
1676 dialect.execution_ctx_cls._init_statement,
1677 statement,
1678 distilled_parameters,
1679 execution_options,
1680 statement,
1681 distilled_parameters,
1682 )
1684 if not future:
1685 if self._has_events or self.engine._has_events:
1686 self.dispatch.after_execute(
1687 self,
1688 statement,
1689 event_multiparams,
1690 event_params,
1691 execution_options,
1692 ret,
1693 )
1694 return ret
1696 def _execute_20(
1697 self,
1698 statement,
1699 parameters=None,
1700 execution_options=_EMPTY_EXECUTION_OPTS,
1701 ):
1702 args_10style, kwargs_10style = _distill_params_20(parameters)
1703 try:
1704 meth = statement._execute_on_connection
1705 except AttributeError as err:
1706 util.raise_(
1707 exc.ObjectNotExecutableError(statement), replace_context=err
1708 )
1709 else:
1710 return meth(self, args_10style, kwargs_10style, execution_options)
1712 def exec_driver_sql(
1713 self, statement, parameters=None, execution_options=None
1714 ):
1715 r"""Executes a SQL statement construct and returns a
1716 :class:`_engine.CursorResult`.
1718 :param statement: The statement str to be executed. Bound parameters
1719 must use the underlying DBAPI's paramstyle, such as "qmark",
1720 "pyformat", "format", etc.
1722 :param parameters: represent bound parameter values to be used in the
1723 execution. The format is one of: a dictionary of named parameters,
1724 a tuple of positional parameters, or a list containing either
1725 dictionaries or tuples for multiple-execute support.
1727 E.g. multiple dictionaries::
1730 conn.exec_driver_sql(
1731 "INSERT INTO table (id, value) VALUES (%(id)s, %(value)s)",
1732 [{"id":1, "value":"v1"}, {"id":2, "value":"v2"}]
1733 )
1735 Single dictionary::
1737 conn.exec_driver_sql(
1738 "INSERT INTO table (id, value) VALUES (%(id)s, %(value)s)",
1739 dict(id=1, value="v1")
1740 )
1742 Single tuple::
1744 conn.exec_driver_sql(
1745 "INSERT INTO table (id, value) VALUES (?, ?)",
1746 (1, 'v1')
1747 )
1749 .. note:: The :meth:`_engine.Connection.exec_driver_sql` method does
1750 not participate in the
1751 :meth:`_events.ConnectionEvents.before_execute` and
1752 :meth:`_events.ConnectionEvents.after_execute` events. To
1753 intercept calls to :meth:`_engine.Connection.exec_driver_sql`, use
1754 :meth:`_events.ConnectionEvents.before_cursor_execute` and
1755 :meth:`_events.ConnectionEvents.after_cursor_execute`.
1757 .. seealso::
1759 :pep:`249`
1761 """
1763 args_10style, kwargs_10style = _distill_params_20(parameters)
1765 return self._exec_driver_sql(
1766 statement,
1767 args_10style,
1768 kwargs_10style,
1769 execution_options,
1770 future=True,
1771 )
1773 def _execute_context(
1774 self,
1775 dialect,
1776 constructor,
1777 statement,
1778 parameters,
1779 execution_options,
1780 *args,
1781 **kw
1782 ):
1783 """Create an :class:`.ExecutionContext` and execute, returning
1784 a :class:`_engine.CursorResult`."""
1786 branched = self
1787 if self.__branch_from:
1788 # if this is a "branched" connection, do everything in terms
1789 # of the "root" connection, *except* for .close(), which is
1790 # the only feature that branching provides
1791 self = self.__branch_from
1793 if execution_options:
1794 yp = execution_options.get("yield_per", None)
1795 if yp:
1796 execution_options = execution_options.union(
1797 {"stream_results": True, "max_row_buffer": yp}
1798 )
1800 try:
1801 conn = self._dbapi_connection
1802 if conn is None:
1803 conn = self._revalidate_connection()
1805 context = constructor(
1806 dialect, self, conn, execution_options, *args, **kw
1807 )
1808 except (exc.PendingRollbackError, exc.ResourceClosedError):
1809 raise
1810 except BaseException as e:
1811 self._handle_dbapi_exception(
1812 e, util.text_type(statement), parameters, None, None
1813 )
1815 if (
1816 self._transaction
1817 and not self._transaction.is_active
1818 or (
1819 self._nested_transaction
1820 and not self._nested_transaction.is_active
1821 )
1822 ):
1823 self._invalid_transaction()
1825 elif self._trans_context_manager:
1826 TransactionalContext._trans_ctx_check(self)
1828 if self._is_future and self._transaction is None:
1829 self._autobegin()
1831 context.pre_exec()
1833 if dialect.use_setinputsizes:
1834 context._set_input_sizes()
1836 cursor, statement, parameters = (
1837 context.cursor,
1838 context.statement,
1839 context.parameters,
1840 )
1842 if not context.executemany:
1843 parameters = parameters[0]
1845 if self._has_events or self.engine._has_events:
1846 for fn in self.dispatch.before_cursor_execute:
1847 statement, parameters = fn(
1848 self,
1849 cursor,
1850 statement,
1851 parameters,
1852 context,
1853 context.executemany,
1854 )
1856 if self._echo:
1858 self._log_info(statement)
1860 stats = context._get_cache_stats()
1862 if not self.engine.hide_parameters:
1863 self._log_info(
1864 "[%s] %r",
1865 stats,
1866 sql_util._repr_params(
1867 parameters, batches=10, ismulti=context.executemany
1868 ),
1869 )
1870 else:
1871 self._log_info(
1872 "[%s] [SQL parameters hidden due to hide_parameters=True]"
1873 % (stats,)
1874 )
1876 evt_handled = False
1877 try:
1878 if context.executemany:
1879 if self.dialect._has_events:
1880 for fn in self.dialect.dispatch.do_executemany:
1881 if fn(cursor, statement, parameters, context):
1882 evt_handled = True
1883 break
1884 if not evt_handled:
1885 self.dialect.do_executemany(
1886 cursor, statement, parameters, context
1887 )
1888 elif not parameters and context.no_parameters:
1889 if self.dialect._has_events:
1890 for fn in self.dialect.dispatch.do_execute_no_params:
1891 if fn(cursor, statement, context):
1892 evt_handled = True
1893 break
1894 if not evt_handled:
1895 self.dialect.do_execute_no_params(
1896 cursor, statement, context
1897 )
1898 else:
1899 if self.dialect._has_events:
1900 for fn in self.dialect.dispatch.do_execute:
1901 if fn(cursor, statement, parameters, context):
1902 evt_handled = True
1903 break
1904 if not evt_handled:
1905 self.dialect.do_execute(
1906 cursor, statement, parameters, context
1907 )
1909 if self._has_events or self.engine._has_events:
1910 self.dispatch.after_cursor_execute(
1911 self,
1912 cursor,
1913 statement,
1914 parameters,
1915 context,
1916 context.executemany,
1917 )
1919 context.post_exec()
1921 result = context._setup_result_proxy()
1923 if not self._is_future:
1924 should_close_with_result = branched.should_close_with_result
1926 if not result._soft_closed and should_close_with_result:
1927 result._autoclose_connection = True
1929 if (
1930 # usually we're in a transaction so avoid relatively
1931 # expensive / legacy should_autocommit call
1932 self._transaction is None
1933 and context.should_autocommit
1934 ):
1935 self._commit_impl(autocommit=True)
1937 # for "connectionless" execution, we have to close this
1938 # Connection after the statement is complete.
1939 # legacy stuff.
1940 if should_close_with_result and context._soft_closed:
1941 assert not self._is_future
1943 # CursorResult already exhausted rows / has no rows.
1944 # close us now
1945 branched.close()
1947 except BaseException as e:
1948 self._handle_dbapi_exception(
1949 e, statement, parameters, cursor, context
1950 )
1952 return result
1954 def _cursor_execute(self, cursor, statement, parameters, context=None):
1955 """Execute a statement + params on the given cursor.
1957 Adds appropriate logging and exception handling.
1959 This method is used by DefaultDialect for special-case
1960 executions, such as for sequences and column defaults.
1961 The path of statement execution in the majority of cases
1962 terminates at _execute_context().
1964 """
1965 if self._has_events or self.engine._has_events:
1966 for fn in self.dispatch.before_cursor_execute:
1967 statement, parameters = fn(
1968 self, cursor, statement, parameters, context, False
1969 )
1971 if self._echo:
1972 self._log_info(statement)
1973 self._log_info("[raw sql] %r", parameters)
1974 try:
1975 for fn in (
1976 ()
1977 if not self.dialect._has_events
1978 else self.dialect.dispatch.do_execute
1979 ):
1980 if fn(cursor, statement, parameters, context):
1981 break
1982 else:
1983 self.dialect.do_execute(cursor, statement, parameters, context)
1984 except BaseException as e:
1985 self._handle_dbapi_exception(
1986 e, statement, parameters, cursor, context
1987 )
1989 if self._has_events or self.engine._has_events:
1990 self.dispatch.after_cursor_execute(
1991 self, cursor, statement, parameters, context, False
1992 )
1994 def _safe_close_cursor(self, cursor):
1995 """Close the given cursor, catching exceptions
1996 and turning into log warnings.
1998 """
1999 try:
2000 cursor.close()
2001 except Exception:
2002 # log the error through the connection pool's logger.
2003 self.engine.pool.logger.error(
2004 "Error closing cursor", exc_info=True
2005 )
2007 _reentrant_error = False
2008 _is_disconnect = False
2010 def _handle_dbapi_exception(
2011 self, e, statement, parameters, cursor, context
2012 ):
2013 exc_info = sys.exc_info()
2015 is_exit_exception = util.is_exit_exception(e)
2017 if not self._is_disconnect:
2018 self._is_disconnect = (
2019 isinstance(e, self.dialect.dbapi.Error)
2020 and not self.closed
2021 and self.dialect.is_disconnect(
2022 e,
2023 self._dbapi_connection if not self.invalidated else None,
2024 cursor,
2025 )
2026 ) or (is_exit_exception and not self.closed)
2028 invalidate_pool_on_disconnect = not is_exit_exception
2030 if self._reentrant_error:
2031 util.raise_(
2032 exc.DBAPIError.instance(
2033 statement,
2034 parameters,
2035 e,
2036 self.dialect.dbapi.Error,
2037 hide_parameters=self.engine.hide_parameters,
2038 dialect=self.dialect,
2039 ismulti=context.executemany
2040 if context is not None
2041 else None,
2042 ),
2043 with_traceback=exc_info[2],
2044 from_=e,
2045 )
2046 self._reentrant_error = True
2047 try:
2048 # non-DBAPI error - if we already got a context,
2049 # or there's no string statement, don't wrap it
2050 should_wrap = isinstance(e, self.dialect.dbapi.Error) or (
2051 statement is not None
2052 and context is None
2053 and not is_exit_exception
2054 )
2056 if should_wrap:
2057 sqlalchemy_exception = exc.DBAPIError.instance(
2058 statement,
2059 parameters,
2060 e,
2061 self.dialect.dbapi.Error,
2062 hide_parameters=self.engine.hide_parameters,
2063 connection_invalidated=self._is_disconnect,
2064 dialect=self.dialect,
2065 ismulti=context.executemany
2066 if context is not None
2067 else None,
2068 )
2069 else:
2070 sqlalchemy_exception = None
2072 newraise = None
2074 if (
2075 self._has_events or self.engine._has_events
2076 ) and not self._execution_options.get(
2077 "skip_user_error_events", False
2078 ):
2079 ctx = ExceptionContextImpl(
2080 e,
2081 sqlalchemy_exception,
2082 self.engine,
2083 self,
2084 cursor,
2085 statement,
2086 parameters,
2087 context,
2088 self._is_disconnect,
2089 invalidate_pool_on_disconnect,
2090 )
2092 for fn in self.dispatch.handle_error:
2093 try:
2094 # handler returns an exception;
2095 # call next handler in a chain
2096 per_fn = fn(ctx)
2097 if per_fn is not None:
2098 ctx.chained_exception = newraise = per_fn
2099 except Exception as _raised:
2100 # handler raises an exception - stop processing
2101 newraise = _raised
2102 break
2104 if self._is_disconnect != ctx.is_disconnect:
2105 self._is_disconnect = ctx.is_disconnect
2106 if sqlalchemy_exception:
2107 sqlalchemy_exception.connection_invalidated = (
2108 ctx.is_disconnect
2109 )
2111 # set up potentially user-defined value for
2112 # invalidate pool.
2113 invalidate_pool_on_disconnect = (
2114 ctx.invalidate_pool_on_disconnect
2115 )
2117 if should_wrap and context:
2118 context.handle_dbapi_exception(e)
2120 if not self._is_disconnect:
2121 if cursor:
2122 self._safe_close_cursor(cursor)
2123 with util.safe_reraise(warn_only=True):
2124 self._autorollback()
2126 if newraise:
2127 util.raise_(newraise, with_traceback=exc_info[2], from_=e)
2128 elif should_wrap:
2129 util.raise_(
2130 sqlalchemy_exception, with_traceback=exc_info[2], from_=e
2131 )
2132 else:
2133 util.raise_(exc_info[1], with_traceback=exc_info[2])
2135 finally:
2136 del self._reentrant_error
2137 if self._is_disconnect:
2138 del self._is_disconnect
2139 if not self.invalidated:
2140 dbapi_conn_wrapper = self._dbapi_connection
2141 if invalidate_pool_on_disconnect:
2142 self.engine.pool._invalidate(dbapi_conn_wrapper, e)
2143 self.invalidate(e)
2144 if self.should_close_with_result:
2145 assert not self._is_future
2146 self.close()
2148 @classmethod
2149 def _handle_dbapi_exception_noconnection(cls, e, dialect, engine):
2150 exc_info = sys.exc_info()
2152 is_disconnect = dialect.is_disconnect(e, None, None)
2154 should_wrap = isinstance(e, dialect.dbapi.Error)
2156 if should_wrap:
2157 sqlalchemy_exception = exc.DBAPIError.instance(
2158 None,
2159 None,
2160 e,
2161 dialect.dbapi.Error,
2162 hide_parameters=engine.hide_parameters,
2163 connection_invalidated=is_disconnect,
2164 )
2165 else:
2166 sqlalchemy_exception = None
2168 newraise = None
2170 if engine._has_events:
2171 ctx = ExceptionContextImpl(
2172 e,
2173 sqlalchemy_exception,
2174 engine,
2175 None,
2176 None,
2177 None,
2178 None,
2179 None,
2180 is_disconnect,
2181 True,
2182 )
2183 for fn in engine.dispatch.handle_error:
2184 try:
2185 # handler returns an exception;
2186 # call next handler in a chain
2187 per_fn = fn(ctx)
2188 if per_fn is not None:
2189 ctx.chained_exception = newraise = per_fn
2190 except Exception as _raised:
2191 # handler raises an exception - stop processing
2192 newraise = _raised
2193 break
2195 if sqlalchemy_exception and is_disconnect != ctx.is_disconnect:
2196 sqlalchemy_exception.connection_invalidated = (
2197 is_disconnect
2198 ) = ctx.is_disconnect
2200 if newraise:
2201 util.raise_(newraise, with_traceback=exc_info[2], from_=e)
2202 elif should_wrap:
2203 util.raise_(
2204 sqlalchemy_exception, with_traceback=exc_info[2], from_=e
2205 )
2206 else:
2207 util.raise_(exc_info[1], with_traceback=exc_info[2])
2209 def _run_ddl_visitor(self, visitorcallable, element, **kwargs):
2210 """run a DDL visitor.
2212 This method is only here so that the MockConnection can change the
2213 options given to the visitor so that "checkfirst" is skipped.
2215 """
2216 visitorcallable(self.dialect, self, **kwargs).traverse_single(element)
2218 @util.deprecated(
2219 "1.4",
2220 "The :meth:`_engine.Connection.transaction` "
2221 "method is deprecated and will be "
2222 "removed in a future release. Use the :meth:`_engine.Engine.begin` "
2223 "context manager instead.",
2224 )
2225 def transaction(self, callable_, *args, **kwargs):
2226 r"""Execute the given function within a transaction boundary.
2228 The function is passed this :class:`_engine.Connection`
2229 as the first argument, followed by the given \*args and \**kwargs,
2230 e.g.::
2232 def do_something(conn, x, y):
2233 conn.execute(text("some statement"), {'x':x, 'y':y})
2235 conn.transaction(do_something, 5, 10)
2237 The operations inside the function are all invoked within the
2238 context of a single :class:`.Transaction`.
2239 Upon success, the transaction is committed. If an
2240 exception is raised, the transaction is rolled back
2241 before propagating the exception.
2243 .. note::
2245 The :meth:`.transaction` method is superseded by
2246 the usage of the Python ``with:`` statement, which can
2247 be used with :meth:`_engine.Connection.begin`::
2249 with conn.begin():
2250 conn.execute(text("some statement"), {'x':5, 'y':10})
2252 As well as with :meth:`_engine.Engine.begin`::
2254 with engine.begin() as conn:
2255 conn.execute(text("some statement"), {'x':5, 'y':10})
2257 .. seealso::
2259 :meth:`_engine.Engine.begin` - engine-level transactional
2260 context
2262 :meth:`_engine.Engine.transaction` - engine-level version of
2263 :meth:`_engine.Connection.transaction`
2265 """
2267 kwargs["_sa_skip_warning"] = True
2268 trans = self.begin()
2269 try:
2270 ret = self.run_callable(callable_, *args, **kwargs)
2271 trans.commit()
2272 return ret
2273 except:
2274 with util.safe_reraise():
2275 trans.rollback()
2277 @util.deprecated(
2278 "1.4",
2279 "The :meth:`_engine.Connection.run_callable` "
2280 "method is deprecated and will "
2281 "be removed in a future release. Invoke the callable function "
2282 "directly, passing the Connection.",
2283 )
2284 def run_callable(self, callable_, *args, **kwargs):
2285 r"""Given a callable object or function, execute it, passing
2286 a :class:`_engine.Connection` as the first argument.
2288 The given \*args and \**kwargs are passed subsequent
2289 to the :class:`_engine.Connection` argument.
2291 This function, along with :meth:`_engine.Engine.run_callable`,
2292 allows a function to be run with a :class:`_engine.Connection`
2293 or :class:`_engine.Engine` object without the need to know
2294 which one is being dealt with.
2296 """
2297 return callable_(self, *args, **kwargs)
2300class ExceptionContextImpl(ExceptionContext):
2301 """Implement the :class:`.ExceptionContext` interface."""
2303 def __init__(
2304 self,
2305 exception,
2306 sqlalchemy_exception,
2307 engine,
2308 connection,
2309 cursor,
2310 statement,
2311 parameters,
2312 context,
2313 is_disconnect,
2314 invalidate_pool_on_disconnect,
2315 ):
2316 self.engine = engine
2317 self.connection = connection
2318 self.sqlalchemy_exception = sqlalchemy_exception
2319 self.original_exception = exception
2320 self.execution_context = context
2321 self.statement = statement
2322 self.parameters = parameters
2323 self.is_disconnect = is_disconnect
2324 self.invalidate_pool_on_disconnect = invalidate_pool_on_disconnect
2327class Transaction(TransactionalContext):
2328 """Represent a database transaction in progress.
2330 The :class:`.Transaction` object is procured by
2331 calling the :meth:`_engine.Connection.begin` method of
2332 :class:`_engine.Connection`::
2334 from sqlalchemy import create_engine
2335 engine = create_engine("postgresql://scott:tiger@localhost/test")
2336 connection = engine.connect()
2337 trans = connection.begin()
2338 connection.execute(text("insert into x (a, b) values (1, 2)"))
2339 trans.commit()
2341 The object provides :meth:`.rollback` and :meth:`.commit`
2342 methods in order to control transaction boundaries. It
2343 also implements a context manager interface so that
2344 the Python ``with`` statement can be used with the
2345 :meth:`_engine.Connection.begin` method::
2347 with connection.begin():
2348 connection.execute(text("insert into x (a, b) values (1, 2)"))
2350 The Transaction object is **not** threadsafe.
2352 .. seealso::
2354 :meth:`_engine.Connection.begin`
2356 :meth:`_engine.Connection.begin_twophase`
2358 :meth:`_engine.Connection.begin_nested`
2360 .. index::
2361 single: thread safety; Transaction
2362 """
2364 __slots__ = ()
2366 _is_root = False
2368 def __init__(self, connection):
2369 raise NotImplementedError()
2371 def _do_deactivate(self):
2372 """do whatever steps are necessary to set this transaction as
2373 "deactive", however leave this transaction object in place as far
2374 as the connection's state.
2376 for a "real" transaction this should roll back the transaction
2377 and ensure this transaction is no longer a reset agent.
2379 this is used for nesting of marker transactions where the marker
2380 can set the "real" transaction as rolled back, however it stays
2381 in place.
2383 for 2.0 we hope to remove this nesting feature.
2385 """
2386 raise NotImplementedError()
2388 @property
2389 def _deactivated_from_connection(self):
2390 """True if this transaction is totally deactivated from the connection
2391 and therefore can no longer affect its state.
2393 """
2394 raise NotImplementedError()
2396 def _do_close(self):
2397 raise NotImplementedError()
2399 def _do_rollback(self):
2400 raise NotImplementedError()
2402 def _do_commit(self):
2403 raise NotImplementedError()
2405 @property
2406 def is_valid(self):
2407 return self.is_active and not self.connection.invalidated
2409 def close(self):
2410 """Close this :class:`.Transaction`.
2412 If this transaction is the base transaction in a begin/commit
2413 nesting, the transaction will rollback(). Otherwise, the
2414 method returns.
2416 This is used to cancel a Transaction without affecting the scope of
2417 an enclosing transaction.
2419 """
2420 try:
2421 self._do_close()
2422 finally:
2423 assert not self.is_active
2425 def rollback(self):
2426 """Roll back this :class:`.Transaction`.
2428 The implementation of this may vary based on the type of transaction in
2429 use:
2431 * For a simple database transaction (e.g. :class:`.RootTransaction`),
2432 it corresponds to a ROLLBACK.
2434 * For a :class:`.NestedTransaction`, it corresponds to a
2435 "ROLLBACK TO SAVEPOINT" operation.
2437 * For a :class:`.TwoPhaseTransaction`, DBAPI-specific methods for two
2438 phase transactions may be used.
2441 """
2442 try:
2443 self._do_rollback()
2444 finally:
2445 assert not self.is_active
2447 def commit(self):
2448 """Commit this :class:`.Transaction`.
2450 The implementation of this may vary based on the type of transaction in
2451 use:
2453 * For a simple database transaction (e.g. :class:`.RootTransaction`),
2454 it corresponds to a COMMIT.
2456 * For a :class:`.NestedTransaction`, it corresponds to a
2457 "RELEASE SAVEPOINT" operation.
2459 * For a :class:`.TwoPhaseTransaction`, DBAPI-specific methods for two
2460 phase transactions may be used.
2462 """
2463 try:
2464 self._do_commit()
2465 finally:
2466 assert not self.is_active
2468 def _get_subject(self):
2469 return self.connection
2471 def _transaction_is_active(self):
2472 return self.is_active
2474 def _transaction_is_closed(self):
2475 return not self._deactivated_from_connection
2477 def _rollback_can_be_called(self):
2478 # for RootTransaction / NestedTransaction, it's safe to call
2479 # rollback() even if the transaction is deactive and no warnings
2480 # will be emitted. tested in
2481 # test_transaction.py -> test_no_rollback_in_deactive(?:_savepoint)?
2482 return True
2485class MarkerTransaction(Transaction):
2486 """A 'marker' transaction that is used for nested begin() calls.
2488 .. deprecated:: 1.4 future connection for 2.0 won't support this pattern.
2490 """
2492 __slots__ = ("connection", "_is_active", "_transaction")
2494 def __init__(self, connection):
2495 assert connection._transaction is not None
2496 if not connection._transaction.is_active:
2497 raise exc.InvalidRequestError(
2498 "the current transaction on this connection is inactive. "
2499 "Please issue a rollback first."
2500 )
2502 assert not connection._is_future
2503 util.warn_deprecated_20(
2504 "Calling .begin() when a transaction is already begun, creating "
2505 "a 'sub' transaction, is deprecated "
2506 "and will be removed in 2.0. See the documentation section "
2507 "'Migrating from the nesting pattern' for background on how "
2508 "to migrate from this pattern."
2509 )
2511 self.connection = connection
2513 if connection._trans_context_manager:
2514 TransactionalContext._trans_ctx_check(connection)
2516 if connection._nested_transaction is not None:
2517 self._transaction = connection._nested_transaction
2518 else:
2519 self._transaction = connection._transaction
2520 self._is_active = True
2522 @property
2523 def _deactivated_from_connection(self):
2524 return not self.is_active
2526 @property
2527 def is_active(self):
2528 return self._is_active and self._transaction.is_active
2530 def _deactivate(self):
2531 self._is_active = False
2533 def _do_close(self):
2534 # does not actually roll back the root
2535 self._deactivate()
2537 def _do_rollback(self):
2538 # does roll back the root
2539 if self._is_active:
2540 try:
2541 self._transaction._do_deactivate()
2542 finally:
2543 self._deactivate()
2545 def _do_commit(self):
2546 self._deactivate()
2549class RootTransaction(Transaction):
2550 """Represent the "root" transaction on a :class:`_engine.Connection`.
2552 This corresponds to the current "BEGIN/COMMIT/ROLLBACK" that's occurring
2553 for the :class:`_engine.Connection`. The :class:`_engine.RootTransaction`
2554 is created by calling upon the :meth:`_engine.Connection.begin` method, and
2555 remains associated with the :class:`_engine.Connection` throughout its
2556 active span. The current :class:`_engine.RootTransaction` in use is
2557 accessible via the :attr:`_engine.Connection.get_transaction` method of
2558 :class:`_engine.Connection`.
2560 In :term:`2.0 style` use, the :class:`_future.Connection` also employs
2561 "autobegin" behavior that will create a new
2562 :class:`_engine.RootTransaction` whenever a connection in a
2563 non-transactional state is used to emit commands on the DBAPI connection.
2564 The scope of the :class:`_engine.RootTransaction` in 2.0 style
2565 use can be controlled using the :meth:`_future.Connection.commit` and
2566 :meth:`_future.Connection.rollback` methods.
2569 """
2571 _is_root = True
2573 __slots__ = ("connection", "is_active")
2575 def __init__(self, connection):
2576 assert connection._transaction is None
2577 if connection._trans_context_manager:
2578 TransactionalContext._trans_ctx_check(connection)
2579 self.connection = connection
2580 self._connection_begin_impl()
2581 connection._transaction = self
2583 self.is_active = True
2585 def _deactivate_from_connection(self):
2586 if self.is_active:
2587 assert self.connection._transaction is self
2588 self.is_active = False
2590 elif self.connection._transaction is not self:
2591 util.warn("transaction already deassociated from connection")
2593 @property
2594 def _deactivated_from_connection(self):
2595 return self.connection._transaction is not self
2597 def _do_deactivate(self):
2598 # called from a MarkerTransaction to cancel this root transaction.
2599 # the transaction stays in place as connection._transaction, but
2600 # is no longer active and is no longer the reset agent for the
2601 # pooled connection. the connection won't support a new begin()
2602 # until this transaction is explicitly closed, rolled back,
2603 # or committed.
2605 assert self.connection._transaction is self
2607 if self.is_active:
2608 self._connection_rollback_impl()
2610 # handle case where a savepoint was created inside of a marker
2611 # transaction that refers to a root. nested has to be cancelled
2612 # also.
2613 if self.connection._nested_transaction:
2614 self.connection._nested_transaction._cancel()
2616 self._deactivate_from_connection()
2618 def _connection_begin_impl(self):
2619 self.connection._begin_impl(self)
2621 def _connection_rollback_impl(self):
2622 self.connection._rollback_impl()
2624 def _connection_commit_impl(self):
2625 self.connection._commit_impl()
2627 def _close_impl(self, try_deactivate=False):
2628 try:
2629 if self.is_active:
2630 self._connection_rollback_impl()
2632 if self.connection._nested_transaction:
2633 self.connection._nested_transaction._cancel()
2634 finally:
2635 if self.is_active or try_deactivate:
2636 self._deactivate_from_connection()
2637 if self.connection._transaction is self:
2638 self.connection._transaction = None
2640 assert not self.is_active
2641 assert self.connection._transaction is not self
2643 def _do_close(self):
2644 self._close_impl()
2646 def _do_rollback(self):
2647 self._close_impl(try_deactivate=True)
2649 def _do_commit(self):
2650 if self.is_active:
2651 assert self.connection._transaction is self
2653 try:
2654 self._connection_commit_impl()
2655 finally:
2656 # whether or not commit succeeds, cancel any
2657 # nested transactions, make this transaction "inactive"
2658 # and remove it as a reset agent
2659 if self.connection._nested_transaction:
2660 self.connection._nested_transaction._cancel()
2662 self._deactivate_from_connection()
2664 # ...however only remove as the connection's current transaction
2665 # if commit succeeded. otherwise it stays on so that a rollback
2666 # needs to occur.
2667 self.connection._transaction = None
2668 else:
2669 if self.connection._transaction is self:
2670 self.connection._invalid_transaction()
2671 else:
2672 raise exc.InvalidRequestError("This transaction is inactive")
2674 assert not self.is_active
2675 assert self.connection._transaction is not self
2678class NestedTransaction(Transaction):
2679 """Represent a 'nested', or SAVEPOINT transaction.
2681 The :class:`.NestedTransaction` object is created by calling the
2682 :meth:`_engine.Connection.begin_nested` method of
2683 :class:`_engine.Connection`.
2685 When using :class:`.NestedTransaction`, the semantics of "begin" /
2686 "commit" / "rollback" are as follows:
2688 * the "begin" operation corresponds to the "BEGIN SAVEPOINT" command, where
2689 the savepoint is given an explicit name that is part of the state
2690 of this object.
2692 * The :meth:`.NestedTransaction.commit` method corresponds to a
2693 "RELEASE SAVEPOINT" operation, using the savepoint identifier associated
2694 with this :class:`.NestedTransaction`.
2696 * The :meth:`.NestedTransaction.rollback` method corresponds to a
2697 "ROLLBACK TO SAVEPOINT" operation, using the savepoint identifier
2698 associated with this :class:`.NestedTransaction`.
2700 The rationale for mimicking the semantics of an outer transaction in
2701 terms of savepoints so that code may deal with a "savepoint" transaction
2702 and an "outer" transaction in an agnostic way.
2704 .. seealso::
2706 :ref:`session_begin_nested` - ORM version of the SAVEPOINT API.
2708 """
2710 __slots__ = ("connection", "is_active", "_savepoint", "_previous_nested")
2712 def __init__(self, connection):
2713 assert connection._transaction is not None
2714 if connection._trans_context_manager:
2715 TransactionalContext._trans_ctx_check(connection)
2716 self.connection = connection
2717 self._savepoint = self.connection._savepoint_impl()
2718 self.is_active = True
2719 self._previous_nested = connection._nested_transaction
2720 connection._nested_transaction = self
2722 def _deactivate_from_connection(self, warn=True):
2723 if self.connection._nested_transaction is self:
2724 self.connection._nested_transaction = self._previous_nested
2725 elif warn:
2726 util.warn(
2727 "nested transaction already deassociated from connection"
2728 )
2730 @property
2731 def _deactivated_from_connection(self):
2732 return self.connection._nested_transaction is not self
2734 def _cancel(self):
2735 # called by RootTransaction when the outer transaction is
2736 # committed, rolled back, or closed to cancel all savepoints
2737 # without any action being taken
2738 self.is_active = False
2739 self._deactivate_from_connection()
2740 if self._previous_nested:
2741 self._previous_nested._cancel()
2743 def _close_impl(self, deactivate_from_connection, warn_already_deactive):
2744 try:
2745 if self.is_active and self.connection._transaction.is_active:
2746 self.connection._rollback_to_savepoint_impl(self._savepoint)
2747 finally:
2748 self.is_active = False
2750 if deactivate_from_connection:
2751 self._deactivate_from_connection(warn=warn_already_deactive)
2753 assert not self.is_active
2754 if deactivate_from_connection:
2755 assert self.connection._nested_transaction is not self
2757 def _do_deactivate(self):
2758 self._close_impl(False, False)
2760 def _do_close(self):
2761 self._close_impl(True, False)
2763 def _do_rollback(self):
2764 self._close_impl(True, True)
2766 def _do_commit(self):
2767 if self.is_active:
2768 try:
2769 self.connection._release_savepoint_impl(self._savepoint)
2770 finally:
2771 # nested trans becomes inactive on failed release
2772 # unconditionally. this prevents it from trying to
2773 # emit SQL when it rolls back.
2774 self.is_active = False
2776 # but only de-associate from connection if it succeeded
2777 self._deactivate_from_connection()
2778 else:
2779 if self.connection._nested_transaction is self:
2780 self.connection._invalid_transaction()
2781 else:
2782 raise exc.InvalidRequestError(
2783 "This nested transaction is inactive"
2784 )
2787class TwoPhaseTransaction(RootTransaction):
2788 """Represent a two-phase transaction.
2790 A new :class:`.TwoPhaseTransaction` object may be procured
2791 using the :meth:`_engine.Connection.begin_twophase` method.
2793 The interface is the same as that of :class:`.Transaction`
2794 with the addition of the :meth:`prepare` method.
2796 """
2798 __slots__ = ("connection", "is_active", "xid", "_is_prepared")
2800 def __init__(self, connection, xid):
2801 self._is_prepared = False
2802 self.xid = xid
2803 super(TwoPhaseTransaction, self).__init__(connection)
2805 def prepare(self):
2806 """Prepare this :class:`.TwoPhaseTransaction`.
2808 After a PREPARE, the transaction can be committed.
2810 """
2811 if not self.is_active:
2812 raise exc.InvalidRequestError("This transaction is inactive")
2813 self.connection._prepare_twophase_impl(self.xid)
2814 self._is_prepared = True
2816 def _connection_begin_impl(self):
2817 self.connection._begin_twophase_impl(self)
2819 def _connection_rollback_impl(self):
2820 self.connection._rollback_twophase_impl(self.xid, self._is_prepared)
2822 def _connection_commit_impl(self):
2823 self.connection._commit_twophase_impl(self.xid, self._is_prepared)
2826class Engine(Connectable, log.Identified):
2827 """
2828 Connects a :class:`~sqlalchemy.pool.Pool` and
2829 :class:`~sqlalchemy.engine.interfaces.Dialect` together to provide a
2830 source of database connectivity and behavior.
2832 This is the **SQLAlchemy 1.x version** of :class:`_engine.Engine`. For
2833 the :term:`2.0 style` version, which includes some API differences,
2834 see :class:`_future.Engine`.
2836 An :class:`_engine.Engine` object is instantiated publicly using the
2837 :func:`~sqlalchemy.create_engine` function.
2839 .. seealso::
2841 :doc:`/core/engines`
2843 :ref:`connections_toplevel`
2845 """
2847 _execution_options = _EMPTY_EXECUTION_OPTS
2848 _has_events = False
2849 _connection_cls = Connection
2850 _sqla_logger_namespace = "sqlalchemy.engine.Engine"
2851 _is_future = False
2853 _schema_translate_map = None
2855 def __init__(
2856 self,
2857 pool,
2858 dialect,
2859 url,
2860 logging_name=None,
2861 echo=None,
2862 query_cache_size=500,
2863 execution_options=None,
2864 hide_parameters=False,
2865 ):
2866 self.pool = pool
2867 self.url = url
2868 self.dialect = dialect
2869 if logging_name:
2870 self.logging_name = logging_name
2871 self.echo = echo
2872 self.hide_parameters = hide_parameters
2873 if query_cache_size != 0:
2874 self._compiled_cache = util.LRUCache(
2875 query_cache_size, size_alert=self._lru_size_alert
2876 )
2877 else:
2878 self._compiled_cache = None
2879 log.instance_logger(self, echoflag=echo)
2880 if execution_options:
2881 self.update_execution_options(**execution_options)
2883 def _lru_size_alert(self, cache):
2884 if self._should_log_info:
2885 self.logger.info(
2886 "Compiled cache size pruning from %d items to %d. "
2887 "Increase cache size to reduce the frequency of pruning.",
2888 len(cache),
2889 cache.capacity,
2890 )
2892 @property
2893 def engine(self):
2894 return self
2896 def clear_compiled_cache(self):
2897 """Clear the compiled cache associated with the dialect.
2899 This applies **only** to the built-in cache that is established
2900 via the :paramref:`_engine.create_engine.query_cache_size` parameter.
2901 It will not impact any dictionary caches that were passed via the
2902 :paramref:`.Connection.execution_options.query_cache` parameter.
2904 .. versionadded:: 1.4
2906 """
2907 if self._compiled_cache:
2908 self._compiled_cache.clear()
2910 def update_execution_options(self, **opt):
2911 r"""Update the default execution_options dictionary
2912 of this :class:`_engine.Engine`.
2914 The given keys/values in \**opt are added to the
2915 default execution options that will be used for
2916 all connections. The initial contents of this dictionary
2917 can be sent via the ``execution_options`` parameter
2918 to :func:`_sa.create_engine`.
2920 .. seealso::
2922 :meth:`_engine.Connection.execution_options`
2924 :meth:`_engine.Engine.execution_options`
2926 """
2927 self._execution_options = self._execution_options.union(opt)
2928 self.dispatch.set_engine_execution_options(self, opt)
2929 self.dialect.set_engine_execution_options(self, opt)
2931 def execution_options(self, **opt):
2932 """Return a new :class:`_engine.Engine` that will provide
2933 :class:`_engine.Connection` objects with the given execution options.
2935 The returned :class:`_engine.Engine` remains related to the original
2936 :class:`_engine.Engine` in that it shares the same connection pool and
2937 other state:
2939 * The :class:`_pool.Pool` used by the new :class:`_engine.Engine`
2940 is the
2941 same instance. The :meth:`_engine.Engine.dispose`
2942 method will replace
2943 the connection pool instance for the parent engine as well
2944 as this one.
2945 * Event listeners are "cascaded" - meaning, the new
2946 :class:`_engine.Engine`
2947 inherits the events of the parent, and new events can be associated
2948 with the new :class:`_engine.Engine` individually.
2949 * The logging configuration and logging_name is copied from the parent
2950 :class:`_engine.Engine`.
2952 The intent of the :meth:`_engine.Engine.execution_options` method is
2953 to implement "sharding" schemes where multiple :class:`_engine.Engine`
2954 objects refer to the same connection pool, but are differentiated
2955 by options that would be consumed by a custom event::
2957 primary_engine = create_engine("mysql://")
2958 shard1 = primary_engine.execution_options(shard_id="shard1")
2959 shard2 = primary_engine.execution_options(shard_id="shard2")
2961 Above, the ``shard1`` engine serves as a factory for
2962 :class:`_engine.Connection`
2963 objects that will contain the execution option
2964 ``shard_id=shard1``, and ``shard2`` will produce
2965 :class:`_engine.Connection`
2966 objects that contain the execution option ``shard_id=shard2``.
2968 An event handler can consume the above execution option to perform
2969 a schema switch or other operation, given a connection. Below
2970 we emit a MySQL ``use`` statement to switch databases, at the same
2971 time keeping track of which database we've established using the
2972 :attr:`_engine.Connection.info` dictionary,
2973 which gives us a persistent
2974 storage space that follows the DBAPI connection::
2976 from sqlalchemy import event
2977 from sqlalchemy.engine import Engine
2979 shards = {"default": "base", shard_1: "db1", "shard_2": "db2"}
2981 @event.listens_for(Engine, "before_cursor_execute")
2982 def _switch_shard(conn, cursor, stmt,
2983 params, context, executemany):
2984 shard_id = conn._execution_options.get('shard_id', "default")
2985 current_shard = conn.info.get("current_shard", None)
2987 if current_shard != shard_id:
2988 cursor.execute("use %s" % shards[shard_id])
2989 conn.info["current_shard"] = shard_id
2991 .. seealso::
2993 :meth:`_engine.Connection.execution_options`
2994 - update execution options
2995 on a :class:`_engine.Connection` object.
2997 :meth:`_engine.Engine.update_execution_options`
2998 - update the execution
2999 options for a given :class:`_engine.Engine` in place.
3001 :meth:`_engine.Engine.get_execution_options`
3004 """
3005 return self._option_cls(self, opt)
3007 def get_execution_options(self):
3008 """Get the non-SQL options which will take effect during execution.
3010 .. versionadded: 1.3
3012 .. seealso::
3014 :meth:`_engine.Engine.execution_options`
3015 """
3016 return self._execution_options
3018 @property
3019 def name(self):
3020 """String name of the :class:`~sqlalchemy.engine.interfaces.Dialect`
3021 in use by this :class:`Engine`."""
3023 return self.dialect.name
3025 @property
3026 def driver(self):
3027 """Driver name of the :class:`~sqlalchemy.engine.interfaces.Dialect`
3028 in use by this :class:`Engine`."""
3030 return self.dialect.driver
3032 echo = log.echo_property()
3034 def __repr__(self):
3035 return "Engine(%r)" % (self.url,)
3037 def dispose(self, close=True):
3038 """Dispose of the connection pool used by this
3039 :class:`_engine.Engine`.
3041 A new connection pool is created immediately after the old one has been
3042 disposed. The previous connection pool is disposed either actively, by
3043 closing out all currently checked-in connections in that pool, or
3044 passively, by losing references to it but otherwise not closing any
3045 connections. The latter strategy is more appropriate for an initializer
3046 in a forked Python process.
3048 :param close: if left at its default of ``True``, has the
3049 effect of fully closing all **currently checked in**
3050 database connections. Connections that are still checked out
3051 will **not** be closed, however they will no longer be associated
3052 with this :class:`_engine.Engine`,
3053 so when they are closed individually, eventually the
3054 :class:`_pool.Pool` which they are associated with will
3055 be garbage collected and they will be closed out fully, if
3056 not already closed on checkin.
3058 If set to ``False``, the previous connection pool is de-referenced,
3059 and otherwise not touched in any way.
3061 .. versionadded:: 1.4.33 Added the :paramref:`.Engine.dispose.close`
3062 parameter to allow the replacement of a connection pool in a child
3063 process without interfering with the connections used by the parent
3064 process.
3067 .. seealso::
3069 :ref:`engine_disposal`
3071 :ref:`pooling_multiprocessing`
3073 """
3074 if close:
3075 self.pool.dispose()
3076 self.pool = self.pool.recreate()
3077 self.dispatch.engine_disposed(self)
3079 def _execute_default(
3080 self, default, multiparams=(), params=util.EMPTY_DICT
3081 ):
3082 with self.connect() as conn:
3083 return conn._execute_default(default, multiparams, params)
3085 @contextlib.contextmanager
3086 def _optional_conn_ctx_manager(self, connection=None):
3087 if connection is None:
3088 with self.connect() as conn:
3089 yield conn
3090 else:
3091 yield connection
3093 class _trans_ctx(object):
3094 def __init__(self, conn, transaction, close_with_result):
3095 self.conn = conn
3096 self.transaction = transaction
3097 self.close_with_result = close_with_result
3099 def __enter__(self):
3100 self.transaction.__enter__()
3101 return self.conn
3103 def __exit__(self, type_, value, traceback):
3104 try:
3105 self.transaction.__exit__(type_, value, traceback)
3106 finally:
3107 if not self.close_with_result:
3108 self.conn.close()
3110 def begin(self, close_with_result=False):
3111 """Return a context manager delivering a :class:`_engine.Connection`
3112 with a :class:`.Transaction` established.
3114 E.g.::
3116 with engine.begin() as conn:
3117 conn.execute(
3118 text("insert into table (x, y, z) values (1, 2, 3)")
3119 )
3120 conn.execute(text("my_special_procedure(5)"))
3122 Upon successful operation, the :class:`.Transaction`
3123 is committed. If an error is raised, the :class:`.Transaction`
3124 is rolled back.
3126 Legacy use only: the ``close_with_result`` flag is normally ``False``,
3127 and indicates that the :class:`_engine.Connection` will be closed when
3128 the operation is complete. When set to ``True``, it indicates the
3129 :class:`_engine.Connection` is in "single use" mode, where the
3130 :class:`_engine.CursorResult` returned by the first call to
3131 :meth:`_engine.Connection.execute` will close the
3132 :class:`_engine.Connection` when that :class:`_engine.CursorResult` has
3133 exhausted all result rows.
3135 .. seealso::
3137 :meth:`_engine.Engine.connect` - procure a
3138 :class:`_engine.Connection` from
3139 an :class:`_engine.Engine`.
3141 :meth:`_engine.Connection.begin` - start a :class:`.Transaction`
3142 for a particular :class:`_engine.Connection`.
3144 """
3145 if self._connection_cls._is_future:
3146 conn = self.connect()
3147 else:
3148 conn = self.connect(close_with_result=close_with_result)
3149 try:
3150 trans = conn.begin()
3151 except:
3152 with util.safe_reraise():
3153 conn.close()
3154 return Engine._trans_ctx(conn, trans, close_with_result)
3156 @util.deprecated(
3157 "1.4",
3158 "The :meth:`_engine.Engine.transaction` "
3159 "method is deprecated and will be "
3160 "removed in a future release. Use the :meth:`_engine.Engine.begin` "
3161 "context "
3162 "manager instead.",
3163 )
3164 def transaction(self, callable_, *args, **kwargs):
3165 r"""Execute the given function within a transaction boundary.
3167 The function is passed a :class:`_engine.Connection` newly procured
3168 from :meth:`_engine.Engine.connect` as the first argument,
3169 followed by the given \*args and \**kwargs.
3171 e.g.::
3173 def do_something(conn, x, y):
3174 conn.execute(text("some statement"), {'x':x, 'y':y})
3176 engine.transaction(do_something, 5, 10)
3178 The operations inside the function are all invoked within the
3179 context of a single :class:`.Transaction`.
3180 Upon success, the transaction is committed. If an
3181 exception is raised, the transaction is rolled back
3182 before propagating the exception.
3184 .. note::
3186 The :meth:`.transaction` method is superseded by
3187 the usage of the Python ``with:`` statement, which can
3188 be used with :meth:`_engine.Engine.begin`::
3190 with engine.begin() as conn:
3191 conn.execute(text("some statement"), {'x':5, 'y':10})
3193 .. seealso::
3195 :meth:`_engine.Engine.begin` - engine-level transactional
3196 context
3198 :meth:`_engine.Connection.transaction`
3199 - connection-level version of
3200 :meth:`_engine.Engine.transaction`
3202 """
3203 kwargs["_sa_skip_warning"] = True
3204 with self.connect() as conn:
3205 return conn.transaction(callable_, *args, **kwargs)
3207 @util.deprecated(
3208 "1.4",
3209 "The :meth:`_engine.Engine.run_callable` "
3210 "method is deprecated and will be "
3211 "removed in a future release. Use the :meth:`_engine.Engine.begin` "
3212 "context manager instead.",
3213 )
3214 def run_callable(self, callable_, *args, **kwargs):
3215 r"""Given a callable object or function, execute it, passing
3216 a :class:`_engine.Connection` as the first argument.
3218 The given \*args and \**kwargs are passed subsequent
3219 to the :class:`_engine.Connection` argument.
3221 This function, along with :meth:`_engine.Connection.run_callable`,
3222 allows a function to be run with a :class:`_engine.Connection`
3223 or :class:`_engine.Engine` object without the need to know
3224 which one is being dealt with.
3226 """
3227 kwargs["_sa_skip_warning"] = True
3228 with self.connect() as conn:
3229 return conn.run_callable(callable_, *args, **kwargs)
3231 def _run_ddl_visitor(self, visitorcallable, element, **kwargs):
3232 with self.begin() as conn:
3233 conn._run_ddl_visitor(visitorcallable, element, **kwargs)
3235 @util.deprecated_20(
3236 ":meth:`_engine.Engine.execute`",
3237 alternative="All statement execution in SQLAlchemy 2.0 is performed "
3238 "by the :meth:`_engine.Connection.execute` method of "
3239 ":class:`_engine.Connection`, "
3240 "or in the ORM by the :meth:`.Session.execute` method of "
3241 ":class:`.Session`.",
3242 )
3243 def execute(self, statement, *multiparams, **params):
3244 """Executes the given construct and returns a
3245 :class:`_engine.CursorResult`.
3247 The arguments are the same as those used by
3248 :meth:`_engine.Connection.execute`.
3250 Here, a :class:`_engine.Connection` is acquired using the
3251 :meth:`_engine.Engine.connect` method, and the statement executed
3252 with that connection. The returned :class:`_engine.CursorResult`
3253 is flagged
3254 such that when the :class:`_engine.CursorResult` is exhausted and its
3255 underlying cursor is closed, the :class:`_engine.Connection`
3256 created here
3257 will also be closed, which allows its associated DBAPI connection
3258 resource to be returned to the connection pool.
3260 """
3261 connection = self.connect(close_with_result=True)
3262 return connection.execute(statement, *multiparams, **params)
3264 @util.deprecated_20(
3265 ":meth:`_engine.Engine.scalar`",
3266 alternative="All statement execution in SQLAlchemy 2.0 is performed "
3267 "by the :meth:`_engine.Connection.execute` method of "
3268 ":class:`_engine.Connection`, "
3269 "or in the ORM by the :meth:`.Session.execute` method of "
3270 ":class:`.Session`; the :meth:`_future.Result.scalar` "
3271 "method can then be "
3272 "used to return a scalar result.",
3273 )
3274 def scalar(self, statement, *multiparams, **params):
3275 """Executes and returns the first column of the first row.
3277 The underlying result/cursor is closed after execution.
3278 """
3279 return self.execute(statement, *multiparams, **params).scalar()
3281 def _execute_clauseelement(
3282 self,
3283 elem,
3284 multiparams=None,
3285 params=None,
3286 execution_options=_EMPTY_EXECUTION_OPTS,
3287 ):
3288 connection = self.connect(close_with_result=True)
3289 return connection._execute_clauseelement(
3290 elem, multiparams, params, execution_options
3291 )
3293 def _execute_compiled(
3294 self,
3295 compiled,
3296 multiparams,
3297 params,
3298 execution_options=_EMPTY_EXECUTION_OPTS,
3299 ):
3300 connection = self.connect(close_with_result=True)
3301 return connection._execute_compiled(
3302 compiled, multiparams, params, execution_options
3303 )
3305 def connect(self, close_with_result=False):
3306 """Return a new :class:`_engine.Connection` object.
3308 The :class:`_engine.Connection` object is a facade that uses a DBAPI
3309 connection internally in order to communicate with the database. This
3310 connection is procured from the connection-holding :class:`_pool.Pool`
3311 referenced by this :class:`_engine.Engine`. When the
3312 :meth:`_engine.Connection.close` method of the
3313 :class:`_engine.Connection` object
3314 is called, the underlying DBAPI connection is then returned to the
3315 connection pool, where it may be used again in a subsequent call to
3316 :meth:`_engine.Engine.connect`.
3318 """
3320 return self._connection_cls(self, close_with_result=close_with_result)
3322 @util.deprecated(
3323 "1.4",
3324 "The :meth:`_engine.Engine.table_names` "
3325 "method is deprecated and will be "
3326 "removed in a future release. Please refer to "
3327 ":meth:`_reflection.Inspector.get_table_names`.",
3328 )
3329 def table_names(self, schema=None, connection=None):
3330 """Return a list of all table names available in the database.
3332 :param schema: Optional, retrieve names from a non-default schema.
3334 :param connection: Optional, use a specified connection.
3335 """
3336 with self._optional_conn_ctx_manager(connection) as conn:
3337 insp = inspection.inspect(conn)
3338 return insp.get_table_names(schema)
3340 @util.deprecated(
3341 "1.4",
3342 "The :meth:`_engine.Engine.has_table` "
3343 "method is deprecated and will be "
3344 "removed in a future release. Please refer to "
3345 ":meth:`_reflection.Inspector.has_table`.",
3346 )
3347 def has_table(self, table_name, schema=None):
3348 """Return True if the given backend has a table of the given name.
3350 .. seealso::
3352 :ref:`metadata_reflection_inspector` - detailed schema inspection
3353 using the :class:`_reflection.Inspector` interface.
3355 :class:`.quoted_name` - used to pass quoting information along
3356 with a schema identifier.
3358 """
3359 with self._optional_conn_ctx_manager(None) as conn:
3360 insp = inspection.inspect(conn)
3361 return insp.has_table(table_name, schema=schema)
3363 def _wrap_pool_connect(self, fn, connection):
3364 dialect = self.dialect
3365 try:
3366 return fn()
3367 except dialect.dbapi.Error as e:
3368 if connection is None:
3369 Connection._handle_dbapi_exception_noconnection(
3370 e, dialect, self
3371 )
3372 else:
3373 util.raise_(
3374 sys.exc_info()[1], with_traceback=sys.exc_info()[2]
3375 )
3377 def raw_connection(self, _connection=None):
3378 """Return a "raw" DBAPI connection from the connection pool.
3380 The returned object is a proxied version of the DBAPI
3381 connection object used by the underlying driver in use.
3382 The object will have all the same behavior as the real DBAPI
3383 connection, except that its ``close()`` method will result in the
3384 connection being returned to the pool, rather than being closed
3385 for real.
3387 This method provides direct DBAPI connection access for
3388 special situations when the API provided by
3389 :class:`_engine.Connection`
3390 is not needed. When a :class:`_engine.Connection` object is already
3391 present, the DBAPI connection is available using
3392 the :attr:`_engine.Connection.connection` accessor.
3394 .. seealso::
3396 :ref:`dbapi_connections`
3398 """
3399 return self._wrap_pool_connect(self.pool.connect, _connection)
3402class OptionEngineMixin(object):
3403 _sa_propagate_class_events = False
3405 def __init__(self, proxied, execution_options):
3406 self._proxied = proxied
3407 self.url = proxied.url
3408 self.dialect = proxied.dialect
3409 self.logging_name = proxied.logging_name
3410 self.echo = proxied.echo
3411 self._compiled_cache = proxied._compiled_cache
3412 self.hide_parameters = proxied.hide_parameters
3413 log.instance_logger(self, echoflag=self.echo)
3415 # note: this will propagate events that are assigned to the parent
3416 # engine after this OptionEngine is created. Since we share
3417 # the events of the parent we also disallow class-level events
3418 # to apply to the OptionEngine class directly.
3419 #
3420 # the other way this can work would be to transfer existing
3421 # events only, using:
3422 # self.dispatch._update(proxied.dispatch)
3423 #
3424 # that might be more appropriate however it would be a behavioral
3425 # change for logic that assigns events to the parent engine and
3426 # would like it to take effect for the already-created sub-engine.
3427 self.dispatch = self.dispatch._join(proxied.dispatch)
3429 self._execution_options = proxied._execution_options
3430 self.update_execution_options(**execution_options)
3432 def _get_pool(self):
3433 return self._proxied.pool
3435 def _set_pool(self, pool):
3436 self._proxied.pool = pool
3438 pool = property(_get_pool, _set_pool)
3440 def _get_has_events(self):
3441 return self._proxied._has_events or self.__dict__.get(
3442 "_has_events", False
3443 )
3445 def _set_has_events(self, value):
3446 self.__dict__["_has_events"] = value
3448 _has_events = property(_get_has_events, _set_has_events)
3451class OptionEngine(OptionEngineMixin, Engine):
3452 pass
3455Engine._option_cls = OptionEngine