codekingpro/portable-devtools
115k
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
7
8
9"""Base constructs for connection pools."""
10
11from __future__ import annotations
12
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 Dict
24from typing import List
25from typing import Optional
26from typing import Tuple
27from typing import TYPE_CHECKING
28from typing import Union
29import weakref
30
31from .. import event
32from .. import exc
33from .. import log
34from .. import util
35from ..util.typing import Literal
36from ..util.typing import Protocol
37from ..util.typing import Self
38
39if TYPE_CHECKING:
40 from ..engine.interfaces import DBAPIConnection
41 from ..engine.interfaces import DBAPICursor
42 from ..engine.interfaces import Dialect
43 from ..event import _DispatchCommon
44 from ..event import _ListenerFnType
45 from ..event import dispatcher
46 from ..sql._typing import _InfoType
47
48
49@dataclasses.dataclass(frozen=True)
50class PoolResetState:
51 """describes the state of a DBAPI connection as it is being passed to
52 the :meth:`.PoolEvents.reset` connection pool event.
53
54 .. versionadded:: 2.0.0b3
55
56 """
57
58 __slots__ = ("transaction_was_reset", "terminate_only", "asyncio_safe")
59
60 transaction_was_reset: bool
61 """Indicates if the transaction on the DBAPI connection was already
62 essentially "reset" back by the :class:`.Connection` object.
63
64 This boolean is True if the :class:`.Connection` had transactional
65 state present upon it, which was then not closed using the
66 :meth:`.Connection.rollback` or :meth:`.Connection.commit` method;
67 instead, the transaction was closed inline within the
68 :meth:`.Connection.close` method so is guaranteed to remain non-present
69 when this event is reached.
70
71 """
72
73 terminate_only: bool
74 """indicates if the connection is to be immediately terminated and
75 not checked in to the pool.
76
77 This occurs for connections that were invalidated, as well as asyncio
78 connections that were not cleanly handled by the calling code that
79 are instead being garbage collected. In the latter case,
80 operations can't be safely run on asyncio connections within garbage
81 collection as there is not necessarily an event loop present.
82
83 """
84
85 asyncio_safe: bool
86 """Indicates if the reset operation is occurring within a scope where
87 an enclosing event loop is expected to be present for asyncio applications.
88
89 Will be False in the case that the connection is being garbage collected.
90
91 """
92
93
94class ResetStyle(Enum):
95 """Describe options for "reset on return" behaviors."""
96
97 reset_rollback = 0
98 reset_commit = 1
99 reset_none = 2
100
101
102_ResetStyleArgType = Union[
103 ResetStyle,
104 Literal[True, None, False, "commit", "rollback"],
105]
106reset_rollback, reset_commit, reset_none = list(ResetStyle)
107
108
109class _ConnDialect:
110 """partial implementation of :class:`.Dialect`
111 which provides DBAPI connection methods.
112
113 When a :class:`_pool.Pool` is combined with an :class:`_engine.Engine`,
114 the :class:`_engine.Engine` replaces this with its own
115 :class:`.Dialect`.
116
117 """
118
119 is_async = False
120 has_terminate = False
121
122 def do_rollback(self, dbapi_connection: PoolProxiedConnection) -> None:
123 dbapi_connection.rollback()
124
125 def do_commit(self, dbapi_connection: PoolProxiedConnection) -> None:
126 dbapi_connection.commit()
127
128 def do_terminate(self, dbapi_connection: DBAPIConnection) -> None:
129 dbapi_connection.close()
130
131 def do_close(self, dbapi_connection: DBAPIConnection) -> None:
132 dbapi_connection.close()
133
134 def _do_ping_w_event(self, dbapi_connection: DBAPIConnection) -> bool:
135 raise NotImplementedError(
136 "The ping feature requires that a dialect is "
137 "passed to the connection pool."
138 )
139
140 def get_driver_connection(self, connection: DBAPIConnection) -> Any:
141 return connection
142
143
144class _AsyncConnDialect(_ConnDialect):
145 is_async = True
146
147
148class _CreatorFnType(Protocol):
149 def __call__(self) -> DBAPIConnection: ...
150
151
152class _CreatorWRecFnType(Protocol):
153 def __call__(self, rec: ConnectionPoolEntry) -> DBAPIConnection: ...
154
155
156class Pool(log.Identified, event.EventTarget):
157 """Abstract base class for connection pools."""
158
159 dispatch: dispatcher[Pool]
160 echo: log._EchoFlagType
161
162 _orig_logging_name: Optional[str]
163 _dialect: Union[_ConnDialect, Dialect] = _ConnDialect()
164 _creator_arg: Union[_CreatorFnType, _CreatorWRecFnType]
165 _invoke_creator: _CreatorWRecFnType
166 _invalidate_time: float
167
168 def __init__(
169 self,
170 creator: Union[_CreatorFnType, _CreatorWRecFnType],
171 recycle: int = -1,
172 echo: log._EchoFlagType = None,
173 logging_name: Optional[str] = None,
174 reset_on_return: _ResetStyleArgType = True,
175 events: Optional[List[Tuple[_ListenerFnType, str]]] = None,
176 dialect: Optional[Union[_ConnDialect, Dialect]] = None,
177 pre_ping: bool = False,
178 _dispatch: Optional[_DispatchCommon[Pool]] = None,
179 ):
180 """
181 Construct a Pool.
182
183 :param creator: a callable function that returns a DB-API
184 connection object. The function will be called with
185 parameters.
186
187 :param recycle: If set to a value other than -1, number of
188 seconds between connection recycling, which means upon
189 checkout, if this timeout is surpassed the connection will be
190 closed and replaced with a newly opened connection. Defaults to -1.
191
192 :param logging_name: String identifier which will be used within
193 the "name" field of logging records generated within the
194 "sqlalchemy.pool" logger. Defaults to a hexstring of the object's
195 id.
196
197 :param echo: if True, the connection pool will log
198 informational output such as when connections are invalidated
199 as well as when connections are recycled to the default log handler,
200 which defaults to ``sys.stdout`` for output.. If set to the string
201 ``"debug"``, the logging will include pool checkouts and checkins.
202
203 The :paramref:`_pool.Pool.echo` parameter can also be set from the
204 :func:`_sa.create_engine` call by using the
205 :paramref:`_sa.create_engine.echo_pool` parameter.
206
207 .. seealso::
208
209 :ref:`dbengine_logging` - further detail on how to configure
210 logging.
211
212 :param reset_on_return: Determine steps to take on
213 connections as they are returned to the pool, which were
214 not otherwise handled by a :class:`_engine.Connection`.
215 Available from :func:`_sa.create_engine` via the
216 :paramref:`_sa.create_engine.pool_reset_on_return` parameter.
217
218 :paramref:`_pool.Pool.reset_on_return` can have any of these values:
219
220 * ``"rollback"`` - call rollback() on the connection,
221 to release locks and transaction resources.
222 This is the default value. The vast majority
223 of use cases should leave this value set.
224 * ``"commit"`` - call commit() on the connection,
225 to release locks and transaction resources.
226 A commit here may be desirable for databases that
227 cache query plans if a commit is emitted,
228 such as Microsoft SQL Server. However, this
229 value is more dangerous than 'rollback' because
230 any data changes present on the transaction
231 are committed unconditionally.
232 * ``None`` - don't do anything on the connection.
233 This setting may be appropriate if the database / DBAPI
234 works in pure "autocommit" mode at all times, or if
235 a custom reset handler is established using the
236 :meth:`.PoolEvents.reset` event handler.
237
238 * ``True`` - same as 'rollback', this is here for
239 backwards compatibility.
240 * ``False`` - same as None, this is here for
241 backwards compatibility.
242
243 For further customization of reset on return, the
244 :meth:`.PoolEvents.reset` event hook may be used which can perform
245 any connection activity desired on reset.
246
247 .. seealso::
248
249 :ref:`pool_reset_on_return`
250
251 :meth:`.PoolEvents.reset`
252
253 :param events: a list of 2-tuples, each of the form
254 ``(callable, target)`` which will be passed to :func:`.event.listen`
255 upon construction. Provided here so that event listeners
256 can be assigned via :func:`_sa.create_engine` before dialect-level
257 listeners are applied.
258
259 :param dialect: a :class:`.Dialect` that will handle the job
260 of calling rollback(), close(), or commit() on DBAPI connections.
261 If omitted, a built-in "stub" dialect is used. Applications that
262 make use of :func:`_sa.create_engine` should not use this parameter
263 as it is handled by the engine creation strategy.
264
265 :param pre_ping: if True, the pool will emit a "ping" (typically
266 "SELECT 1", but is dialect-specific) on the connection
267 upon checkout, to test if the connection is alive or not. If not,
268 the connection is transparently re-connected and upon success, all
269 other pooled connections established prior to that timestamp are
270 invalidated. Requires that a dialect is passed as well to
271 interpret the disconnection error.
272
273 .. versionadded:: 1.2
274
275 """
276 if logging_name:
277 self.logging_name = self._orig_logging_name = logging_name
278 else:
279 self._orig_logging_name = None
280
281 log.instance_logger(self, echoflag=echo)
282 self._creator = creator
283 self._recycle = recycle
284 self._invalidate_time = 0
285 self._pre_ping = pre_ping
286 self._reset_on_return = util.parse_user_argument_for_enum(
287 reset_on_return,
288 {
289 ResetStyle.reset_rollback: ["rollback", True],
290 ResetStyle.reset_none: ["none", None, False],
291 ResetStyle.reset_commit: ["commit"],
292 },
293 "reset_on_return",
294 )
295
296 self.echo = echo
297
298 if _dispatch:
299 self.dispatch._update(_dispatch, only_propagate=False)
300 if dialect:
301 self._dialect = dialect
302 if events:
303 for fn, target in events:
304 event.listen(self, target, fn)
305
306 @util.hybridproperty
307 def _is_asyncio(self) -> bool:
308 return self._dialect.is_async
309
310 @property
311 def _creator(self) -> Union[_CreatorFnType, _CreatorWRecFnType]:
312 return self._creator_arg
313
314 @_creator.setter
315 def _creator(
316 self, creator: Union[_CreatorFnType, _CreatorWRecFnType]
317 ) -> None:
318 self._creator_arg = creator
319
320 # mypy seems to get super confused assigning functions to
321 # attributes
322 self._invoke_creator = self._should_wrap_creator(creator)
323
324 @_creator.deleter
325 def _creator(self) -> None:
326 # needed for mock testing
327 del self._creator_arg
328 del self._invoke_creator
329
330 def _should_wrap_creator(
331 self, creator: Union[_CreatorFnType, _CreatorWRecFnType]
332 ) -> _CreatorWRecFnType:
333 """Detect if creator accepts a single argument, or is sent
334 as a legacy style no-arg function.
335
336 """
337
338 try:
339 argspec = util.get_callable_argspec(self._creator, no_self=True)
340 except TypeError:
341 creator_fn = cast(_CreatorFnType, creator)
342 return lambda rec: creator_fn()
343
344 if argspec.defaults is not None:
345 defaulted = len(argspec.defaults)
346 else:
347 defaulted = 0
348 positionals = len(argspec[0]) - defaulted
349
350 # look for the exact arg signature that DefaultStrategy
351 # sends us
352 if (argspec[0], argspec[3]) == (["connection_record"], (None,)):
353 return cast(_CreatorWRecFnType, creator)
354 # or just a single positional
355 elif positionals == 1:
356 return cast(_CreatorWRecFnType, creator)
357 # all other cases, just wrap and assume legacy "creator" callable
358 # thing
359 else:
360 creator_fn = cast(_CreatorFnType, creator)
361 return lambda rec: creator_fn()
362
363 def _close_connection(
364 self, connection: DBAPIConnection, *, terminate: bool = False
365 ) -> None:
366 self.logger.debug(
367 "%s connection %r",
368 "Hard-closing" if terminate else "Closing",
369 connection,
370 )
371 try:
372 if terminate:
373 self._dialect.do_terminate(connection)
374 else:
375 self._dialect.do_close(connection)
376 except BaseException as e:
377 self.logger.error(
378 f"Exception {'terminating' if terminate else 'closing'} "
379 f"connection %r",
380 connection,
381 exc_info=True,
382 )
383 if not isinstance(e, Exception):
384 raise
385
386 def _create_connection(self) -> ConnectionPoolEntry:
387 """Called by subclasses to create a new ConnectionRecord."""
388
389 return _ConnectionRecord(self)
390
391 def _invalidate(
392 self,
393 connection: PoolProxiedConnection,
394 exception: Optional[BaseException] = None,
395 _checkin: bool = True,
396 ) -> None:
397 """Mark all connections established within the generation
398 of the given connection as invalidated.
399
400 If this pool's last invalidate time is before when the given
401 connection was created, update the timestamp til now. Otherwise,
402 no action is performed.
403
404 Connections with a start time prior to this pool's invalidation
405 time will be recycled upon next checkout.
406 """
407 rec = getattr(connection, "_connection_record", None)
408 if not rec or self._invalidate_time < rec.starttime:
409 self._invalidate_time = time.time()
410 if _checkin and getattr(connection, "is_valid", False):
411 connection.invalidate(exception)
412
413 def recreate(self) -> Pool:
414 """Return a new :class:`_pool.Pool`, of the same class as this one
415 and configured with identical creation arguments.
416
417 This method is used in conjunction with :meth:`dispose`
418 to close out an entire :class:`_pool.Pool` and create a new one in
419 its place.
420
421 """
422
423 raise NotImplementedError()
424
425 def dispose(self) -> None:
426 """Dispose of this pool.
427
428 This method leaves the possibility of checked-out connections
429 remaining open, as it only affects connections that are
430 idle in the pool.
431
432 .. seealso::
433
434 :meth:`Pool.recreate`
435
436 """
437
438 raise NotImplementedError()
439
440 def connect(self) -> PoolProxiedConnection:
441 """Return a DBAPI connection from the pool.
442
443 The connection is instrumented such that when its
444 ``close()`` method is called, the connection will be returned to
445 the pool.
446
447 """
448 return _ConnectionFairy._checkout(self)
449
450 def _return_conn(self, record: ConnectionPoolEntry) -> None:
451 """Given a _ConnectionRecord, return it to the :class:`_pool.Pool`.
452
453 This method is called when an instrumented DBAPI connection
454 has its ``close()`` method called.
455
456 """
457 self._do_return_conn(record)
458
459 def _do_get(self) -> ConnectionPoolEntry:
460 """Implementation for :meth:`get`, supplied by subclasses."""
461
462 raise NotImplementedError()
463
464 def _do_return_conn(self, record: ConnectionPoolEntry) -> None:
465 """Implementation for :meth:`return_conn`, supplied by subclasses."""
466
467 raise NotImplementedError()
468
469 def status(self) -> str:
470 """Returns a brief description of the state of this pool."""
471 raise NotImplementedError()
472
473
474class ManagesConnection:
475 """Common base for the two connection-management interfaces
476 :class:`.PoolProxiedConnection` and :class:`.ConnectionPoolEntry`.
477
478 These two objects are typically exposed in the public facing API
479 via the connection pool event hooks, documented at :class:`.PoolEvents`.
480
481 .. versionadded:: 2.0
482
483 """
484
485 __slots__ = ()
486
487 dbapi_connection: Optional[DBAPIConnection]
488 """A reference to the actual DBAPI connection being tracked.
489
490 This is a :pep:`249`-compliant object that for traditional sync-style
491 dialects is provided by the third-party
492 DBAPI implementation in use. For asyncio dialects, the implementation
493 is typically an adapter object provided by the SQLAlchemy dialect
494 itself; the underlying asyncio object is available via the
495 :attr:`.ManagesConnection.driver_connection` attribute.
496
497 SQLAlchemy's interface for the DBAPI connection is based on the
498 :class:`.DBAPIConnection` protocol object
499
500 .. seealso::
501
502 :attr:`.ManagesConnection.driver_connection`
503
504 :ref:`faq_dbapi_connection`
505
506 """
507
508 driver_connection: Optional[Any]
509 """The "driver level" connection object as used by the Python
510 DBAPI or database driver.
511
512 For traditional :pep:`249` DBAPI implementations, this object will
513 be the same object as that of
514 :attr:`.ManagesConnection.dbapi_connection`. For an asyncio database
515 driver, this will be the ultimate "connection" object used by that
516 driver, such as the ``asyncpg.Connection`` object which will not have
517 standard pep-249 methods.
518
519 .. versionadded:: 1.4.24
520
521 .. seealso::
522
523 :attr:`.ManagesConnection.dbapi_connection`
524
525 :ref:`faq_dbapi_connection`
526
527 """
528
529 @util.ro_memoized_property
530 def info(self) -> _InfoType:
531 """Info dictionary associated with the underlying DBAPI connection
532 referred to by this :class:`.ManagesConnection` instance, allowing
533 user-defined data to be associated with the connection.
534
535 The data in this dictionary is persistent for the lifespan
536 of the DBAPI connection itself, including across pool checkins
537 and checkouts. When the connection is invalidated
538 and replaced with a new one, this dictionary is cleared.
539
540 For a :class:`.PoolProxiedConnection` instance that's not associated
541 with a :class:`.ConnectionPoolEntry`, such as if it were detached, the
542 attribute returns a dictionary that is local to that
543 :class:`.ConnectionPoolEntry`. Therefore the
544 :attr:`.ManagesConnection.info` attribute will always provide a Python
545 dictionary.
546
547 .. seealso::
548
549 :attr:`.ManagesConnection.record_info`
550
551
552 """
553 raise NotImplementedError()
554
555 @util.ro_memoized_property
556 def record_info(self) -> Optional[_InfoType]:
557 """Persistent info dictionary associated with this
558 :class:`.ManagesConnection`.
559
560 Unlike the :attr:`.ManagesConnection.info` dictionary, the lifespan
561 of this dictionary is that of the :class:`.ConnectionPoolEntry`
562 which owns it; therefore this dictionary will persist across
563 reconnects and connection invalidation for a particular entry
564 in the connection pool.
565
566 For a :class:`.PoolProxiedConnection` instance that's not associated
567 with a :class:`.ConnectionPoolEntry`, such as if it were detached, the
568 attribute returns None. Contrast to the :attr:`.ManagesConnection.info`
569 dictionary which is never None.
570
571
572 .. seealso::
573
574 :attr:`.ManagesConnection.info`
575
576 """
577 raise NotImplementedError()
578
579 def invalidate(
580 self, e: Optional[BaseException] = None, soft: bool = False
581 ) -> None:
582 """Mark the managed connection as invalidated.
583
584 :param e: an exception object indicating a reason for the invalidation.
585
586 :param soft: if True, the connection isn't closed; instead, this
587 connection will be recycled on next checkout.
588
589 .. seealso::
590
591 :ref:`pool_connection_invalidation`
592
593
594 """
595 raise NotImplementedError()
596
597
598class ConnectionPoolEntry(ManagesConnection):
599 """Interface for the object that maintains an individual database
600 connection on behalf of a :class:`_pool.Pool` instance.
601
602 The :class:`.ConnectionPoolEntry` object represents the long term
603 maintenance of a particular connection for a pool, including expiring or
604 invalidating that connection to have it replaced with a new one, which will
605 continue to be maintained by that same :class:`.ConnectionPoolEntry`
606 instance. Compared to :class:`.PoolProxiedConnection`, which is the
607 short-term, per-checkout connection manager, this object lasts for the
608 lifespan of a particular "slot" within a connection pool.
609
610 The :class:`.ConnectionPoolEntry` object is mostly visible to public-facing
611 API code when it is delivered to connection pool event hooks, such as
612 :meth:`_events.PoolEvents.connect` and :meth:`_events.PoolEvents.checkout`.
613
614 .. versionadded:: 2.0 :class:`.ConnectionPoolEntry` provides the public
615 facing interface for the :class:`._ConnectionRecord` internal class.
616
617 """
618
619 __slots__ = ()
620
621 @property
622 def in_use(self) -> bool:
623 """Return True the connection is currently checked out"""
624
625 raise NotImplementedError()
626
627 def close(self) -> None:
628 """Close the DBAPI connection managed by this connection pool entry."""
629 raise NotImplementedError()
630
631
632class _ConnectionRecord(ConnectionPoolEntry):
633 """Maintains a position in a connection pool which references a pooled
634 connection.
635
636 This is an internal object used by the :class:`_pool.Pool` implementation
637 to provide context management to a DBAPI connection maintained by
638 that :class:`_pool.Pool`. The public facing interface for this class
639 is described by the :class:`.ConnectionPoolEntry` class. See that
640 class for public API details.
641
642 .. seealso::
643
644 :class:`.ConnectionPoolEntry`
645
646 :class:`.PoolProxiedConnection`
647
648 """
649
650 __slots__ = (
651 "__pool",
652 "fairy_ref",
653 "finalize_callback",
654 "fresh",
655 "starttime",
656 "dbapi_connection",
657 "__weakref__",
658 "__dict__",
659 )
660
661 finalize_callback: Deque[Callable[[DBAPIConnection], None]]
662 fresh: bool
663 fairy_ref: Optional[weakref.ref[_ConnectionFairy]]
664 starttime: float
665
666 def __init__(self, pool: Pool, connect: bool = True):
667 self.fresh = False
668 self.fairy_ref = None
669 self.starttime = 0
670 self.dbapi_connection = None
671
672 self.__pool = pool
673 if connect:
674 self.__connect()
675 self.finalize_callback = deque()
676
677 dbapi_connection: Optional[DBAPIConnection]
678
679 @property
680 def driver_connection(self) -> Optional[Any]: # type: ignore[override] # mypy#4125 # noqa: E501
681 if self.dbapi_connection is None:
682 return None
683 else:
684 return self.__pool._dialect.get_driver_connection(
685 self.dbapi_connection
686 )
687
688 @property
689 @util.deprecated(
690 "2.0",
691 "The _ConnectionRecord.connection attribute is deprecated; "
692 "please use 'driver_connection'",
693 )
694 def connection(self) -> Optional[DBAPIConnection]:
695 return self.dbapi_connection
696
697 _soft_invalidate_time: float = 0
698
699 @util.ro_memoized_property
700 def info(self) -> _InfoType:
701 return {}
702
703 @util.ro_memoized_property
704 def record_info(self) -> Optional[_InfoType]:
705 return {}
706
707 @classmethod
708 def checkout(cls, pool: Pool) -> _ConnectionFairy:
709 if TYPE_CHECKING:
710 rec = cast(_ConnectionRecord, pool._do_get())
711 else:
712 rec = pool._do_get()
713
714 try:
715 dbapi_connection = rec.get_connection()
716 except BaseException as err:
717 with util.safe_reraise():
718 rec._checkin_failed(err, _fairy_was_created=False)
719
720 # not reached, for code linters only
721 raise
722
723 echo = pool._should_log_debug()
724 fairy = _ConnectionFairy(pool, dbapi_connection, rec, echo)
725
726 rec.fairy_ref = ref = weakref.ref(
727 fairy,
728 lambda ref: (
729 _finalize_fairy(
730 None, rec, pool, ref, echo, transaction_was_reset=False
731 )
732 if _finalize_fairy is not None
733 else None
734 ),
735 )
736 _strong_ref_connection_records[ref] = rec
737 if echo:
738 pool.logger.debug(
739 "Connection %r checked out from pool", dbapi_connection
740 )
741 return fairy
742
743 def _checkin_failed(
744 self, err: BaseException, _fairy_was_created: bool = True
745 ) -> None:
746 self.invalidate(e=err)
747 self.checkin(
748 _fairy_was_created=_fairy_was_created,
749 )
750
751 def checkin(self, _fairy_was_created: bool = True) -> None:
752 if self.fairy_ref is None and _fairy_was_created:
753 # _fairy_was_created is False for the initial get connection phase;
754 # meaning there was no _ConnectionFairy and we must unconditionally
755 # do a checkin.
756 #
757 # otherwise, if fairy_was_created==True, if fairy_ref is None here
758 # that means we were checked in already, so this looks like
759 # a double checkin.
760 util.warn("Double checkin attempted on %s" % self)
761 return
762 self.fairy_ref = None
763 connection = self.dbapi_connection
764 pool = self.__pool
765 while self.finalize_callback:
766 finalizer = self.finalize_callback.pop()
767 if connection is not None:
768 finalizer(connection)
769 if pool.dispatch.checkin:
770 pool.dispatch.checkin(connection, self)
771
772 pool._return_conn(self)
773
774 @property
775 def in_use(self) -> bool:
776 return self.fairy_ref is not None
777
778 @property
779 def last_connect_time(self) -> float:
780 return self.starttime
781
782 def close(self) -> None:
783 if self.dbapi_connection is not None:
784 self.__close()
785
786 def invalidate(
787 self, e: Optional[BaseException] = None, soft: bool = False
788 ) -> None:
789 # already invalidated
790 if self.dbapi_connection is None:
791 return
792 if soft:
793 self.__pool.dispatch.soft_invalidate(
794 self.dbapi_connection, self, e
795 )
796 else:
797 self.__pool.dispatch.invalidate(self.dbapi_connection, self, e)
798 if e is not None:
799 self.__pool.logger.info(
800 "%sInvalidate connection %r (reason: %s:%s)",
801 "Soft " if soft else "",
802 self.dbapi_connection,
803 e.__class__.__name__,
804 e,
805 )
806 else:
807 self.__pool.logger.info(
808 "%sInvalidate connection %r",
809 "Soft " if soft else "",
810 self.dbapi_connection,
811 )
812
813 if soft:
814 self._soft_invalidate_time = time.time()
815 else:
816 self.__close(terminate=True)
817 self.dbapi_connection = None
818
819 def get_connection(self) -> DBAPIConnection:
820 recycle = False
821
822 # NOTE: the various comparisons here are assuming that measurable time
823 # passes between these state changes. however, time.time() is not
824 # guaranteed to have sub-second precision. comparisons of
825 # "invalidation time" to "starttime" should perhaps use >= so that the
826 # state change can take place assuming no measurable time has passed,
827 # however this does not guarantee correct behavior here as if time
828 # continues to not pass, it will try to reconnect repeatedly until
829 # these timestamps diverge, so in that sense using > is safer. Per
830 # https://stackoverflow.com/a/1938096/34549, Windows time.time() may be
831 # within 16 milliseconds accuracy, so unit tests for connection
832 # invalidation need a sleep of at least this long between initial start
833 # time and invalidation for the logic below to work reliably.
834
835 if self.dbapi_connection is None:
836 self.info.clear()
837 self.__connect()
838 elif (
839 self.__pool._recycle > -1
840 and time.time() - self.starttime > self.__pool._recycle
841 ):
842 self.__pool.logger.info(
843 "Connection %r exceeded timeout; recycling",
844 self.dbapi_connection,
845 )
846 recycle = True
847 elif self.__pool._invalidate_time > self.starttime:
848 self.__pool.logger.info(
849 "Connection %r invalidated due to pool invalidation; "
850 + "recycling",
851 self.dbapi_connection,
852 )
853 recycle = True
854 elif self._soft_invalidate_time > self.starttime:
855 self.__pool.logger.info(
856 "Connection %r invalidated due to local soft invalidation; "
857 + "recycling",
858 self.dbapi_connection,
859 )
860 recycle = True
861
862 if recycle:
863 self.__close(terminate=True)
864 self.info.clear()
865
866 self.__connect()
867
868 assert self.dbapi_connection is not None
869 return self.dbapi_connection
870
871 def _is_hard_or_soft_invalidated(self) -> bool:
872 return (
873 self.dbapi_connection is None
874 or self.__pool._invalidate_time > self.starttime
875 or (self._soft_invalidate_time > self.starttime)
876 )
877
878 def __close(self, *, terminate: bool = False) -> None:
879 self.finalize_callback.clear()
880 if self.__pool.dispatch.close:
881 self.__pool.dispatch.close(self.dbapi_connection, self)
882 assert self.dbapi_connection is not None
883 self.__pool._close_connection(
884 self.dbapi_connection, terminate=terminate
885 )
886 self.dbapi_connection = None
887
888 def __connect(self) -> None:
889 pool = self.__pool
890
891 # ensure any existing connection is removed, so that if
892 # creator fails, this attribute stays None
893 self.dbapi_connection = None
894 try:
895 self.starttime = time.time()
896 self.dbapi_connection = connection = pool._invoke_creator(self)
897 pool.logger.debug("Created new connection %r", connection)
898 self.fresh = True
899 except BaseException as e:
900 with util.safe_reraise():
901 pool.logger.debug("Error on connect(): %s", e)
902 else:
903 # in SQLAlchemy 1.4 the first_connect event is not used by
904 # the engine, so this will usually not be set
905 if pool.dispatch.first_connect:
906 pool.dispatch.first_connect.for_modify(
907 pool.dispatch
908 ).exec_once_unless_exception(self.dbapi_connection, self)
909
910 # init of the dialect now takes place within the connect
911 # event, so ensure a mutex is used on the first run
912 pool.dispatch.connect.for_modify(
913 pool.dispatch
914 )._exec_w_sync_on_first_run(self.dbapi_connection, self)
915
916
917def _finalize_fairy(
918 dbapi_connection: Optional[DBAPIConnection],
919 connection_record: Optional[_ConnectionRecord],
920 pool: Pool,
921 ref: Optional[
922 weakref.ref[_ConnectionFairy]
923 ], # this is None when called directly, not by the gc
924 echo: Optional[log._EchoFlagType],
925 transaction_was_reset: bool = False,
926 fairy: Optional[_ConnectionFairy] = None,
927) -> None:
928 """Cleanup for a :class:`._ConnectionFairy` whether or not it's already
929 been garbage collected.
930
931 When using an async dialect no IO can happen here (without using
932 a dedicated thread), since this is called outside the greenlet
933 context and with an already running loop. In this case function
934 will only log a message and raise a warning.
935 """
936
937 is_gc_cleanup = ref is not None
938
939 if is_gc_cleanup:
940 assert ref is not None
941 _strong_ref_connection_records.pop(ref, None)
942 assert connection_record is not None
943 if connection_record.fairy_ref is not ref:
944 return
945 assert dbapi_connection is None
946 dbapi_connection = connection_record.dbapi_connection
947
948 elif fairy:
949 _strong_ref_connection_records.pop(weakref.ref(fairy), None)
950
951 # null pool is not _is_asyncio but can be used also with async dialects
952 dont_restore_gced = pool._dialect.is_async
953
954 if dont_restore_gced:
955 detach = connection_record is None or is_gc_cleanup
956 can_manipulate_connection = not is_gc_cleanup
957 can_close_or_terminate_connection = (
958 not pool._dialect.is_async or pool._dialect.has_terminate
959 )
960 requires_terminate_for_close = (
961 pool._dialect.is_async and pool._dialect.has_terminate
962 )
963
964 else:
965 detach = connection_record is None
966 can_manipulate_connection = can_close_or_terminate_connection = True
967 requires_terminate_for_close = False
968
969 if dbapi_connection is not None:
970 if connection_record and echo:
971 pool.logger.debug(
972 "Connection %r being returned to pool", dbapi_connection
973 )
974
975 try:
976 if not fairy:
977 assert connection_record is not None
978 fairy = _ConnectionFairy(
979 pool,
980 dbapi_connection,
981 connection_record,
982 echo,
983 )
984 assert fairy.dbapi_connection is dbapi_connection
985
986 fairy._reset(
987 pool,
988 transaction_was_reset=transaction_was_reset,
989 terminate_only=detach,
990 asyncio_safe=can_manipulate_connection,
991 )
992
993 if detach:
994 if connection_record:
995 fairy._pool = pool
996 fairy.detach()
997
998 if can_close_or_terminate_connection:
999 if pool.dispatch.close_detached:
1000 pool.dispatch.close_detached(dbapi_connection)
1001
1002 pool._close_connection(
1003 dbapi_connection,
1004 terminate=requires_terminate_for_close,
1005 )
1006
1007 except BaseException as e:
1008 pool.logger.error(
1009 "Exception during reset or similar", exc_info=True
1010 )
1011 if connection_record:
1012 connection_record.invalidate(e=e)
1013 if not isinstance(e, Exception):
1014 raise
1015 finally:
1016 if detach and is_gc_cleanup and dont_restore_gced:
1017 message = (
1018 "The garbage collector is trying to clean up "
1019 f"non-checked-in connection {dbapi_connection!r}, "
1020 f"""which will be {
1021 'dropped, as it cannot be safely terminated'
1022 if not can_close_or_terminate_connection
1023 else 'terminated'
1024 }. """
1025 "Please ensure that SQLAlchemy pooled connections are "
1026 "returned to "
1027 "the pool explicitly, either by calling ``close()`` "
1028 "or by using appropriate context managers to manage "
1029 "their lifecycle."
1030 )
1031 pool.logger.error(message)
1032 util.warn(message)
1033
1034 if connection_record and connection_record.fairy_ref is not None:
1035 connection_record.checkin()
1036
1037 # give gc some help. See
1038 # test/engine/test_pool.py::PoolEventsTest::test_checkin_event_gc[True]
1039 # which actually started failing when pytest warnings plugin was
1040 # turned on, due to util.warn() above
1041 if fairy is not None:
1042 fairy.dbapi_connection = None # type: ignore
1043 fairy._connection_record = None
1044 del dbapi_connection
1045 del connection_record
1046 del fairy
1047
1048
1049# a dictionary of the _ConnectionFairy weakrefs to _ConnectionRecord, so that
1050# GC under pypy will call ConnectionFairy finalizers. linked directly to the
1051# weakref that will empty itself when collected so that it should not create
1052# any unmanaged memory references.
1053_strong_ref_connection_records: Dict[
1054 weakref.ref[_ConnectionFairy], _ConnectionRecord
1055] = {}
1056
1057
1058class PoolProxiedConnection(ManagesConnection):
1059 """A connection-like adapter for a :pep:`249` DBAPI connection, which
1060 includes additional methods specific to the :class:`.Pool` implementation.
1061
1062 :class:`.PoolProxiedConnection` is the public-facing interface for the
1063 internal :class:`._ConnectionFairy` implementation object; users familiar
1064 with :class:`._ConnectionFairy` can consider this object to be equivalent.
1065
1066 .. versionadded:: 2.0 :class:`.PoolProxiedConnection` provides the public-
1067 facing interface for the :class:`._ConnectionFairy` internal class.
1068
1069 """
1070
1071 __slots__ = ()
1072
1073 if typing.TYPE_CHECKING:
1074
1075 def commit(self) -> None: ...
1076
1077 def cursor(self, *args: Any, **kwargs: Any) -> DBAPICursor: ...
1078
1079 def rollback(self) -> None: ...
1080
1081 def __getattr__(self, key: str) -> Any: ...
1082
1083 @property
1084 def is_valid(self) -> bool:
1085 """Return True if this :class:`.PoolProxiedConnection` still refers
1086 to an active DBAPI connection."""
1087
1088 raise NotImplementedError()
1089
1090 @property
1091 def is_detached(self) -> bool:
1092 """Return True if this :class:`.PoolProxiedConnection` is detached
1093 from its pool."""
1094
1095 raise NotImplementedError()
1096
1097 def detach(self) -> None:
1098 """Separate this connection from its Pool.
1099
1100 This means that the connection will no longer be returned to the
1101 pool when closed, and will instead be literally closed. The
1102 associated :class:`.ConnectionPoolEntry` is de-associated from this
1103 DBAPI connection.
1104
1105 Note that any overall connection limiting constraints imposed by a
1106 Pool implementation may be violated after a detach, as the detached
1107 connection is removed from the pool's knowledge and control.
1108
1109 """
1110
1111 raise NotImplementedError()
1112
1113 def close(self) -> None:
1114 """Release this connection back to the pool.
1115
1116 The :meth:`.PoolProxiedConnection.close` method shadows the
1117 :pep:`249` ``.close()`` method, altering its behavior to instead
1118 :term:`release` the proxied connection back to the connection pool.
1119
1120 Upon release to the pool, whether the connection stays "opened" and
1121 pooled in the Python process, versus actually closed out and removed
1122 from the Python process, is based on the pool implementation in use and
1123 its configuration and current state.
1124
1125 """
1126 raise NotImplementedError()
1127
1128 def __enter__(self) -> Self:
1129 return self
1130
1131 def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
1132 self.close()
1133 return None
1134
1135
1136class _AdhocProxiedConnection(PoolProxiedConnection):
1137 """provides the :class:`.PoolProxiedConnection` interface for cases where
1138 the DBAPI connection is not actually proxied.
1139
1140 This is used by the engine internals to pass a consistent
1141 :class:`.PoolProxiedConnection` object to consuming dialects in response to
1142 pool events that may not always have the :class:`._ConnectionFairy`
1143 available.
1144
1145 """
1146
1147 __slots__ = ("dbapi_connection", "_connection_record", "_is_valid")
1148
1149 dbapi_connection: DBAPIConnection
1150 _connection_record: ConnectionPoolEntry
1151
1152 def __init__(
1153 self,
1154 dbapi_connection: DBAPIConnection,
1155 connection_record: ConnectionPoolEntry,
1156 ):
1157 self.dbapi_connection = dbapi_connection
1158 self._connection_record = connection_record
1159 self._is_valid = True
1160
1161 @property
1162 def driver_connection(self) -> Any: # type: ignore[override] # mypy#4125
1163 return self._connection_record.driver_connection
1164
1165 @property
1166 def connection(self) -> DBAPIConnection:
1167 return self.dbapi_connection
1168
1169 @property
1170 def is_valid(self) -> bool:
1171 """Implement is_valid state attribute.
1172
1173 for the adhoc proxied connection it's assumed the connection is valid
1174 as there is no "invalidate" routine.
1175
1176 """
1177 return self._is_valid
1178
1179 def invalidate(
1180 self, e: Optional[BaseException] = None, soft: bool = False
1181 ) -> None:
1182 self._is_valid = False
1183
1184 @util.ro_non_memoized_property
1185 def record_info(self) -> Optional[_InfoType]:
1186 return self._connection_record.record_info
1187
1188 def cursor(self, *args: Any, **kwargs: Any) -> DBAPICursor:
1189 return self.dbapi_connection.cursor(*args, **kwargs)
1190
1191 def __getattr__(self, key: Any) -> Any:
1192 return getattr(self.dbapi_connection, key)
1193
1194
1195class _ConnectionFairy(PoolProxiedConnection):
1196 """Proxies a DBAPI connection and provides return-on-dereference
1197 support.
1198
1199 This is an internal object used by the :class:`_pool.Pool` implementation
1200 to provide context management to a DBAPI connection delivered by