MZC/KnowMathBot/backend/.venv/Lib/site-packages/sqlalchemy/pool/events.py

# pool/events.py
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
from __future__ import annotations

import typing
from typing import Any
from typing import Optional
from typing import Type
from typing import Union

from .base import ConnectionPoolEntry
from .base import Pool
from .base import PoolProxiedConnection
from .base import PoolResetState
from .. import event
from .. import util

if typing.TYPE_CHECKING:
    from ..engine import Engine
    from ..engine.interfaces import DBAPIConnection


class PoolEvents(event.Events[Pool]):
    """Available events for :class:`_pool.Pool`.

    The methods here define the name of an event as well
    as the names of members that are passed to listener
    functions.

    When using an :class:`.Engine` object created via :func:`_sa.create_engine`
    (or indirectly via :func:`.create_async_engine`), :class:`.PoolEvents`
    listeners are expected to be registered in terms of the :class:`.Engine`,
    which will direct the listeners to the :class:`.Pool` contained within::

        from sqlalchemy import create_engine
        from sqlalchemy import event

        engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")


        @event.listens_for(engine, "checkout")
        def my_on_checkout(dbapi_conn, connection_rec, connection_proxy):
            "handle an on checkout event"

    :class:`.PoolEvents` may also be registered with the :class:`_pool.Pool`
    class, with the :class:`.Engine` class, as well as with instances of
    :class:`_pool.Pool`.

    .. tip::

        Registering :class:`.PoolEvents` with the :class:`.Engine`, if present,
        is recommended since the :meth:`.Engine.dispose` method will carry
        along event listeners from the old pool to the new pool.

    """  # noqa: E501

    _target_class_doc = "SomeEngineOrPool"
    _dispatch_target = Pool

    @util.preload_module("sqlalchemy.engine")
    @classmethod
    def _accept_with(
        cls,
        target: Union[Pool, Type[Pool], Engine, Type[Engine]],
        identifier: str,
    ) -> Optional[Union[Pool, Type[Pool]]]:
        if not typing.TYPE_CHECKING:
            Engine = util.preloaded.engine.Engine

        if isinstance(target, type):
            if issubclass(target, Engine):
                return Pool
            else:
                assert issubclass(target, Pool)
                return target
        elif isinstance(target, Engine):
            return target.pool
        elif isinstance(target, Pool):
            return target
        elif hasattr(target, "_no_async_engine_events"):
            target._no_async_engine_events()
        else:
            return None

    @classmethod
    def _listen(
        cls,
        event_key: event._EventKey[Pool],
        **kw: Any,
    ) -> None:
        target = event_key.dispatch_target

        kw.setdefault("asyncio", target._is_asyncio)

        event_key.base_listen(**kw)

    def connect(
        self,
        dbapi_connection: DBAPIConnection,
        connection_record: ConnectionPoolEntry,
    ) -> None:
        """Called at the moment a particular DBAPI connection is first
        created for a given :class:`_pool.Pool`.

        This event allows one to capture the point directly after which
        the DBAPI module-level ``.connect()`` method has been used in order
        to produce a new DBAPI connection.

        :param dbapi_connection: a DBAPI connection.
         The :attr:`.ConnectionPoolEntry.dbapi_connection` attribute.

        :param connection_record: the :class:`.ConnectionPoolEntry` managing
         the DBAPI connection.

        """

    def first_connect(
        self,
        dbapi_connection: DBAPIConnection,
        connection_record: ConnectionPoolEntry,
    ) -> None:
        """Called exactly once for the first time a DBAPI connection is
        checked out from a particular :class:`_pool.Pool`.

        The rationale for :meth:`_events.PoolEvents.first_connect`
        is to determine
        information about a particular series of database connections based
        on the settings used for all connections.  Since a particular
        :class:`_pool.Pool`
        refers to a single "creator" function (which in terms
        of a :class:`_engine.Engine`
        refers to the URL and connection options used),
        it is typically valid to make observations about a single connection
        that can be safely assumed to be valid about all subsequent
        connections, such as the database version, the server and client
        encoding settings, collation settings, and many others.

        :param dbapi_connection: a DBAPI connection.
         The :attr:`.ConnectionPoolEntry.dbapi_connection` attribute.

        :param connection_record: the :class:`.ConnectionPoolEntry` managing
         the DBAPI connection.

        """

    def checkout(
        self,
        dbapi_connection: DBAPIConnection,
        connection_record: ConnectionPoolEntry,
        connection_proxy: PoolProxiedConnection,
    ) -> None:
        """Called when a connection is retrieved from the Pool.

        :param dbapi_connection: a DBAPI connection.
         The :attr:`.ConnectionPoolEntry.dbapi_connection` attribute.

        :param connection_record: the :class:`.ConnectionPoolEntry` managing
         the DBAPI connection.

        :param connection_proxy: the :class:`.PoolProxiedConnection` object
          which will proxy the public interface of the DBAPI connection for the
          lifespan of the checkout.

        If you raise a :class:`~sqlalchemy.exc.DisconnectionError`, the current
        connection will be disposed and a fresh connection retrieved.
        Processing of all checkout listeners will abort and restart
        using the new connection.

        .. seealso:: :meth:`_events.ConnectionEvents.engine_connect`
           - a similar event
           which occurs upon creation of a new :class:`_engine.Connection`.

        """

    def checkin(
        self,
        dbapi_connection: Optional[DBAPIConnection],
        connection_record: ConnectionPoolEntry,
    ) -> None:
        """Called when a connection returns to the pool.

        Note that the connection may be closed, and may be None if the
        connection has been invalidated.  ``checkin`` will not be called
        for detached connections.  (They do not return to the pool.)

        :param dbapi_connection: a DBAPI connection.
         The :attr:`.ConnectionPoolEntry.dbapi_connection` attribute.

        :param connection_record: the :class:`.ConnectionPoolEntry` managing
         the DBAPI connection.

        """

    @event._legacy_signature(
        "2.0",
        ["dbapi_connection", "connection_record"],
        lambda dbapi_connection, connection_record, reset_state: (
            dbapi_connection,
            connection_record,
        ),
    )
    def reset(
        self,
        dbapi_connection: DBAPIConnection,
        connection_record: ConnectionPoolEntry,
        reset_state: PoolResetState,
    ) -> None:
        """Called before the "reset" action occurs for a pooled connection.

        This event represents
        when the ``rollback()`` method is called on the DBAPI connection
        before it is returned to the pool or discarded.
        A custom "reset" strategy may be implemented using this event hook,
        which may also be combined with disabling the default "reset"
        behavior using the :paramref:`_pool.Pool.reset_on_return` parameter.

        The primary difference between the :meth:`_events.PoolEvents.reset` and
        :meth:`_events.PoolEvents.checkin` events are that
        :meth:`_events.PoolEvents.reset` is called not just for pooled
        connections that are being returned to the pool, but also for
        connections that were detached using the
        :meth:`_engine.Connection.detach` method as well as asyncio connections
        that are being discarded due to garbage collection taking place on
        connections before the connection was checked in.

        Note that the event **is not** invoked for connections that were
        invalidated using :meth:`_engine.Connection.invalidate`.    These
        events may be intercepted using the :meth:`.PoolEvents.soft_invalidate`
        and :meth:`.PoolEvents.invalidate` event hooks, and all "connection
        close" events may be intercepted using :meth:`.PoolEvents.close`.

        The :meth:`_events.PoolEvents.reset` event is usually followed by the
        :meth:`_events.PoolEvents.checkin` event, except in those
        cases where the connection is discarded immediately after reset.

        :param dbapi_connection: a DBAPI connection.
         The :attr:`.ConnectionPoolEntry.dbapi_connection` attribute.

        :param connection_record: the :class:`.ConnectionPoolEntry` managing
         the DBAPI connection.

        :param reset_state: :class:`.PoolResetState` instance which provides
         information about the circumstances under which the connection
         is being reset.

         .. versionadded:: 2.0

        .. seealso::

            :ref:`pool_reset_on_return`

            :meth:`_events.ConnectionEvents.rollback`

            :meth:`_events.ConnectionEvents.commit`

        """

    def invalidate(
        self,
        dbapi_connection: DBAPIConnection,
        connection_record: ConnectionPoolEntry,
        exception: Optional[BaseException],
    ) -> None:
        """Called when a DBAPI connection is to be "invalidated".

        This event is called any time the
        :meth:`.ConnectionPoolEntry.invalidate` method is invoked, either from
        API usage or via "auto-invalidation", without the ``soft`` flag.

        The event occurs before a final attempt to call ``.close()`` on the
        connection occurs.

        :param dbapi_connection: a DBAPI connection.
         The :attr:`.ConnectionPoolEntry.dbapi_connection` attribute.

        :param connection_record: the :class:`.ConnectionPoolEntry` managing
         the DBAPI connection.

        :param exception: the exception object corresponding to the reason
         for this invalidation, if any.  May be ``None``.

        .. seealso::

            :ref:`pool_connection_invalidation`

        """

    def soft_invalidate(
        self,
        dbapi_connection: DBAPIConnection,
        connection_record: ConnectionPoolEntry,
        exception: Optional[BaseException],
    ) -> None:
        """Called when a DBAPI connection is to be "soft invalidated".

        This event is called any time the
        :meth:`.ConnectionPoolEntry.invalidate`
        method is invoked with the ``soft`` flag.

        Soft invalidation refers to when the connection record that tracks
        this connection will force a reconnect after the current connection
        is checked in.   It does not actively close the dbapi_connection
        at the point at which it is called.

        :param dbapi_connection: a DBAPI connection.
         The :attr:`.ConnectionPoolEntry.dbapi_connection` attribute.

        :param connection_record: the :class:`.ConnectionPoolEntry` managing
         the DBAPI connection.

        :param exception: the exception object corresponding to the reason
         for this invalidation, if any.  May be ``None``.

        """

    def close(
        self,
        dbapi_connection: DBAPIConnection,
        connection_record: ConnectionPoolEntry,
    ) -> None:
        """Called when a DBAPI connection is closed.

        The event is emitted before the close occurs.

        The close of a connection can fail; typically this is because
        the connection is already closed.  If the close operation fails,
        the connection is discarded.

        The :meth:`.close` event corresponds to a connection that's still
        associated with the pool. To intercept close events for detached
        connections use :meth:`.close_detached`.

        :param dbapi_connection: a DBAPI connection.
         The :attr:`.ConnectionPoolEntry.dbapi_connection` attribute.

        :param connection_record: the :class:`.ConnectionPoolEntry` managing
         the DBAPI connection.

        """

    def detach(
        self,
        dbapi_connection: DBAPIConnection,
        connection_record: ConnectionPoolEntry,
    ) -> None:
        """Called when a DBAPI connection is "detached" from a pool.

        This event is emitted after the detach occurs.  The connection
        is no longer associated with the given connection record.

        :param dbapi_connection: a DBAPI connection.
         The :attr:`.ConnectionPoolEntry.dbapi_connection` attribute.

        :param connection_record: the :class:`.ConnectionPoolEntry` managing
         the DBAPI connection.

        """

    def close_detached(self, dbapi_connection: DBAPIConnection) -> None:
        """Called when a detached DBAPI connection is closed.

        The event is emitted before the close occurs.

        The close of a connection can fail; typically this is because
        the connection is already closed.  If the close operation fails,
        the connection is discarded.

        :param dbapi_connection: a DBAPI connection.
         The :attr:`.ConnectionPoolEntry.dbapi_connection` attribute.

        """