Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/pool/base.py: 57%
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 self.invalidate(e=err)
737 self.checkin(
738 _fairy_was_created=_fairy_was_created,
739 )
741 def checkin(self, _fairy_was_created: bool = True) -> None:
742 if self.fairy_ref is None and _fairy_was_created:
743 # _fairy_was_created is False for the initial get connection phase;
744 # meaning there was no _ConnectionFairy and we must unconditionally
745 # do a checkin.
746 #
747 # otherwise, if fairy_was_created==True, if fairy_ref is None here
748 # that means we were checked in already, so this looks like
749 # a double checkin.
750 util.warn("Double checkin attempted on %s" % self)
751 return
752 self.fairy_ref = None
753 connection = self.dbapi_connection
754 pool = self.__pool
755 while self.finalize_callback:
756 finalizer = self.finalize_callback.pop()
757 if connection is not None:
758 finalizer(connection)
759 if pool.dispatch.checkin:
760 pool.dispatch.checkin(connection, self)
762 pool._return_conn(self)
764 @property
765 def in_use(self) -> bool:
766 return self.fairy_ref is not None
768 @property
769 def needs_gc(self) -> bool:
770 ref = self.fairy_ref
771 return ref is not None and ref() is None
773 @property
774 def last_connect_time(self) -> float:
775 return self.starttime
777 def close(self) -> None:
778 if self.dbapi_connection is not None:
779 self.__close()
781 def invalidate(
782 self, e: Optional[BaseException] = None, soft: bool = False
783 ) -> None:
784 # already invalidated
785 if self.dbapi_connection is None:
786 return
787 if soft:
788 self.__pool.dispatch.soft_invalidate(
789 self.dbapi_connection, self, e
790 )
791 else:
792 self.__pool.dispatch.invalidate(self.dbapi_connection, self, e)
793 if e is not None:
794 self.__pool.logger.info(
795 "%sInvalidate connection %r (reason: %s:%s)",
796 "Soft " if soft else "",
797 self.dbapi_connection,
798 e.__class__.__name__,
799 e,
800 )
801 else:
802 self.__pool.logger.info(
803 "%sInvalidate connection %r",
804 "Soft " if soft else "",
805 self.dbapi_connection,
806 )
808 if soft:
809 self._soft_invalidate_time = time.time()
810 else:
811 self.__close(terminate=True)
812 self.dbapi_connection = None
814 def get_connection(self) -> DBAPIConnection:
815 recycle = False
817 # NOTE: the various comparisons here are assuming that measurable time
818 # passes between these state changes. however, time.time() is not
819 # guaranteed to have sub-second precision. comparisons of
820 # "invalidation time" to "starttime" should perhaps use >= so that the
821 # state change can take place assuming no measurable time has passed,
822 # however this does not guarantee correct behavior here as if time
823 # continues to not pass, it will try to reconnect repeatedly until
824 # these timestamps diverge, so in that sense using > is safer. Per
825 # https://stackoverflow.com/a/1938096/34549, Windows time.time() may be
826 # within 16 milliseconds accuracy, so unit tests for connection
827 # invalidation need a sleep of at least this long between initial start
828 # time and invalidation for the logic below to work reliably.
830 if self.dbapi_connection is None:
831 self.info.clear()
832 self.__connect()
833 elif (
834 self.__pool._recycle > -1
835 and time.time() - self.starttime > self.__pool._recycle
836 ):
837 self.__pool.logger.info(
838 "Connection %r exceeded timeout; recycling",
839 self.dbapi_connection,
840 )
841 recycle = True
842 elif self.__pool._invalidate_time > self.starttime:
843 self.__pool.logger.info(
844 "Connection %r invalidated due to pool invalidation; "
845 + "recycling",
846 self.dbapi_connection,
847 )
848 recycle = True
849 elif self._soft_invalidate_time > self.starttime:
850 self.__pool.logger.info(
851 "Connection %r invalidated due to local soft invalidation; "
852 + "recycling",
853 self.dbapi_connection,
854 )
855 recycle = True
857 if recycle:
858 self.__close(terminate=True)
859 self.info.clear()
861 self.__connect()
863 assert self.dbapi_connection is not None
864 return self.dbapi_connection
866 def _is_hard_or_soft_invalidated(self) -> bool:
867 return (
868 self.dbapi_connection is None
869 or self.__pool._invalidate_time > self.starttime
870 or (self._soft_invalidate_time > self.starttime)
871 )
873 def __close(self, *, terminate: bool = False) -> None:
874 self.finalize_callback.clear()
875 if self.__pool.dispatch.close:
876 self.__pool.dispatch.close(self.dbapi_connection, self)
877 assert self.dbapi_connection is not None
878 self.__pool._close_connection(
879 self.dbapi_connection, terminate=terminate
880 )
881 self.dbapi_connection = None
883 def __connect(self) -> None:
884 pool = self.__pool
886 # ensure any existing connection is removed, so that if
887 # creator fails, this attribute stays None
888 self.dbapi_connection = None
889 try:
890 self.starttime = time.time()
891 self.dbapi_connection = connection = pool._invoke_creator(self)
892 pool.logger.debug("Created new connection %r", connection)
893 self.fresh = True
894 except BaseException as e:
895 with util.safe_reraise():
896 pool.logger.debug("Error on connect(): %s", e)
897 else:
898 try:
899 # in SQLAlchemy 1.4 the first_connect event is not used by
900 # the engine, so this will usually not be set
901 if pool.dispatch.first_connect:
902 pool.dispatch.first_connect.for_modify(
903 pool.dispatch
904 ).exec_once_unless_exception(self.dbapi_connection, self)
906 # init of the dialect now takes place within the connect
907 # event, so ensure a mutex is used on the first run
908 pool.dispatch.connect.for_modify(
909 pool.dispatch
910 )._exec_w_sync_on_first_run(self.dbapi_connection, self)
911 except BaseException:
912 # the connection is established but the events that
913 # configure it did not complete; nothing else has a
914 # reference to it, so close it here or it is stranded.
915 # for an asyncio driver in particular there is no other
916 # opportunity, as the garbage collector cannot close a
917 # connection that needs the event loop
918 with util.safe_reraise():
919 try:
920 pool._close_connection(connection, terminate=True)
921 except BaseException:
922 pool.logger.exception(
923 "Exception closing connection %r stranded by a "
924 "failed connect event",
925 connection,
926 )
927 self.dbapi_connection = None
930def _finalize_fairy(
931 dbapi_connection: Optional[DBAPIConnection],
932 connection_record: Optional[_ConnectionRecord],
933 pool: Pool,
934 echo: Optional[log._EchoFlagType],
935 transaction_was_reset: bool = False,
936 fairy: Optional[_ConnectionFairy] = None,
937 is_gc_cleanup: bool = False,
938) -> None:
939 """Cleanup for a :class:`._ConnectionFairy` whether or not it's already
940 been garbage collected.
942 When using an async dialect no IO can happen here (without using
943 a dedicated thread), since this is called outside the greenlet
944 context and with an already running loop. In this case function
945 will only log a message and raise a warning.
946 """
948 if is_gc_cleanup:
949 assert connection_record is not None
951 # check connection record to see that we're the current
952 # fairy for this record. if not, then return; assume the
953 # record is either checked in, or another fairy supersedes us
954 # (can happen with StaticPool)
955 if not connection_record.needs_gc:
956 return
957 assert dbapi_connection is None
958 dbapi_connection = connection_record.dbapi_connection
960 # null pool is not _is_asyncio but can be used also with async dialects
961 dont_restore_gced = pool._dialect.is_async
963 if dont_restore_gced:
964 detach = connection_record is None or is_gc_cleanup
965 can_manipulate_connection = not is_gc_cleanup
966 can_close_or_terminate_connection = (
967 not pool._dialect.is_async or pool._dialect.has_terminate
968 )
969 requires_terminate_for_close = (
970 pool._dialect.is_async and pool._dialect.has_terminate
971 )
973 else:
974 detach = connection_record is None
975 can_manipulate_connection = can_close_or_terminate_connection = True
976 requires_terminate_for_close = False
978 if dbapi_connection is not None:
979 if connection_record and echo:
980 pool.logger.debug(
981 "Connection %r being returned to pool", dbapi_connection
982 )
984 try:
985 if not fairy:
986 assert connection_record is not None
987 fairy = _ConnectionFairy(
988 pool,
989 dbapi_connection,
990 connection_record,
991 echo,
992 )
993 assert fairy.dbapi_connection is dbapi_connection
995 fairy._reset(
996 pool,
997 transaction_was_reset=transaction_was_reset,
998 terminate_only=detach,
999 asyncio_safe=can_manipulate_connection,
1000 )
1002 if detach:
1003 if connection_record:
1004 fairy._pool = pool
1005 fairy.detach()
1007 if can_close_or_terminate_connection:
1008 if pool.dispatch.close_detached:
1009 pool.dispatch.close_detached(dbapi_connection)
1011 pool._close_connection(
1012 dbapi_connection,
1013 terminate=requires_terminate_for_close,
1014 )
1016 except BaseException as e:
1017 pool.logger.error(
1018 "Exception during reset or similar", exc_info=True
1019 )
1020 if connection_record:
1021 connection_record.invalidate(e=e)
1022 if not isinstance(e, Exception):
1023 raise
1024 finally:
1025 if detach and is_gc_cleanup and dont_restore_gced:
1026 message = (
1027 "The garbage collector is trying to clean up "
1028 f"non-checked-in connection {dbapi_connection!r}, "
1029 f"""which will be {
1030 'dropped, as it cannot be safely terminated'
1031 if not can_close_or_terminate_connection
1032 else 'terminated'
1033 }. """
1034 "Please ensure that SQLAlchemy pooled connections are "
1035 "returned to "
1036 "the pool explicitly, either by calling ``close()`` "
1037 "or by using appropriate context managers to manage "
1038 "their lifecycle."
1039 )
1040 pool.logger.error(message)
1041 util.warn(message)
1043 if connection_record and connection_record.in_use:
1044 connection_record.checkin()
1046 # give gc some help. See
1047 # test/engine/test_pool.py::PoolEventsTest::test_checkin_event_gc[True]
1048 # which actually started failing when pytest warnings plugin was
1049 # turned on, due to util.warn() above
1050 if fairy is not None:
1051 # don't need the finalizer anymore since we are cleaning up here
1052 fairy._finalizer.detach()
1053 fairy.dbapi_connection = None # type: ignore[assignment]
1054 fairy._connection_record = None
1055 del dbapi_connection
1056 del connection_record
1057 del fairy
1060class PoolProxiedConnection(ManagesConnection):
1061 """A connection-like adapter for a :pep:`249` DBAPI connection, which
1062 includes additional methods specific to the :class:`.Pool` implementation.
1064 :class:`.PoolProxiedConnection` is the public-facing interface for the
1065 internal :class:`._ConnectionFairy` implementation object; users familiar
1066 with :class:`._ConnectionFairy` can consider this object to be equivalent.
1068 .. versionadded:: 2.0 :class:`.PoolProxiedConnection` provides the public-
1069 facing interface for the :class:`._ConnectionFairy` internal class.
1071 """
1073 __slots__ = ()
1075 if typing.TYPE_CHECKING:
1077 def commit(self) -> None: ...
1079 def cursor(self, *args: Any, **kwargs: Any) -> DBAPICursor: ...
1081 def rollback(self) -> None: ...
1083 def __getattr__(self, key: str) -> Any: ...
1085 @property
1086 def is_valid(self) -> bool:
1087 """Return True if this :class:`.PoolProxiedConnection` still refers
1088 to an active DBAPI connection."""
1090 raise NotImplementedError()
1092 @property
1093 def is_detached(self) -> bool:
1094 """Return True if this :class:`.PoolProxiedConnection` is detached
1095 from its pool."""
1097 raise NotImplementedError()
1099 def detach(self) -> None:
1100 """Separate this connection from its Pool.
1102 This means that the connection will no longer be returned to the
1103 pool when closed, and will instead be literally closed. The
1104 associated :class:`.ConnectionPoolEntry` is de-associated from this
1105 DBAPI connection.
1107 Note that any overall connection limiting constraints imposed by a
1108 Pool implementation may be violated after a detach, as the detached
1109 connection is removed from the pool's knowledge and control.
1111 """
1113 raise NotImplementedError()
1115 def close(self) -> None:
1116 """Release this connection back to the pool.
1118 The :meth:`.PoolProxiedConnection.close` method shadows the
1119 :pep:`249` ``.close()`` method, altering its behavior to instead
1120 :term:`release` the proxied connection back to the connection pool.
1122 Upon release to the pool, whether the connection stays "opened" and
1123 pooled in the Python process, versus actually closed out and removed
1124 from the Python process, is based on the pool implementation in use and
1125 its configuration and current state.
1127 """
1128 raise NotImplementedError()
1130 def __enter__(self) -> Self:
1131 return self
1133 def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
1134 self.close()
1135 return None
1138class _AdhocProxiedConnection(PoolProxiedConnection):
1139 """provides the :class:`.PoolProxiedConnection` interface for cases where
1140 the DBAPI connection is not actually proxied.
1142 This is used by the engine internals to pass a consistent
1143 :class:`.PoolProxiedConnection` object to consuming dialects in response to
1144 pool events that may not always have the :class:`._ConnectionFairy`
1145 available.
1147 """
1149 __slots__ = ("dbapi_connection", "_connection_record", "_is_valid")
1151 dbapi_connection: DBAPIConnection
1152 _connection_record: ConnectionPoolEntry
1154 def __init__(
1155 self,
1156 dbapi_connection: DBAPIConnection,
1157 connection_record: ConnectionPoolEntry,
1158 ):
1159 self.dbapi_connection = dbapi_connection
1160 self._connection_record = connection_record
1161 self._is_valid = True
1163 @property
1164 def driver_connection(self) -> Any: # type: ignore[override] # mypy#4125
1165 return self._connection_record.driver_connection
1167 @property
1168 def connection(self) -> DBAPIConnection:
1169 return self.dbapi_connection
1171 @property
1172 def is_valid(self) -> bool:
1173 """Implement is_valid state attribute.
1175 for the adhoc proxied connection it's assumed the connection is valid
1176 as there is no "invalidate" routine.
1178 """
1179 return self._is_valid
1181 def invalidate(
1182 self, e: Optional[BaseException] = None, soft: bool = False
1183 ) -> None:
1184 self._is_valid = False
1186 @util.ro_non_memoized_property
1187 def record_info(self) -> Optional[_InfoType]:
1188 return self._connection_record.record_info
1190 def cursor(self, *args: Any, **kwargs: Any) -> DBAPICursor:
1191 return self.dbapi_connection.cursor(*args, **kwargs)
1193 def __getattr__(self, key: Any) -> Any:
1194 return getattr(self.dbapi_connection, key)
1197class _ConnectionFairy(PoolProxiedConnection):
1198 """Proxies a DBAPI connection and provides return-on-dereference
1199 support.
1201 This is an internal object used by the :class:`_pool.Pool` implementation
1202 to provide context management to a DBAPI connection delivered by
1203 that :class:`_pool.Pool`. The public facing interface for this class
1204 is described by the :class:`.PoolProxiedConnection` class. See that
1205 class for public API details.
1207 The name "fairy" is inspired by the fact that the
1208 :class:`._ConnectionFairy` object's lifespan is transitory, as it lasts
1209 only for the length of a specific DBAPI connection being checked out from
1210 the pool, and additionally that as a transparent proxy, it is mostly
1211 invisible.
1213 .. seealso::
1215 :class:`.PoolProxiedConnection`
1217 :class:`.ConnectionPoolEntry`
1220 """
1222 __slots__ = (
1223 "dbapi_connection",
1224 "_connection_record",
1225 "_finalizer",
1226 "_echo",
1227 "_pool",
1228 "_counter",
1229 "__weakref__",
1230 "__dict__",
1231 )
1233 pool: Pool
1234 dbapi_connection: DBAPIConnection
1235 _echo: log._EchoFlagType
1237 def __init__(
1238 self,
1239 pool: Pool,
1240 dbapi_connection: DBAPIConnection,
1241 connection_record: _ConnectionRecord,
1242 echo: log._EchoFlagType,
1243 ):
1244 self._pool = pool
1245 self._counter = 0
1246 self.dbapi_connection = dbapi_connection
1247 self._connection_record = connection_record
1248 self._echo = echo
1250 # use weakref.finalize as a destructor
1251 self._finalizer = weakref.finalize(
1252 self,
1253 _finalize_fairy,
1254 None,
1255 connection_record,
1256 pool,
1257 echo,
1258 transaction_was_reset=False,
1259 is_gc_cleanup=True,
1260 )
1262 _connection_record: Optional[_ConnectionRecord]
1264 @property
1265 def driver_connection(self) -> Optional[Any]: # type: ignore[override] # mypy#4125 # noqa: E501
1266 if self._connection_record is None:
1267 return None
1268 return self._connection_record.driver_connection
1270 @property
1271 @util.deprecated(
1272 "2.0",
1273 "The _ConnectionFairy.connection attribute is deprecated; "
1274 "please use 'driver_connection'",
1275 )
1276 def connection(self) -> DBAPIConnection:
1277 return self.dbapi_connection
1279 @classmethod
1280 def _checkout(
1281 cls,
1282 pool: Pool,
1283 threadconns: Optional[threading.local] = None,
1284 fairy: Optional[_ConnectionFairy] = None,
1285 ) -> _ConnectionFairy:
1286 if not fairy:
1287 fairy = _ConnectionRecord.checkout(pool)
1289 if threadconns is not None:
1290 threadconns.current = weakref.ref(fairy)
1292 assert (
1293 fairy._connection_record is not None
1294 ), "can't 'checkout' a detached connection fairy"
1295 assert (
1296 fairy.dbapi_connection is not None
1297 ), "can't 'checkout' an invalidated connection fairy"
1299 fairy._counter += 1
1300 if (
1301 not pool.dispatch.checkout and not pool._pre_ping
1302 ) or fairy._counter != 1:
1303 return fairy
1305 # Pool listeners can trigger a reconnection on checkout, as well
1306 # as the pre-pinger.
1307 # there are three attempts made here, but note that if the database
1308 # is not accessible from a connection standpoint, those won't proceed
1309 # here.
1311 attempts = 2
1313 while attempts > 0:
1314 connection_is_fresh = fairy._connection_record.fresh
1315 fairy._connection_record.fresh = False
1316 try:
1317 if pool._pre_ping:
1318 if not connection_is_fresh:
1319 if fairy._echo:
1320 pool.logger.debug(
1321 "Pool pre-ping on connection %s",
1322 fairy.dbapi_connection,
1323 )
1324 result = pool._dialect._do_ping_w_event(
1325 fairy.dbapi_connection
1326 )
1327 if not result:
1328 if fairy._echo:
1329 pool.logger.debug(
1330 "Pool pre-ping on connection %s failed, "
1331 "will invalidate pool",
1332 fairy.dbapi_connection,
1333 )
1334 raise exc.InvalidatePoolError()
1335 elif fairy._echo:
1336 pool.logger.debug(
1337 "Connection %s is fresh, skipping pre-ping",
1338 fairy.dbapi_connection,
1339 )
1341 pool.dispatch.checkout(
1342 fairy.dbapi_connection, fairy._connection_record, fairy
1343 )
1344 return fairy
1345 except exc.DisconnectionError as e:
1346 if e.invalidate_pool:
1347 pool.logger.info(
1348 "Disconnection detected on checkout, "
1349 "invalidating all pooled connections prior to "
1350 "current timestamp (reason: %r)",
1351 e,
1352 )
1353 fairy._connection_record.invalidate(e)
1354 pool._invalidate(fairy, e, _checkin=False)
1355 else:
1356 pool.logger.info(
1357 "Disconnection detected on checkout, "
1358 "invalidating individual connection %s (reason: %r)",
1359 fairy.dbapi_connection,
1360 e,
1361 )
1362 fairy._connection_record.invalidate(e)
1363 try:
1364 fairy.dbapi_connection = (
1365 fairy._connection_record.get_connection()
1366 )
1367 except BaseException as err:
1368 with util.safe_reraise():
1369 fairy._connection_record._checkin_failed(
1370 err,
1371 _fairy_was_created=True,
1372 )
1374 # prevent _ConnectionFairy from being carried
1375 # in the stack trace. Do this after the
1376 # connection record has been checked in, so that
1377 # if the del triggers a finalize fairy, it won't
1378 # try to checkin a second time.
1379 del fairy
1381 # never called, this is for code linters
1382 raise
1384 attempts -= 1
1385 except BaseException as be_outer:
1386 with util.safe_reraise():
1387 rec = fairy._connection_record
1388 if rec is not None:
1389 rec._checkin_failed(
1390 be_outer,
1391 _fairy_was_created=True,
1392 )
1394 # prevent _ConnectionFairy from being carried
1395 # in the stack trace, see above
1396 del fairy
1398 # never called, this is for code linters
1399 raise
1401 pool.logger.info("Reconnection attempts exhausted on checkout")
1402 fairy.invalidate()
1403 raise exc.InvalidRequestError("This connection is closed")
1405 def _checkout_existing(self) -> _ConnectionFairy:
1406 return _ConnectionFairy._checkout(self._pool, fairy=self)
1408 def _checkin(self, transaction_was_reset: bool = False) -> None:
1409 _finalize_fairy(
1410 self.dbapi_connection,
1411 self._connection_record,
1412 self._pool,
1413 self._echo,
1414 transaction_was_reset=transaction_was_reset,
1415 fairy=self,
1416 )
1418 def _close(self) -> None:
1419 self._checkin()
1421 def _reset(
1422 self,
1423 pool: Pool,
1424 transaction_was_reset: bool,
1425 terminate_only: bool,
1426 asyncio_safe: bool,
1427 ) -> None:
1428 if pool.dispatch.reset:
1429 pool.dispatch.reset(
1430 self.dbapi_connection,
1431 self._connection_record,
1432 PoolResetState(
1433 transaction_was_reset=transaction_was_reset,
1434 terminate_only=terminate_only,
1435 asyncio_safe=asyncio_safe,
1436 ),
1437 )
1439 if not asyncio_safe:
1440 return
1442 if pool._reset_on_return is reset_rollback:
1443 if transaction_was_reset:
1444 if self._echo:
1445 pool.logger.debug(
1446 "Connection %s reset, transaction already reset",
1447 self.dbapi_connection,
1448 )
1449 else:
1450 if self._echo:
1451 pool.logger.debug(
1452 "Connection %s rollback-on-return",
1453 self.dbapi_connection,
1454 )
1455 pool._dialect.do_rollback(self)
1456 elif pool._reset_on_return is reset_commit:
1457 if self._echo:
1458 pool.logger.debug(
1459 "Connection %s commit-on-return",
1460 self.dbapi_connection,
1461 )
1462 pool._dialect.do_commit(self)
1464 @property
1465 def _logger(self) -> log._IdentifiedLoggerType:
1466 return self._pool.logger
1468 @property
1469 def is_valid(self) -> bool:
1470 return self.dbapi_connection is not None
1472 @property
1473 def is_detached(self) -> bool:
1474 return self._connection_record is None
1476 @util.ro_memoized_property
1477 def info(self) -> _InfoType:
1478 if self._connection_record is None:
1479 return {}
1480 else:
1481 return self._connection_record.info
1483 @util.ro_non_memoized_property
1484 def record_info(self) -> Optional[_InfoType]:
1485 if self._connection_record is None:
1486 return None
1487 else:
1488 return self._connection_record.record_info
1490 def invalidate(
1491 self, e: Optional[BaseException] = None, soft: bool = False
1492 ) -> None:
1493 if self.dbapi_connection is None:
1494 util.warn("Can't invalidate an already-closed connection.")
1495 return
1496 if self._connection_record:
1497 self._connection_record.invalidate(e=e, soft=soft)
1498 if not soft:
1499 # prevent any rollback / reset actions etc. on
1500 # the connection
1501 self.dbapi_connection = None # type: ignore[assignment]
1503 # finalize
1504 self._checkin()
1506 def cursor(self, *args: Any, **kwargs: Any) -> DBAPICursor:
1507 assert self.dbapi_connection is not None
1508 return self.dbapi_connection.cursor(*args, **kwargs)
1510 def __getattr__(self, key: str) -> Any:
1511 return getattr(self.dbapi_connection, key)
1513 def detach(self) -> None:
1514 if self._connection_record is not None:
1515 rec = self._connection_record
1516 rec.fairy_ref = None
1518 # cancel the finalizer, as current behavior is that there's no
1519 # GC "cleanup" for a detached connection. not sure if this
1520 # is the most appropriate decision; see issue #13570
1521 self._finalizer.detach()
1523 rec.dbapi_connection = None
1524 # TODO: should this be _return_conn?
1525 self._pool._do_return_conn(self._connection_record)
1527 # can't get the descriptor assignment to work here
1528 # in pylance. mypy is OK w/ it
1529 self.info = self.info.copy() # type: ignore[misc]
1531 self._connection_record = None
1533 if self._pool.dispatch.detach:
1534 self._pool.dispatch.detach(self.dbapi_connection, rec)
1536 def close(self) -> None:
1537 self._counter -= 1
1538 if self._counter == 0:
1539 self._checkin()
1541 def _close_special(self, transaction_reset: bool = False) -> None:
1542 self._counter -= 1
1543 if self._counter == 0:
1544 self._checkin(transaction_was_reset=transaction_reset)