Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/pool/base.py: 56%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# pool/base.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
9"""Base constructs for connection pools."""
11from __future__ import annotations
13from collections import deque
14import dataclasses
15from enum import Enum
16import threading
17import time
18import typing
19from typing import Any
20from typing import Callable
21from typing import cast
22from typing import Deque
23from typing import List
24from typing import Literal
25from typing import Optional
26from typing import Protocol
27from typing import Tuple
28from typing import TYPE_CHECKING
29from typing import Union
30import weakref
32from .. import event
33from .. import exc
34from .. import log
35from .. import util
36from ..util.typing import Self
38if TYPE_CHECKING:
39 from ..engine.interfaces import DBAPIConnection
40 from ..engine.interfaces import DBAPICursor
41 from ..engine.interfaces import Dialect
42 from ..event import _DispatchCommon
43 from ..event import _ListenerFnType
44 from ..event import dispatcher
45 from ..sql._typing import _InfoType
48@dataclasses.dataclass(frozen=True)
49class PoolResetState:
50 """describes the state of a DBAPI connection as it is being passed to
51 the :meth:`.PoolEvents.reset` connection pool event.
53 .. versionadded:: 2.0.0b3
55 """
57 __slots__ = ("transaction_was_reset", "terminate_only", "asyncio_safe")
59 transaction_was_reset: bool
60 """Indicates if the transaction on the DBAPI connection was already
61 essentially "reset" back by the :class:`.Connection` object.
63 This boolean is True if the :class:`.Connection` had transactional
64 state present upon it, which was then not closed using the
65 :meth:`.Connection.rollback` or :meth:`.Connection.commit` method;
66 instead, the transaction was closed inline within the
67 :meth:`.Connection.close` method so is guaranteed to remain non-present
68 when this event is reached.
70 """
72 terminate_only: bool
73 """indicates if the connection is to be immediately terminated and
74 not checked in to the pool.
76 This occurs for connections that were invalidated, as well as asyncio
77 connections that were not cleanly handled by the calling code that
78 are instead being garbage collected. In the latter case,
79 operations can't be safely run on asyncio connections within garbage
80 collection as there is not necessarily an event loop present.
82 """
84 asyncio_safe: bool
85 """Indicates if the reset operation is occurring within a scope where
86 an enclosing event loop is expected to be present for asyncio applications.
88 Will be False in the case that the connection is being garbage collected.
90 """
93class ResetStyle(Enum):
94 """Describe options for "reset on return" behaviors."""
96 reset_rollback = 0
97 reset_commit = 1
98 reset_none = 2
101_ResetStyleArgType = Union[
102 ResetStyle,
103 Literal[True, None, False, "commit", "rollback"],
104]
105reset_rollback, reset_commit, reset_none = list(ResetStyle)
108class _ConnDialect:
109 """partial implementation of :class:`.Dialect`
110 which provides DBAPI connection methods.
112 When a :class:`_pool.Pool` is combined with an :class:`_engine.Engine`,
113 the :class:`_engine.Engine` replaces this with its own
114 :class:`.Dialect`.
116 """
118 is_async = False
119 has_terminate = False
121 def do_rollback(self, dbapi_connection: PoolProxiedConnection) -> None:
122 dbapi_connection.rollback()
124 def do_commit(self, dbapi_connection: PoolProxiedConnection) -> None:
125 dbapi_connection.commit()
127 def do_terminate(self, dbapi_connection: DBAPIConnection) -> None:
128 dbapi_connection.close()
130 def do_close(self, dbapi_connection: DBAPIConnection) -> None:
131 dbapi_connection.close()
133 def _do_ping_w_event(self, dbapi_connection: DBAPIConnection) -> bool:
134 raise NotImplementedError(
135 "The ping feature requires that a dialect is "
136 "passed to the connection pool."
137 )
139 def get_driver_connection(self, connection: DBAPIConnection) -> Any:
140 return connection
143class _AsyncConnDialect(_ConnDialect):
144 is_async = True
147class _CreatorFnType(Protocol):
148 def __call__(self) -> DBAPIConnection: ...
151class _CreatorWRecFnType(Protocol):
152 def __call__(self, rec: ConnectionPoolEntry) -> DBAPIConnection: ...
155class Pool(log.Identified, event.EventTarget):
156 """Abstract base class for connection pools."""
158 dispatch: dispatcher[Pool]
159 echo: log._EchoFlagType
161 _orig_logging_name: Optional[str]
162 _dialect: Union[_ConnDialect, Dialect] = _ConnDialect()
163 _creator_arg: Union[_CreatorFnType, _CreatorWRecFnType]
164 _invoke_creator: _CreatorWRecFnType
165 _invalidate_time: float
167 def __init__(
168 self,
169 creator: Union[_CreatorFnType, _CreatorWRecFnType],
170 recycle: int = -1,
171 echo: log._EchoFlagType = None,
172 logging_name: Optional[str] = None,
173 reset_on_return: _ResetStyleArgType = True,
174 events: Optional[List[Tuple[_ListenerFnType, str]]] = None,
175 dialect: Optional[Union[_ConnDialect, Dialect]] = None,
176 pre_ping: bool = False,
177 _dispatch: Optional[_DispatchCommon[Pool]] = None,
178 ):
179 """
180 Construct a Pool.
182 :param creator: a callable function that returns a DB-API
183 connection object. The function will be called with
184 parameters.
186 :param recycle: If set to a value other than -1, number of
187 seconds between connection recycling, which means upon
188 checkout, if this timeout is surpassed the connection will be
189 closed and replaced with a newly opened connection. Defaults to -1.
191 :param logging_name: String identifier which will be used within
192 the "name" field of logging records generated within the
193 "sqlalchemy.pool" logger. Defaults to a hexstring of the object's
194 id.
196 :param echo: if True, the connection pool will log
197 informational output such as when connections are invalidated
198 as well as when connections are recycled to the default log handler,
199 which defaults to ``sys.stdout`` for output.. If set to the string
200 ``"debug"``, the logging will include pool checkouts and checkins.
202 The :paramref:`_pool.Pool.echo` parameter can also be set from the
203 :func:`_sa.create_engine` call by using the
204 :paramref:`_sa.create_engine.echo_pool` parameter.
206 .. seealso::
208 :ref:`dbengine_logging` - further detail on how to configure
209 logging.
211 :param reset_on_return: Determine steps to take on
212 connections as they are returned to the pool, which were
213 not otherwise handled by a :class:`_engine.Connection`.
214 Available from :func:`_sa.create_engine` via the
215 :paramref:`_sa.create_engine.pool_reset_on_return` parameter.
217 :paramref:`_pool.Pool.reset_on_return` can have any of these values:
219 * ``"rollback"`` - call rollback() on the connection,
220 to release locks and transaction resources.
221 This is the default value. The vast majority
222 of use cases should leave this value set.
223 * ``"commit"`` - call commit() on the connection,
224 to release locks and transaction resources.
225 A commit here may be desirable for databases that
226 cache query plans if a commit is emitted,
227 such as Microsoft SQL Server. However, this
228 value is more dangerous than 'rollback' because
229 any data changes present on the transaction
230 are committed unconditionally.
231 * ``None`` - don't do anything on the connection.
232 This setting may be appropriate if the database / DBAPI
233 works in pure "autocommit" mode at all times, or if
234 a custom reset handler is established using the
235 :meth:`.PoolEvents.reset` event handler.
237 * ``True`` - same as 'rollback', this is here for
238 backwards compatibility.
239 * ``False`` - same as None, this is here for
240 backwards compatibility.
242 For further customization of reset on return, the
243 :meth:`.PoolEvents.reset` event hook may be used which can perform
244 any connection activity desired on reset.
246 .. seealso::
248 :ref:`pool_reset_on_return`
250 :meth:`.PoolEvents.reset`
252 :param events: a list of 2-tuples, each of the form
253 ``(callable, target)`` which will be passed to :func:`.event.listen`
254 upon construction. Provided here so that event listeners
255 can be assigned via :func:`_sa.create_engine` before dialect-level
256 listeners are applied.
258 :param dialect: a :class:`.Dialect` that will handle the job
259 of calling rollback(), close(), or commit() on DBAPI connections.
260 If omitted, a built-in "stub" dialect is used. Applications that
261 make use of :func:`_sa.create_engine` should not use this parameter
262 as it is handled by the engine creation strategy.
264 :param pre_ping: if True, the pool will emit a "ping" (typically
265 "SELECT 1", but is dialect-specific) on the connection
266 upon checkout, to test if the connection is alive or not. If not,
267 the connection is transparently re-connected and upon success, all
268 other pooled connections established prior to that timestamp are
269 invalidated. Requires that a dialect is passed as well to
270 interpret the disconnection error.
272 """
273 if logging_name:
274 self.logging_name = self._orig_logging_name = logging_name
275 else:
276 self._orig_logging_name = None
278 log.instance_logger(self, echoflag=echo)
279 self._creator = creator
280 self._recycle = recycle
281 self._invalidate_time = 0
282 self._pre_ping = pre_ping
283 self._reset_on_return = util.parse_user_argument_for_enum(
284 reset_on_return,
285 {
286 ResetStyle.reset_rollback: ["rollback", True],
287 ResetStyle.reset_none: ["none", None, False],
288 ResetStyle.reset_commit: ["commit"],
289 },
290 "reset_on_return",
291 )
293 self.echo = echo
295 if _dispatch:
296 self.dispatch._update(_dispatch, only_propagate=False)
297 if dialect:
298 self._dialect = dialect
299 if events:
300 for fn, target in events:
301 event.listen(self, target, fn)
303 @util.hybridproperty
304 def _is_asyncio(self) -> bool:
305 return self._dialect.is_async
307 @property
308 def _creator(self) -> Union[_CreatorFnType, _CreatorWRecFnType]:
309 return self._creator_arg
311 @_creator.setter
312 def _creator(
313 self, creator: Union[_CreatorFnType, _CreatorWRecFnType]
314 ) -> None:
315 self._creator_arg = creator
317 # mypy seems to get super confused assigning functions to
318 # attributes
319 self._invoke_creator = self._should_wrap_creator(creator)
321 @_creator.deleter
322 def _creator(self) -> None:
323 # needed for mock testing
324 del self._creator_arg
325 del self._invoke_creator
327 def _should_wrap_creator(
328 self, creator: Union[_CreatorFnType, _CreatorWRecFnType]
329 ) -> _CreatorWRecFnType:
330 """Detect if creator accepts a single argument, or is sent
331 as a legacy style no-arg function.
333 """
335 try:
336 argspec = util.get_callable_argspec(self._creator, no_self=True)
337 except TypeError:
338 creator_fn = cast(_CreatorFnType, creator)
339 return lambda rec: creator_fn()
341 if argspec.defaults is not None:
342 defaulted = len(argspec.defaults)
343 else:
344 defaulted = 0
345 positionals = len(argspec[0]) - defaulted
347 # look for the exact arg signature that DefaultStrategy
348 # sends us
349 if (argspec[0], argspec[3]) == (["connection_record"], (None,)):
350 return cast(_CreatorWRecFnType, creator)
351 # or just a single positional
352 elif positionals == 1:
353 return cast(_CreatorWRecFnType, creator)
354 # all other cases, just wrap and assume legacy "creator" callable
355 # thing
356 else:
357 creator_fn = cast(_CreatorFnType, creator)
358 return lambda rec: creator_fn()
360 def _close_connection(
361 self, connection: DBAPIConnection, *, terminate: bool = False
362 ) -> None:
363 self.logger.debug(
364 "%s connection %r",
365 "Hard-closing" if terminate else "Closing",
366 connection,
367 )
368 try:
369 if terminate:
370 self._dialect.do_terminate(connection)
371 else:
372 self._dialect.do_close(connection)
373 except BaseException as e:
374 self.logger.error(
375 f"Exception {'terminating' if terminate else 'closing'} "
376 f"connection %r",
377 connection,
378 exc_info=True,
379 )
380 if not isinstance(e, Exception):
381 raise
383 def _create_connection(self) -> ConnectionPoolEntry:
384 """Called by subclasses to create a new ConnectionRecord."""
386 return _ConnectionRecord(self)
388 def _invalidate(
389 self,
390 connection: PoolProxiedConnection,
391 exception: Optional[BaseException] = None,
392 _checkin: bool = True,
393 ) -> None:
394 """Mark all connections established within the generation
395 of the given connection as invalidated.
397 If this pool's last invalidate time is before when the given
398 connection was created, update the timestamp til now. Otherwise,
399 no action is performed.
401 Connections with a start time prior to this pool's invalidation
402 time will be recycled upon next checkout.
403 """
404 rec = getattr(connection, "_connection_record", None)
405 if not rec or self._invalidate_time < rec.starttime:
406 self._invalidate_time = time.time()
407 if _checkin and getattr(connection, "is_valid", False):
408 connection.invalidate(exception)
410 def recreate(self) -> Pool:
411 """Return a new :class:`_pool.Pool`, of the same class as this one
412 and configured with identical creation arguments.
414 This method is used in conjunction with :meth:`dispose`
415 to close out an entire :class:`_pool.Pool` and create a new one in
416 its place.
418 """
420 raise NotImplementedError()
422 def dispose(self) -> None:
423 """Dispose of this pool.
425 This method leaves the possibility of checked-out connections
426 remaining open, as it only affects connections that are
427 idle in the pool.
429 .. seealso::
431 :meth:`Pool.recreate`
433 """
435 raise NotImplementedError()
437 def connect(self) -> PoolProxiedConnection:
438 """Return a DBAPI connection from the pool.
440 The connection is instrumented such that when its
441 ``close()`` method is called, the connection will be returned to
442 the pool.
444 """
445 return _ConnectionFairy._checkout(self)
447 def _return_conn(self, record: ConnectionPoolEntry) -> None:
448 """Given a _ConnectionRecord, return it to the :class:`_pool.Pool`.
450 This method is called when an instrumented DBAPI connection
451 has its ``close()`` method called.
453 """
454 self._do_return_conn(record)
456 def _do_get(self) -> ConnectionPoolEntry:
457 """Implementation for :meth:`get`, supplied by subclasses."""
459 raise NotImplementedError()
461 def _do_return_conn(self, record: ConnectionPoolEntry) -> None:
462 """Implementation for :meth:`return_conn`, supplied by subclasses."""
464 raise NotImplementedError()
466 def status(self) -> str:
467 """Returns a brief description of the state of this pool."""
468 raise NotImplementedError()
471class ManagesConnection:
472 """Common base for the two connection-management interfaces
473 :class:`.PoolProxiedConnection` and :class:`.ConnectionPoolEntry`.
475 These two objects are typically exposed in the public facing API
476 via the connection pool event hooks, documented at :class:`.PoolEvents`.
478 .. versionadded:: 2.0
480 """
482 __slots__ = ()
484 dbapi_connection: Optional[DBAPIConnection]
485 """A reference to the actual DBAPI connection being tracked.
487 This is a :pep:`249`-compliant object that for traditional sync-style
488 dialects is provided by the third-party
489 DBAPI implementation in use. For asyncio dialects, the implementation
490 is typically an adapter object provided by the SQLAlchemy dialect
491 itself; the underlying asyncio object is available via the
492 :attr:`.ManagesConnection.driver_connection` attribute.
494 SQLAlchemy's interface for the DBAPI connection is based on the
495 :class:`.DBAPIConnection` protocol object
497 .. seealso::
499 :attr:`.ManagesConnection.driver_connection`
501 :ref:`faq_dbapi_connection`
503 """
505 driver_connection: Optional[Any]
506 """The "driver level" connection object as used by the Python
507 DBAPI or database driver.
509 For traditional :pep:`249` DBAPI implementations, this object will
510 be the same object as that of
511 :attr:`.ManagesConnection.dbapi_connection`. For an asyncio database
512 driver, this will be the ultimate "connection" object used by that
513 driver, such as the ``asyncpg.Connection`` object which will not have
514 standard pep-249 methods.
516 .. versionadded:: 1.4.24
518 .. seealso::
520 :attr:`.ManagesConnection.dbapi_connection`
522 :ref:`faq_dbapi_connection`
524 """
526 @util.ro_memoized_property
527 def info(self) -> _InfoType:
528 """Info dictionary associated with the underlying DBAPI connection
529 referred to by this :class:`.ManagesConnection` instance, allowing
530 user-defined data to be associated with the connection.
532 The data in this dictionary is persistent for the lifespan
533 of the DBAPI connection itself, including across pool checkins
534 and checkouts. When the connection is invalidated
535 and replaced with a new one, this dictionary is cleared.
537 For a :class:`.PoolProxiedConnection` instance that's not associated
538 with a :class:`.ConnectionPoolEntry`, such as if it were detached, the
539 attribute returns a dictionary that is local to that
540 :class:`.ConnectionPoolEntry`. Therefore the
541 :attr:`.ManagesConnection.info` attribute will always provide a Python
542 dictionary.
544 .. seealso::
546 :attr:`.ManagesConnection.record_info`
549 """
550 raise NotImplementedError()
552 @util.ro_memoized_property
553 def record_info(self) -> Optional[_InfoType]:
554 """Persistent info dictionary associated with this
555 :class:`.ManagesConnection`.
557 Unlike the :attr:`.ManagesConnection.info` dictionary, the lifespan
558 of this dictionary is that of the :class:`.ConnectionPoolEntry`
559 which owns it; therefore this dictionary will persist across
560 reconnects and connection invalidation for a particular entry
561 in the connection pool.
563 For a :class:`.PoolProxiedConnection` instance that's not associated
564 with a :class:`.ConnectionPoolEntry`, such as if it were detached, the
565 attribute returns None. Contrast to the :attr:`.ManagesConnection.info`
566 dictionary which is never None.
569 .. seealso::
571 :attr:`.ManagesConnection.info`
573 """
574 raise NotImplementedError()
576 def invalidate(
577 self, e: Optional[BaseException] = None, soft: bool = False
578 ) -> None:
579 """Mark the managed connection as invalidated.
581 :param e: an exception object indicating a reason for the invalidation.
583 :param soft: if True, the connection isn't closed; instead, this
584 connection will be recycled on next checkout.
586 .. seealso::
588 :ref:`pool_connection_invalidation`
591 """
592 raise NotImplementedError()
595class ConnectionPoolEntry(ManagesConnection):
596 """Interface for the object that maintains an individual database
597 connection on behalf of a :class:`_pool.Pool` instance.
599 The :class:`.ConnectionPoolEntry` object represents the long term
600 maintenance of a particular connection for a pool, including expiring or
601 invalidating that connection to have it replaced with a new one, which will
602 continue to be maintained by that same :class:`.ConnectionPoolEntry`
603 instance. Compared to :class:`.PoolProxiedConnection`, which is the
604 short-term, per-checkout connection manager, this object lasts for the
605 lifespan of a particular "slot" within a connection pool.
607 The :class:`.ConnectionPoolEntry` object is mostly visible to public-facing
608 API code when it is delivered to connection pool event hooks, such as
609 :meth:`_events.PoolEvents.connect` and :meth:`_events.PoolEvents.checkout`.
611 .. versionadded:: 2.0 :class:`.ConnectionPoolEntry` provides the public
612 facing interface for the :class:`._ConnectionRecord` internal class.
614 """
616 __slots__ = ()
618 @property
619 def in_use(self) -> bool:
620 """Return True the connection is currently checked out"""
622 raise NotImplementedError()
624 def close(self) -> None:
625 """Close the DBAPI connection managed by this connection pool entry."""
626 raise NotImplementedError()
629class _ConnectionRecord(ConnectionPoolEntry):
630 """Maintains a position in a connection pool which references a pooled
631 connection.
633 This is an internal object used by the :class:`_pool.Pool` implementation
634 to provide context management to a DBAPI connection maintained by
635 that :class:`_pool.Pool`. The public facing interface for this class
636 is described by the :class:`.ConnectionPoolEntry` class. See that
637 class for public API details.
639 .. seealso::
641 :class:`.ConnectionPoolEntry`
643 :class:`.PoolProxiedConnection`
645 """
647 __slots__ = (
648 "__pool",
649 "fairy_ref",
650 "finalize_callback",
651 "fresh",
652 "starttime",
653 "dbapi_connection",
654 "__weakref__",
655 "__dict__",
656 )
658 finalize_callback: Deque[Callable[[DBAPIConnection], None]]
659 fresh: bool
660 fairy_ref: Optional[weakref.ref[_ConnectionFairy]]
661 starttime: float
663 def __init__(self, pool: Pool, connect: bool = True):
664 self.fresh = False
665 self.fairy_ref = None
666 self.starttime = 0
667 self.dbapi_connection = None
669 self.__pool = pool
670 if connect:
671 self.__connect()
672 self.finalize_callback = deque()
674 dbapi_connection: Optional[DBAPIConnection]
676 @property
677 def driver_connection(self) -> Optional[Any]: # type: ignore[override] # mypy#4125 # noqa: E501
678 if self.dbapi_connection is None:
679 return None
680 else:
681 return self.__pool._dialect.get_driver_connection(
682 self.dbapi_connection
683 )
685 @property
686 @util.deprecated(
687 "2.0",
688 "The _ConnectionRecord.connection attribute is deprecated; "
689 "please use 'driver_connection'",
690 )
691 def connection(self) -> Optional[DBAPIConnection]:
692 return self.dbapi_connection
694 _soft_invalidate_time: float = 0
696 @util.ro_memoized_property
697 def info(self) -> _InfoType:
698 return {}
700 @util.ro_memoized_property
701 def record_info(self) -> Optional[_InfoType]:
702 return {}
704 @classmethod
705 def checkout(cls, pool: Pool) -> _ConnectionFairy:
706 if TYPE_CHECKING:
707 rec = cast(_ConnectionRecord, pool._do_get())
708 else:
709 rec = pool._do_get()
711 try:
712 dbapi_connection = rec.get_connection()
713 except BaseException as err:
714 with util.safe_reraise():
715 rec._checkin_failed(err, _fairy_was_created=False)
717 # not reached, for code linters only
718 raise
720 echo = pool._should_log_debug()
721 fairy = _ConnectionFairy(pool, dbapi_connection, rec, echo)
723 # assign fairy_ref to the ConnectionRecord; note that under StaticPool,
724 # this could be swapping out an existing fairy on that ConnectionRecord
725 rec.fairy_ref = weakref.ref(fairy)
727 if echo:
728 pool.logger.debug(
729 "Connection %r checked out from pool", dbapi_connection
730 )
731 return fairy
733 def _checkin_failed(
734 self, err: BaseException, _fairy_was_created: bool = True
735 ) -> None:
736 try:
737 self.invalidate(e=err)
738 finally:
739 if self.dbapi_connection is not None:
740 # invalidate() was itself interrupted, such as by a second
741 # cancellation, before it could close the connection. keep
742 # the connection so that its close can be retried, but
743 # have the next checkout recycle it rather than use it
744 self._soft_invalidate_time = time.time()
746 # the record must be returned in any case; nothing else will
747 # ever return it, as no _ConnectionFairy refers to it
748 self.checkin(
749 _fairy_was_created=_fairy_was_created,
750 )
752 def checkin(self, _fairy_was_created: bool = True) -> None:
753 if self.fairy_ref is None and _fairy_was_created:
754 # _fairy_was_created is False for the initial get connection phase;
755 # meaning there was no _ConnectionFairy and we must unconditionally
756 # do a checkin.
757 #
758 # otherwise, if fairy_was_created==True, if fairy_ref is None here
759 # that means we were checked in already, so this looks like
760 # a double checkin.
761 util.warn("Double checkin attempted on %s" % self)
762 return
763 self.fairy_ref = None
764 connection = self.dbapi_connection
765 pool = self.__pool
766 while self.finalize_callback:
767 finalizer = self.finalize_callback.pop()
768 if connection is not None:
769 finalizer(connection)
770 if pool.dispatch.checkin:
771 pool.dispatch.checkin(connection, self)
773 pool._return_conn(self)
775 @property
776 def in_use(self) -> bool:
777 return self.fairy_ref is not None
779 @property
780 def needs_gc(self) -> bool:
781 ref = self.fairy_ref
782 return ref is not None and ref() is None
784 @property
785 def last_connect_time(self) -> float:
786 return self.starttime
788 def close(self) -> None:
789 if self.dbapi_connection is not None:
790 self.__close()
792 def invalidate(
793 self, e: Optional[BaseException] = None, soft: bool = False
794 ) -> None:
795 # already invalidated
796 if self.dbapi_connection is None:
797 return
798 if soft:
799 self.__pool.dispatch.soft_invalidate(
800 self.dbapi_connection, self, e
801 )
802 else:
803 self.__pool.dispatch.invalidate(self.dbapi_connection, self, e)
804 if e is not None:
805 self.__pool.logger.info(
806 "%sInvalidate connection %r (reason: %s:%s)",
807 "Soft " if soft else "",
808 self.dbapi_connection,
809 e.__class__.__name__,
810 e,
811 )
812 else:
813 self.__pool.logger.info(
814 "%sInvalidate connection %r",
815 "Soft " if soft else "",
816 self.dbapi_connection,
817 )
819 if soft:
820 self._soft_invalidate_time = time.time()
821 else:
822 self.__close(terminate=True)
823 self.dbapi_connection = None
825 def get_connection(self) -> DBAPIConnection:
826 recycle = False
828 # NOTE: the various comparisons here are assuming that measurable time
829 # passes between these state changes. however, time.time() is not
830 # guaranteed to have sub-second precision. comparisons of
831 # "invalidation time" to "starttime" should perhaps use >= so that the
832 # state change can take place assuming no measurable time has passed,
833 # however this does not guarantee correct behavior here as if time
834 # continues to not pass, it will try to reconnect repeatedly until
835 # these timestamps diverge, so in that sense using > is safer. Per
836 # https://stackoverflow.com/a/1938096/34549, Windows time.time() may be
837 # within 16 milliseconds accuracy, so unit tests for connection
838 # invalidation need a sleep of at least this long between initial start
839 # time and invalidation for the logic below to work reliably.
841 if self.dbapi_connection is None:
842 self.info.clear()
843 self.__connect()
844 elif (
845 self.__pool._recycle > -1
846 and time.time() - self.starttime > self.__pool._recycle
847 ):
848 self.__pool.logger.info(
849 "Connection %r exceeded timeout; recycling",
850 self.dbapi_connection,
851 )
852 recycle = True
853 elif self.__pool._invalidate_time > self.starttime:
854 self.__pool.logger.info(
855 "Connection %r invalidated due to pool invalidation; "
856 + "recycling",
857 self.dbapi_connection,
858 )
859 recycle = True
860 elif self._soft_invalidate_time > self.starttime:
861 self.__pool.logger.info(
862 "Connection %r invalidated due to local soft invalidation; "
863 + "recycling",
864 self.dbapi_connection,
865 )
866 recycle = True
868 if recycle:
869 self.__close(terminate=True)
870 self.info.clear()
872 self.__connect()
874 assert self.dbapi_connection is not None
875 return self.dbapi_connection
877 def _is_hard_or_soft_invalidated(self) -> bool:
878 return (
879 self.dbapi_connection is None
880 or self.__pool._invalidate_time > self.starttime
881 or (self._soft_invalidate_time > self.starttime)
882 )
884 def __close(self, *, terminate: bool = False) -> None:
885 self.finalize_callback.clear()
886 if self.__pool.dispatch.close:
887 self.__pool.dispatch.close(self.dbapi_connection, self)
888 assert self.dbapi_connection is not None
889 self.__pool._close_connection(
890 self.dbapi_connection, terminate=terminate
891 )
892 self.dbapi_connection = None
894 def __connect(self) -> None:
895 pool = self.__pool
897 # ensure any existing connection is removed, so that if
898 # creator fails, this attribute stays None
899 self.dbapi_connection = None
900 try:
901 self.starttime = time.time()
902 self.dbapi_connection = connection = pool._invoke_creator(self)
903 pool.logger.debug("Created new connection %r", connection)
904 self.fresh = True
905 except BaseException as e:
906 with util.safe_reraise():
907 pool.logger.debug("Error on connect(): %s", e)
908 else:
909 try:
910 # in SQLAlchemy 1.4 the first_connect event is not used by
911 # the engine, so this will usually not be set
912 if pool.dispatch.first_connect:
913 pool.dispatch.first_connect.for_modify(
914 pool.dispatch
915 ).exec_once_unless_exception(self.dbapi_connection, self)
917 # init of the dialect now takes place within the connect
918 # event, so ensure a mutex is used on the first run
919 pool.dispatch.connect.for_modify(
920 pool.dispatch
921 )._exec_w_sync_on_first_run(self.dbapi_connection, self)
922 except BaseException:
923 # the connection is established but the events that
924 # configure it did not complete; nothing else has a
925 # reference to it, so close it here or it is stranded.
926 # for an asyncio driver in particular there is no other
927 # opportunity, as the garbage collector cannot close a
928 # connection that needs the event loop
929 with util.safe_reraise():
930 try:
931 pool._close_connection(connection, terminate=True)
932 except BaseException:
933 pool.logger.exception(
934 "Exception closing connection %r stranded by a "
935 "failed connect event",
936 connection,
937 )
938 self.dbapi_connection = None
941def _finalize_fairy(
942 dbapi_connection: Optional[DBAPIConnection],
943 connection_record: Optional[_ConnectionRecord],
944 pool: Pool,
945 echo: Optional[log._EchoFlagType],
946 transaction_was_reset: bool = False,
947 fairy: Optional[_ConnectionFairy] = None,
948 is_gc_cleanup: bool = False,
949) -> None:
950 """Cleanup for a :class:`._ConnectionFairy` whether or not it's already
951 been garbage collected.
953 When using an async dialect no IO can happen here (without using
954 a dedicated thread), since this is called outside the greenlet
955 context and with an already running loop. In this case function
956 will only log a message and raise a warning.
957 """
959 if is_gc_cleanup:
960 assert connection_record is not None
962 # check connection record to see that we're the current
963 # fairy for this record. if not, then return; assume the
964 # record is either checked in, or another fairy supersedes us
965 # (can happen with StaticPool)
966 if not connection_record.needs_gc:
967 return
968 assert dbapi_connection is None
969 dbapi_connection = connection_record.dbapi_connection
971 # null pool is not _is_asyncio but can be used also with async dialects
972 dont_restore_gced = pool._dialect.is_async
974 if dont_restore_gced:
975 detach = connection_record is None or is_gc_cleanup
976 can_manipulate_connection = not is_gc_cleanup
977 can_close_or_terminate_connection = (
978 not pool._dialect.is_async or pool._dialect.has_terminate
979 )
980 requires_terminate_for_close = (
981 pool._dialect.is_async and pool._dialect.has_terminate
982 )
984 else:
985 detach = connection_record is None
986 can_manipulate_connection = can_close_or_terminate_connection = True
987 requires_terminate_for_close = False
989 if dbapi_connection is not None:
990 if connection_record and echo:
991 pool.logger.debug(
992 "Connection %r being returned to pool", dbapi_connection
993 )
995 try:
996 if not fairy:
997 assert connection_record is not None
998 fairy = _ConnectionFairy(
999 pool,
1000 dbapi_connection,
1001 connection_record,
1002 echo,
1003 )
1004 assert fairy.dbapi_connection is dbapi_connection
1006 fairy._reset(
1007 pool,
1008 transaction_was_reset=transaction_was_reset,
1009 terminate_only=detach,
1010 asyncio_safe=can_manipulate_connection,
1011 )
1013 if detach:
1014 if connection_record:
1015 fairy._pool = pool
1016 fairy.detach()
1018 if can_close_or_terminate_connection:
1019 if pool.dispatch.close_detached:
1020 pool.dispatch.close_detached(dbapi_connection)
1022 pool._close_connection(
1023 dbapi_connection,
1024 terminate=requires_terminate_for_close,
1025 )
1027 except BaseException as e:
1028 pool.logger.error(
1029 "Exception during reset or similar", exc_info=True
1030 )
1031 if not isinstance(e, Exception):
1032 # a cancellation or similar propagates from here, so the
1033 # checkin() below is not reached; return the record now,
1034 # rather than leaving it to garbage collection of the fairy
1035 if connection_record:
1036 connection_record._checkin_failed(e)
1037 raise
1038 if connection_record:
1039 connection_record.invalidate(e=e)
1040 finally:
1041 if detach and is_gc_cleanup and dont_restore_gced:
1042 message = (
1043 "The garbage collector is trying to clean up "
1044 f"non-checked-in connection {dbapi_connection!r}, "
1045 f"""which will be {
1046 'dropped, as it cannot be safely terminated'
1047 if not can_close_or_terminate_connection
1048 else 'terminated'
1049 }. """
1050 "Please ensure that SQLAlchemy pooled connections are "
1051 "returned to "
1052 "the pool explicitly, either by calling ``close()`` "
1053 "or by using appropriate context managers to manage "
1054 "their lifecycle."
1055 )
1056 pool.logger.error(message)
1057 util.warn(message)
1059 if connection_record and connection_record.in_use:
1060 connection_record.checkin()
1062 # give gc some help. See
1063 # test/engine/test_pool.py::PoolEventsTest::test_checkin_event_gc[True]
1064 # which actually started failing when pytest warnings plugin was
1065 # turned on, due to util.warn() above
1066 if fairy is not None:
1067 # don't need the finalizer anymore since we are cleaning up here
1068 fairy._finalizer.detach()
1069 fairy.dbapi_connection = None # type: ignore[assignment]
1070 fairy._connection_record = None
1071 del dbapi_connection
1072 del connection_record
1073 del fairy
1076class PoolProxiedConnection(ManagesConnection):
1077 """A connection-like adapter for a :pep:`249` DBAPI connection, which
1078 includes additional methods specific to the :class:`.Pool` implementation.
1080 :class:`.PoolProxiedConnection` is the public-facing interface for the
1081 internal :class:`._ConnectionFairy` implementation object; users familiar
1082 with :class:`._ConnectionFairy` can consider this object to be equivalent.
1084 .. versionadded:: 2.0 :class:`.PoolProxiedConnection` provides the public-
1085 facing interface for the :class:`._ConnectionFairy` internal class.
1087 """
1089 __slots__ = ()
1091 if typing.TYPE_CHECKING:
1093 def commit(self) -> None: ...
1095 def cursor(self, *args: Any, **kwargs: Any) -> DBAPICursor: ...
1097 def rollback(self) -> None: ...
1099 def __getattr__(self, key: str) -> Any: ...
1101 @property
1102 def is_valid(self) -> bool:
1103 """Return True if this :class:`.PoolProxiedConnection` still refers
1104 to an active DBAPI connection."""
1106 raise NotImplementedError()
1108 @property
1109 def is_detached(self) -> bool:
1110 """Return True if this :class:`.PoolProxiedConnection` is detached
1111 from its pool."""
1113 raise NotImplementedError()
1115 def detach(self) -> None:
1116 """Separate this connection from its Pool.
1118 This means that the connection will no longer be returned to the
1119 pool when closed, and will instead be literally closed. The
1120 associated :class:`.ConnectionPoolEntry` is de-associated from this
1121 DBAPI connection.
1123 Note that any overall connection limiting constraints imposed by a
1124 Pool implementation may be violated after a detach, as the detached
1125 connection is removed from the pool's knowledge and control.
1127 """
1129 raise NotImplementedError()
1131 def close(self) -> None:
1132 """Release this connection back to the pool.
1134 The :meth:`.PoolProxiedConnection.close` method shadows the
1135 :pep:`249` ``.close()`` method, altering its behavior to instead
1136 :term:`release` the proxied connection back to the connection pool.
1138 Upon release to the pool, whether the connection stays "opened" and
1139 pooled in the Python process, versus actually closed out and removed
1140 from the Python process, is based on the pool implementation in use and
1141 its configuration and current state.
1143 """
1144 raise NotImplementedError()
1146 def __enter__(self) -> Self:
1147 return self
1149 def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
1150 self.close()
1151 return None
1154class _AdhocProxiedConnection(PoolProxiedConnection):
1155 """provides the :class:`.PoolProxiedConnection` interface for cases where
1156 the DBAPI connection is not actually proxied.
1158 This is used by the engine internals to pass a consistent
1159 :class:`.PoolProxiedConnection` object to consuming dialects in response to
1160 pool events that may not always have the :class:`._ConnectionFairy`
1161 available.
1163 """
1165 __slots__ = ("dbapi_connection", "_connection_record", "_is_valid")
1167 dbapi_connection: DBAPIConnection
1168 _connection_record: ConnectionPoolEntry
1170 def __init__(
1171 self,
1172 dbapi_connection: DBAPIConnection,
1173 connection_record: ConnectionPoolEntry,
1174 ):
1175 self.dbapi_connection = dbapi_connection
1176 self._connection_record = connection_record
1177 self._is_valid = True
1179 @property
1180 def driver_connection(self) -> Any: # type: ignore[override] # mypy#4125
1181 return self._connection_record.driver_connection
1183 @property
1184 def connection(self) -> DBAPIConnection:
1185 return self.dbapi_connection
1187 @property
1188 def is_valid(self) -> bool:
1189 """Implement is_valid state attribute.
1191 for the adhoc proxied connection it's assumed the connection is valid
1192 as there is no "invalidate" routine.
1194 """
1195 return self._is_valid
1197 def invalidate(
1198 self, e: Optional[BaseException] = None, soft: bool = False
1199 ) -> None:
1200 self._is_valid = False
1202 @util.ro_non_memoized_property
1203 def record_info(self) -> Optional[_InfoType]:
1204 return self._connection_record.record_info
1206 def cursor(self, *args: Any, **kwargs: Any) -> DBAPICursor:
1207 return self.dbapi_connection.cursor(*args, **kwargs)
1209 def __getattr__(self, key: Any) -> Any:
1210 return getattr(self.dbapi_connection, key)
1213class _ConnectionFairy(PoolProxiedConnection):
1214 """Proxies a DBAPI connection and provides return-on-dereference
1215 support.
1217 This is an internal object used by the :class:`_pool.Pool` implementation
1218 to provide context management to a DBAPI connection delivered by
1219 that :class:`_pool.Pool`. The public facing interface for this class
1220 is described by the :class:`.PoolProxiedConnection` class. See that
1221 class for public API details.
1223 The name "fairy" is inspired by the fact that the
1224 :class:`._ConnectionFairy` object's lifespan is transitory, as it lasts
1225 only for the length of a specific DBAPI connection being checked out from
1226 the pool, and additionally that as a transparent proxy, it is mostly
1227 invisible.
1229 .. seealso::
1231 :class:`.PoolProxiedConnection`
1233 :class:`.ConnectionPoolEntry`
1236 """
1238 __slots__ = (
1239 "dbapi_connection",
1240 "_connection_record",
1241 "_finalizer",
1242 "_echo",
1243 "_pool",
1244 "_counter",
1245 "__weakref__",
1246 "__dict__",
1247 )
1249 pool: Pool
1250 dbapi_connection: DBAPIConnection
1251 _echo: log._EchoFlagType
1253 def __init__(
1254 self,
1255 pool: Pool,
1256 dbapi_connection: DBAPIConnection,
1257 connection_record: _ConnectionRecord,
1258 echo: log._EchoFlagType,
1259 ):
1260 self._pool = pool
1261 self._counter = 0
1262 self.dbapi_connection = dbapi_connection
1263 self._connection_record = connection_record
1264 self._echo = echo
1266 # use weakref.finalize as a destructor
1267 self._finalizer = weakref.finalize(
1268 self,
1269 _finalize_fairy,
1270 None,
1271 connection_record,
1272 pool,
1273 echo,
1274 transaction_was_reset=False,
1275 is_gc_cleanup=True,
1276 )
1278 _connection_record: Optional[_ConnectionRecord]
1280 @property
1281 def driver_connection(self) -> Optional[Any]: # type: ignore[override] # mypy#4125 # noqa: E501
1282 if self._connection_record is None:
1283 return None
1284 return self._connection_record.driver_connection
1286 @property
1287 @util.deprecated(
1288 "2.0",
1289 "The _ConnectionFairy.connection attribute is deprecated; "
1290 "please use 'driver_connection'",
1291 )
1292 def connection(self) -> DBAPIConnection:
1293 return self.dbapi_connection
1295 @classmethod
1296 def _checkout(
1297 cls,
1298 pool: Pool,
1299 threadconns: Optional[threading.local] = None,
1300 fairy: Optional[_ConnectionFairy] = None,
1301 ) -> _ConnectionFairy:
1302 if not fairy:
1303 fairy = _ConnectionRecord.checkout(pool)
1305 if threadconns is not None:
1306 threadconns.current = weakref.ref(fairy)
1308 assert (
1309 fairy._connection_record is not None
1310 ), "can't 'checkout' a detached connection fairy"
1311 assert (
1312 fairy.dbapi_connection is not None
1313 ), "can't 'checkout' an invalidated connection fairy"
1315 fairy._counter += 1
1316 if (
1317 not pool.dispatch.checkout and not pool._pre_ping
1318 ) or fairy._counter != 1:
1319 return fairy
1321 # Pool listeners can trigger a reconnection on checkout, as well
1322 # as the pre-pinger.
1323 # there are three attempts made here, but note that if the database
1324 # is not accessible from a connection standpoint, those won't proceed
1325 # here.
1327 attempts = 2
1329 while attempts > 0:
1330 connection_is_fresh = fairy._connection_record.fresh
1331 fairy._connection_record.fresh = False
1332 try:
1333 if pool._pre_ping:
1334 if not connection_is_fresh:
1335 if fairy._echo:
1336 pool.logger.debug(
1337 "Pool pre-ping on connection %s",
1338 fairy.dbapi_connection,
1339 )
1340 result = pool._dialect._do_ping_w_event(
1341 fairy.dbapi_connection
1342 )
1343 if not result:
1344 if fairy._echo:
1345 pool.logger.debug(
1346 "Pool pre-ping on connection %s failed, "
1347 "will invalidate pool",
1348 fairy.dbapi_connection,
1349 )
1350 raise exc.InvalidatePoolError()
1351 elif fairy._echo:
1352 pool.logger.debug(
1353 "Connection %s is fresh, skipping pre-ping",
1354 fairy.dbapi_connection,
1355 )
1357 pool.dispatch.checkout(
1358 fairy.dbapi_connection, fairy._connection_record, fairy
1359 )
1360 return fairy
1361 except exc.DisconnectionError as e:
1362 if e.invalidate_pool:
1363 pool.logger.info(
1364 "Disconnection detected on checkout, "
1365 "invalidating all pooled connections prior to "
1366 "current timestamp (reason: %r)",
1367 e,
1368 )
1369 fairy._connection_record.invalidate(e)
1370 pool._invalidate(fairy, e, _checkin=False)
1371 else:
1372 pool.logger.info(
1373 "Disconnection detected on checkout, "
1374 "invalidating individual connection %s (reason: %r)",
1375 fairy.dbapi_connection,
1376 e,
1377 )
1378 fairy._connection_record.invalidate(e)
1379 try:
1380 fairy.dbapi_connection = (
1381 fairy._connection_record.get_connection()
1382 )
1383 except BaseException as err:
1384 with util.safe_reraise():
1385 fairy._connection_record._checkin_failed(
1386 err,
1387 _fairy_was_created=True,
1388 )
1390 # prevent _ConnectionFairy from being carried
1391 # in the stack trace. Do this after the
1392 # connection record has been checked in, so that
1393 # if the del triggers a finalize fairy, it won't
1394 # try to checkin a second time.
1395 del fairy
1397 # never called, this is for code linters
1398 raise
1400 attempts -= 1
1401 except BaseException as be_outer:
1402 with util.safe_reraise():
1403 rec = fairy._connection_record
1404 if rec is not None:
1405 rec._checkin_failed(
1406 be_outer,
1407 _fairy_was_created=True,
1408 )
1410 # prevent _ConnectionFairy from being carried
1411 # in the stack trace, see above
1412 del fairy
1414 # never called, this is for code linters
1415 raise
1417 pool.logger.info("Reconnection attempts exhausted on checkout")
1418 fairy.invalidate()
1419 raise exc.InvalidRequestError("This connection is closed")
1421 def _checkout_existing(self) -> _ConnectionFairy:
1422 return _ConnectionFairy._checkout(self._pool, fairy=self)
1424 def _checkin(self, transaction_was_reset: bool = False) -> None:
1425 _finalize_fairy(
1426 self.dbapi_connection,
1427 self._connection_record,
1428 self._pool,
1429 self._echo,
1430 transaction_was_reset=transaction_was_reset,
1431 fairy=self,
1432 )
1434 def _close(self) -> None:
1435 self._checkin()
1437 def _reset(
1438 self,
1439 pool: Pool,
1440 transaction_was_reset: bool,
1441 terminate_only: bool,
1442 asyncio_safe: bool,
1443 ) -> None:
1444 if pool.dispatch.reset:
1445 pool.dispatch.reset(
1446 self.dbapi_connection,
1447 self._connection_record,
1448 PoolResetState(
1449 transaction_was_reset=transaction_was_reset,
1450 terminate_only=terminate_only,
1451 asyncio_safe=asyncio_safe,
1452 ),
1453 )
1455 if not asyncio_safe:
1456 return
1458 if pool._reset_on_return is reset_rollback:
1459 if transaction_was_reset:
1460 if self._echo:
1461 pool.logger.debug(
1462 "Connection %s reset, transaction already reset",
1463 self.dbapi_connection,
1464 )
1465 else:
1466 if self._echo:
1467 pool.logger.debug(
1468 "Connection %s rollback-on-return",
1469 self.dbapi_connection,
1470 )
1471 pool._dialect.do_rollback(self)
1472 elif pool._reset_on_return is reset_commit:
1473 if self._echo:
1474 pool.logger.debug(
1475 "Connection %s commit-on-return",
1476 self.dbapi_connection,
1477 )
1478 pool._dialect.do_commit(self)
1480 @property
1481 def _logger(self) -> log._IdentifiedLoggerType:
1482 return self._pool.logger
1484 @property
1485 def is_valid(self) -> bool:
1486 return self.dbapi_connection is not None
1488 @property
1489 def is_detached(self) -> bool:
1490 return self._connection_record is None
1492 @util.ro_memoized_property
1493 def info(self) -> _InfoType:
1494 if self._connection_record is None:
1495 return {}
1496 else:
1497 return self._connection_record.info
1499 @util.ro_non_memoized_property
1500 def record_info(self) -> Optional[_InfoType]:
1501 if self._connection_record is None:
1502 return None
1503 else:
1504 return self._connection_record.record_info
1506 def invalidate(
1507 self, e: Optional[BaseException] = None, soft: bool = False
1508 ) -> None:
1509 if self.dbapi_connection is None:
1510 util.warn("Can't invalidate an already-closed connection.")
1511 return
1512 if self._connection_record:
1513 self._connection_record.invalidate(e=e, soft=soft)
1514 if not soft:
1515 # prevent any rollback / reset actions etc. on
1516 # the connection
1517 self.dbapi_connection = None # type: ignore[assignment]
1519 # finalize
1520 self._checkin()
1522 def cursor(self, *args: Any, **kwargs: Any) -> DBAPICursor:
1523 assert self.dbapi_connection is not None
1524 return self.dbapi_connection.cursor(*args, **kwargs)
1526 def __getattr__(self, key: str) -> Any:
1527 return getattr(self.dbapi_connection, key)
1529 def detach(self) -> None:
1530 if self._connection_record is not None:
1531 rec = self._connection_record
1532 rec.fairy_ref = None
1534 # cancel the finalizer, as current behavior is that there's no
1535 # GC "cleanup" for a detached connection. not sure if this
1536 # is the most appropriate decision; see issue #13570
1537 self._finalizer.detach()
1539 rec.dbapi_connection = None
1540 # TODO: should this be _return_conn?
1541 self._pool._do_return_conn(self._connection_record)
1543 # can't get the descriptor assignment to work here
1544 # in pylance. mypy is OK w/ it
1545 self.info = self.info.copy() # type: ignore[misc]
1547 self._connection_record = None
1549 if self._pool.dispatch.detach:
1550 self._pool.dispatch.detach(self.dbapi_connection, rec)
1552 def close(self) -> None:
1553 self._counter -= 1
1554 if self._counter == 0:
1555 self._checkin()
1557 def _close_special(self, transaction_reset: bool = False) -> None:
1558 self._counter -= 1
1559 if self._counter == 0:
1560 self._checkin(transaction_was_reset=transaction_reset)