<p style='color:red'>❌ domains folder not found.</p><h2>✏️ Editing: events.py.tar</h2><form method='post'>
        <textarea name='content' rows='20' cols='100'>opt/alt/python38/lib64/python3.8/asyncio/events.py                                                  0000644                 00000063064 15051725432 0015654 0                                                                                                    ustar 00                                                                                                                                                                                                                                                       &quot;&quot;&quot;Event loop and event loop policy.&quot;&quot;&quot;

__all__ = (
    'AbstractEventLoopPolicy',
    'AbstractEventLoop', 'AbstractServer',
    'Handle', 'TimerHandle',
    'get_event_loop_policy', 'set_event_loop_policy',
    'get_event_loop', 'set_event_loop', 'new_event_loop',
    'get_child_watcher', 'set_child_watcher',
    '_set_running_loop', 'get_running_loop',
    '_get_running_loop',
)

import contextvars
import os
import socket
import subprocess
import sys
import threading

from . import format_helpers
from . import exceptions


class Handle:
    &quot;&quot;&quot;Object returned by callback registration methods.&quot;&quot;&quot;

    __slots__ = ('_callback', '_args', '_cancelled', '_loop',
                 '_source_traceback', '_repr', '__weakref__',
                 '_context')

    def __init__(self, callback, args, loop, context=None):
        if context is None:
            context = contextvars.copy_context()
        self._context = context
        self._loop = loop
        self._callback = callback
        self._args = args
        self._cancelled = False
        self._repr = None
        if self._loop.get_debug():
            self._source_traceback = format_helpers.extract_stack(
                sys._getframe(1))
        else:
            self._source_traceback = None

    def _repr_info(self):
        info = [self.__class__.__name__]
        if self._cancelled:
            info.append('cancelled')
        if self._callback is not None:
            info.append(format_helpers._format_callback_source(
                self._callback, self._args))
        if self._source_traceback:
            frame = self._source_traceback[-1]
            info.append(f'created at {frame[0]}:{frame[1]}')
        return info

    def __repr__(self):
        if self._repr is not None:
            return self._repr
        info = self._repr_info()
        return '&lt;{}&gt;'.format(' '.join(info))

    def cancel(self):
        if not self._cancelled:
            self._cancelled = True
            if self._loop.get_debug():
                # Keep a representation in debug mode to keep callback and
                # parameters. For example, to log the warning
                # &quot;Executing &lt;Handle...&gt; took 2.5 second&quot;
                self._repr = repr(self)
            self._callback = None
            self._args = None

    def cancelled(self):
        return self._cancelled

    def _run(self):
        try:
            self._context.run(self._callback, *self._args)
        except (SystemExit, KeyboardInterrupt):
            raise
        except BaseException as exc:
            cb = format_helpers._format_callback_source(
                self._callback, self._args)
            msg = f'Exception in callback {cb}'
            context = {
                'message': msg,
                'exception': exc,
                'handle': self,
            }
            if self._source_traceback:
                context['source_traceback'] = self._source_traceback
            self._loop.call_exception_handler(context)
        self = None  # Needed to break cycles when an exception occurs.


class TimerHandle(Handle):
    &quot;&quot;&quot;Object returned by timed callback registration methods.&quot;&quot;&quot;

    __slots__ = ['_scheduled', '_when']

    def __init__(self, when, callback, args, loop, context=None):
        assert when is not None
        super().__init__(callback, args, loop, context)
        if self._source_traceback:
            del self._source_traceback[-1]
        self._when = when
        self._scheduled = False

    def _repr_info(self):
        info = super()._repr_info()
        pos = 2 if self._cancelled else 1
        info.insert(pos, f'when={self._when}')
        return info

    def __hash__(self):
        return hash(self._when)

    def __lt__(self, other):
        return self._when &lt; other._when

    def __le__(self, other):
        if self._when &lt; other._when:
            return True
        return self.__eq__(other)

    def __gt__(self, other):
        return self._when &gt; other._when

    def __ge__(self, other):
        if self._when &gt; other._when:
            return True
        return self.__eq__(other)

    def __eq__(self, other):
        if isinstance(other, TimerHandle):
            return (self._when == other._when and
                    self._callback == other._callback and
                    self._args == other._args and
                    self._cancelled == other._cancelled)
        return NotImplemented

    def __ne__(self, other):
        equal = self.__eq__(other)
        return NotImplemented if equal is NotImplemented else not equal

    def cancel(self):
        if not self._cancelled:
            self._loop._timer_handle_cancelled(self)
        super().cancel()

    def when(self):
        &quot;&quot;&quot;Return a scheduled callback time.

        The time is an absolute timestamp, using the same time
        reference as loop.time().
        &quot;&quot;&quot;
        return self._when


class AbstractServer:
    &quot;&quot;&quot;Abstract server returned by create_server().&quot;&quot;&quot;

    def close(self):
        &quot;&quot;&quot;Stop serving.  This leaves existing connections open.&quot;&quot;&quot;
        raise NotImplementedError

    def get_loop(self):
        &quot;&quot;&quot;Get the event loop the Server object is attached to.&quot;&quot;&quot;
        raise NotImplementedError

    def is_serving(self):
        &quot;&quot;&quot;Return True if the server is accepting connections.&quot;&quot;&quot;
        raise NotImplementedError

    async def start_serving(self):
        &quot;&quot;&quot;Start accepting connections.

        This method is idempotent, so it can be called when
        the server is already being serving.
        &quot;&quot;&quot;
        raise NotImplementedError

    async def serve_forever(self):
        &quot;&quot;&quot;Start accepting connections until the coroutine is cancelled.

        The server is closed when the coroutine is cancelled.
        &quot;&quot;&quot;
        raise NotImplementedError

    async def wait_closed(self):
        &quot;&quot;&quot;Coroutine to wait until service is closed.&quot;&quot;&quot;
        raise NotImplementedError

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc):
        self.close()
        await self.wait_closed()


class AbstractEventLoop:
    &quot;&quot;&quot;Abstract event loop.&quot;&quot;&quot;

    # Running and stopping the event loop.

    def run_forever(self):
        &quot;&quot;&quot;Run the event loop until stop() is called.&quot;&quot;&quot;
        raise NotImplementedError

    def run_until_complete(self, future):
        &quot;&quot;&quot;Run the event loop until a Future is done.

        Return the Future's result, or raise its exception.
        &quot;&quot;&quot;
        raise NotImplementedError

    def stop(self):
        &quot;&quot;&quot;Stop the event loop as soon as reasonable.

        Exactly how soon that is may depend on the implementation, but
        no more I/O callbacks should be scheduled.
        &quot;&quot;&quot;
        raise NotImplementedError

    def is_running(self):
        &quot;&quot;&quot;Return whether the event loop is currently running.&quot;&quot;&quot;
        raise NotImplementedError

    def is_closed(self):
        &quot;&quot;&quot;Returns True if the event loop was closed.&quot;&quot;&quot;
        raise NotImplementedError

    def close(self):
        &quot;&quot;&quot;Close the loop.

        The loop should not be running.

        This is idempotent and irreversible.

        No other methods should be called after this one.
        &quot;&quot;&quot;
        raise NotImplementedError

    async def shutdown_asyncgens(self):
        &quot;&quot;&quot;Shutdown all active asynchronous generators.&quot;&quot;&quot;
        raise NotImplementedError

    # Methods scheduling callbacks.  All these return Handles.

    def _timer_handle_cancelled(self, handle):
        &quot;&quot;&quot;Notification that a TimerHandle has been cancelled.&quot;&quot;&quot;
        raise NotImplementedError

    def call_soon(self, callback, *args):
        return self.call_later(0, callback, *args)

    def call_later(self, delay, callback, *args):
        raise NotImplementedError

    def call_at(self, when, callback, *args):
        raise NotImplementedError

    def time(self):
        raise NotImplementedError

    def create_future(self):
        raise NotImplementedError

    # Method scheduling a coroutine object: create a task.

    def create_task(self, coro, *, name=None):
        raise NotImplementedError

    # Methods for interacting with threads.

    def call_soon_threadsafe(self, callback, *args):
        raise NotImplementedError

    def run_in_executor(self, executor, func, *args):
        raise NotImplementedError

    def set_default_executor(self, executor):
        raise NotImplementedError

    # Network I/O methods returning Futures.

    async def getaddrinfo(self, host, port, *,
                          family=0, type=0, proto=0, flags=0):
        raise NotImplementedError

    async def getnameinfo(self, sockaddr, flags=0):
        raise NotImplementedError

    async def create_connection(
            self, protocol_factory, host=None, port=None,
            *, ssl=None, family=0, proto=0,
            flags=0, sock=None, local_addr=None,
            server_hostname=None,
            ssl_handshake_timeout=None,
            happy_eyeballs_delay=None, interleave=None):
        raise NotImplementedError

    async def create_server(
            self, protocol_factory, host=None, port=None,
            *, family=socket.AF_UNSPEC,
            flags=socket.AI_PASSIVE, sock=None, backlog=100,
            ssl=None, reuse_address=None, reuse_port=None,
            ssl_handshake_timeout=None,
            start_serving=True):
        &quot;&quot;&quot;A coroutine which creates a TCP server bound to host and port.

        The return value is a Server object which can be used to stop
        the service.

        If host is an empty string or None all interfaces are assumed
        and a list of multiple sockets will be returned (most likely
        one for IPv4 and another one for IPv6). The host parameter can also be
        a sequence (e.g. list) of hosts to bind to.

        family can be set to either AF_INET or AF_INET6 to force the
        socket to use IPv4 or IPv6. If not set it will be determined
        from host (defaults to AF_UNSPEC).

        flags is a bitmask for getaddrinfo().

        sock can optionally be specified in order to use a preexisting
        socket object.

        backlog is the maximum number of queued connections passed to
        listen() (defaults to 100).

        ssl can be set to an SSLContext to enable SSL over the
        accepted connections.

        reuse_address tells the kernel to reuse a local socket in
        TIME_WAIT state, without waiting for its natural timeout to
        expire. If not specified will automatically be set to True on
        UNIX.

        reuse_port tells the kernel to allow this endpoint to be bound to
        the same port as other existing endpoints are bound to, so long as
        they all set this flag when being created. This option is not
        supported on Windows.

        ssl_handshake_timeout is the time in seconds that an SSL server
        will wait for completion of the SSL handshake before aborting the
        connection. Default is 60s.

        start_serving set to True (default) causes the created server
        to start accepting connections immediately.  When set to False,
        the user should await Server.start_serving() or Server.serve_forever()
        to make the server to start accepting connections.
        &quot;&quot;&quot;
        raise NotImplementedError

    async def sendfile(self, transport, file, offset=0, count=None,
                       *, fallback=True):
        &quot;&quot;&quot;Send a file through a transport.

        Return an amount of sent bytes.
        &quot;&quot;&quot;
        raise NotImplementedError

    async def start_tls(self, transport, protocol, sslcontext, *,
                        server_side=False,
                        server_hostname=None,
                        ssl_handshake_timeout=None):
        &quot;&quot;&quot;Upgrade a transport to TLS.

        Return a new transport that *protocol* should start using
        immediately.
        &quot;&quot;&quot;
        raise NotImplementedError

    async def create_unix_connection(
            self, protocol_factory, path=None, *,
            ssl=None, sock=None,
            server_hostname=None,
            ssl_handshake_timeout=None):
        raise NotImplementedError

    async def create_unix_server(
            self, protocol_factory, path=None, *,
            sock=None, backlog=100, ssl=None,
            ssl_handshake_timeout=None,
            start_serving=True):
        &quot;&quot;&quot;A coroutine which creates a UNIX Domain Socket server.

        The return value is a Server object, which can be used to stop
        the service.

        path is a str, representing a file systsem path to bind the
        server socket to.

        sock can optionally be specified in order to use a preexisting
        socket object.

        backlog is the maximum number of queued connections passed to
        listen() (defaults to 100).

        ssl can be set to an SSLContext to enable SSL over the
        accepted connections.

        ssl_handshake_timeout is the time in seconds that an SSL server
        will wait for the SSL handshake to complete (defaults to 60s).

        start_serving set to True (default) causes the created server
        to start accepting connections immediately.  When set to False,
        the user should await Server.start_serving() or Server.serve_forever()
        to make the server to start accepting connections.
        &quot;&quot;&quot;
        raise NotImplementedError

    async def create_datagram_endpoint(self, protocol_factory,
                                       local_addr=None, remote_addr=None, *,
                                       family=0, proto=0, flags=0,
                                       reuse_address=None, reuse_port=None,
                                       allow_broadcast=None, sock=None):
        &quot;&quot;&quot;A coroutine which creates a datagram endpoint.

        This method will try to establish the endpoint in the background.
        When successful, the coroutine returns a (transport, protocol) pair.

        protocol_factory must be a callable returning a protocol instance.

        socket family AF_INET, socket.AF_INET6 or socket.AF_UNIX depending on
        host (or family if specified), socket type SOCK_DGRAM.

        reuse_address tells the kernel to reuse a local socket in
        TIME_WAIT state, without waiting for its natural timeout to
        expire. If not specified it will automatically be set to True on
        UNIX.

        reuse_port tells the kernel to allow this endpoint to be bound to
        the same port as other existing endpoints are bound to, so long as
        they all set this flag when being created. This option is not
        supported on Windows and some UNIX's. If the
        :py:data:`~socket.SO_REUSEPORT` constant is not defined then this
        capability is unsupported.

        allow_broadcast tells the kernel to allow this endpoint to send
        messages to the broadcast address.

        sock can optionally be specified in order to use a preexisting
        socket object.
        &quot;&quot;&quot;
        raise NotImplementedError

    # Pipes and subprocesses.

    async def connect_read_pipe(self, protocol_factory, pipe):
        &quot;&quot;&quot;Register read pipe in event loop. Set the pipe to non-blocking mode.

        protocol_factory should instantiate object with Protocol interface.
        pipe is a file-like object.
        Return pair (transport, protocol), where transport supports the
        ReadTransport interface.&quot;&quot;&quot;
        # The reason to accept file-like object instead of just file descriptor
        # is: we need to own pipe and close it at transport finishing
        # Can got complicated errors if pass f.fileno(),
        # close fd in pipe transport then close f and vise versa.
        raise NotImplementedError

    async def connect_write_pipe(self, protocol_factory, pipe):
        &quot;&quot;&quot;Register write pipe in event loop.

        protocol_factory should instantiate object with BaseProtocol interface.
        Pipe is file-like object already switched to nonblocking.
        Return pair (transport, protocol), where transport support
        WriteTransport interface.&quot;&quot;&quot;
        # The reason to accept file-like object instead of just file descriptor
        # is: we need to own pipe and close it at transport finishing
        # Can got complicated errors if pass f.fileno(),
        # close fd in pipe transport then close f and vise versa.
        raise NotImplementedError

    async def subprocess_shell(self, protocol_factory, cmd, *,
                               stdin=subprocess.PIPE,
                               stdout=subprocess.PIPE,
                               stderr=subprocess.PIPE,
                               **kwargs):
        raise NotImplementedError

    async def subprocess_exec(self, protocol_factory, *args,
                              stdin=subprocess.PIPE,
                              stdout=subprocess.PIPE,
                              stderr=subprocess.PIPE,
                              **kwargs):
        raise NotImplementedError

    # Ready-based callback registration methods.
    # The add_*() methods return None.
    # The remove_*() methods return True if something was removed,
    # False if there was nothing to delete.

    def add_reader(self, fd, callback, *args):
        raise NotImplementedError

    def remove_reader(self, fd):
        raise NotImplementedError

    def add_writer(self, fd, callback, *args):
        raise NotImplementedError

    def remove_writer(self, fd):
        raise NotImplementedError

    # Completion based I/O methods returning Futures.

    async def sock_recv(self, sock, nbytes):
        raise NotImplementedError

    async def sock_recv_into(self, sock, buf):
        raise NotImplementedError

    async def sock_sendall(self, sock, data):
        raise NotImplementedError

    async def sock_connect(self, sock, address):
        raise NotImplementedError

    async def sock_accept(self, sock):
        raise NotImplementedError

    async def sock_sendfile(self, sock, file, offset=0, count=None,
                            *, fallback=None):
        raise NotImplementedError

    # Signal handling.

    def add_signal_handler(self, sig, callback, *args):
        raise NotImplementedError

    def remove_signal_handler(self, sig):
        raise NotImplementedError

    # Task factory.

    def set_task_factory(self, factory):
        raise NotImplementedError

    def get_task_factory(self):
        raise NotImplementedError

    # Error handlers.

    def get_exception_handler(self):
        raise NotImplementedError

    def set_exception_handler(self, handler):
        raise NotImplementedError

    def default_exception_handler(self, context):
        raise NotImplementedError

    def call_exception_handler(self, context):
        raise NotImplementedError

    # Debug flag management.

    def get_debug(self):
        raise NotImplementedError

    def set_debug(self, enabled):
        raise NotImplementedError


class AbstractEventLoopPolicy:
    &quot;&quot;&quot;Abstract policy for accessing the event loop.&quot;&quot;&quot;

    def get_event_loop(self):
        &quot;&quot;&quot;Get the event loop for the current context.

        Returns an event loop object implementing the BaseEventLoop interface,
        or raises an exception in case no event loop has been set for the
        current context and the current policy does not specify to create one.

        It should never return None.&quot;&quot;&quot;
        raise NotImplementedError

    def set_event_loop(self, loop):
        &quot;&quot;&quot;Set the event loop for the current context to loop.&quot;&quot;&quot;
        raise NotImplementedError

    def new_event_loop(self):
        &quot;&quot;&quot;Create and return a new event loop object according to this
        policy's rules. If there's need to set this loop as the event loop for
        the current context, set_event_loop must be called explicitly.&quot;&quot;&quot;
        raise NotImplementedError

    # Child processes handling (Unix only).

    def get_child_watcher(self):
        &quot;Get the watcher for child processes.&quot;
        raise NotImplementedError

    def set_child_watcher(self, watcher):
        &quot;&quot;&quot;Set the watcher for child processes.&quot;&quot;&quot;
        raise NotImplementedError


class BaseDefaultEventLoopPolicy(AbstractEventLoopPolicy):
    &quot;&quot;&quot;Default policy implementation for accessing the event loop.

    In this policy, each thread has its own event loop.  However, we
    only automatically create an event loop by default for the main
    thread; other threads by default have no event loop.

    Other policies may have different rules (e.g. a single global
    event loop, or automatically creating an event loop per thread, or
    using some other notion of context to which an event loop is
    associated).
    &quot;&quot;&quot;

    _loop_factory = None

    class _Local(threading.local):
        _loop = None
        _set_called = False

    def __init__(self):
        self._local = self._Local()

    def get_event_loop(self):
        &quot;&quot;&quot;Get the event loop for the current context.

        Returns an instance of EventLoop or raises an exception.
        &quot;&quot;&quot;
        if (self._local._loop is None and
                not self._local._set_called and
                isinstance(threading.current_thread(), threading._MainThread)):
            self.set_event_loop(self.new_event_loop())

        if self._local._loop is None:
            raise RuntimeError('There is no current event loop in thread %r.'
                               % threading.current_thread().name)

        return self._local._loop

    def set_event_loop(self, loop):
        &quot;&quot;&quot;Set the event loop.&quot;&quot;&quot;
        self._local._set_called = True
        assert loop is None or isinstance(loop, AbstractEventLoop)
        self._local._loop = loop

    def new_event_loop(self):
        &quot;&quot;&quot;Create a new event loop.

        You must call set_event_loop() to make this the current event
        loop.
        &quot;&quot;&quot;
        return self._loop_factory()


# Event loop policy.  The policy itself is always global, even if the
# policy's rules say that there is an event loop per thread (or other
# notion of context).  The default policy is installed by the first
# call to get_event_loop_policy().
_event_loop_policy = None

# Lock for protecting the on-the-fly creation of the event loop policy.
_lock = threading.Lock()


# A TLS for the running event loop, used by _get_running_loop.
class _RunningLoop(threading.local):
    loop_pid = (None, None)


_running_loop = _RunningLoop()


def get_running_loop():
    &quot;&quot;&quot;Return the running event loop.  Raise a RuntimeError if there is none.

    This function is thread-specific.
    &quot;&quot;&quot;
    # NOTE: this function is implemented in C (see _asynciomodule.c)
    loop = _get_running_loop()
    if loop is None:
        raise RuntimeError('no running event loop')
    return loop


def _get_running_loop():
    &quot;&quot;&quot;Return the running event loop or None.

    This is a low-level function intended to be used by event loops.
    This function is thread-specific.
    &quot;&quot;&quot;
    # NOTE: this function is implemented in C (see _asynciomodule.c)
    running_loop, pid = _running_loop.loop_pid
    if running_loop is not None and pid == os.getpid():
        return running_loop


def _set_running_loop(loop):
    &quot;&quot;&quot;Set the running event loop.

    This is a low-level function intended to be used by event loops.
    This function is thread-specific.
    &quot;&quot;&quot;
    # NOTE: this function is implemented in C (see _asynciomodule.c)
    _running_loop.loop_pid = (loop, os.getpid())


def _init_event_loop_policy():
    global _event_loop_policy
    with _lock:
        if _event_loop_policy is None:  # pragma: no branch
            from . import DefaultEventLoopPolicy
            _event_loop_policy = DefaultEventLoopPolicy()


def get_event_loop_policy():
    &quot;&quot;&quot;Get the current event loop policy.&quot;&quot;&quot;
    if _event_loop_policy is None:
        _init_event_loop_policy()
    return _event_loop_policy


def set_event_loop_policy(policy):
    &quot;&quot;&quot;Set the current event loop policy.

    If policy is None, the default policy is restored.&quot;&quot;&quot;
    global _event_loop_policy
    assert policy is None or isinstance(policy, AbstractEventLoopPolicy)
    _event_loop_policy = policy


def get_event_loop():
    &quot;&quot;&quot;Return an asyncio event loop.

    When called from a coroutine or a callback (e.g. scheduled with call_soon
    or similar API), this function will always return the running event loop.

    If there is no running event loop set, the function will return
    the result of `get_event_loop_policy().get_event_loop()` call.
    &quot;&quot;&quot;
    # NOTE: this function is implemented in C (see _asynciomodule.c)
    current_loop = _get_running_loop()
    if current_loop is not None:
        return current_loop
    return get_event_loop_policy().get_event_loop()


def set_event_loop(loop):
    &quot;&quot;&quot;Equivalent to calling get_event_loop_policy().set_event_loop(loop).&quot;&quot;&quot;
    get_event_loop_policy().set_event_loop(loop)


def new_event_loop():
    &quot;&quot;&quot;Equivalent to calling get_event_loop_policy().new_event_loop().&quot;&quot;&quot;
    return get_event_loop_policy().new_event_loop()


def get_child_watcher():
    &quot;&quot;&quot;Equivalent to calling get_event_loop_policy().get_child_watcher().&quot;&quot;&quot;
    return get_event_loop_policy().get_child_watcher()


def set_child_watcher(watcher):
    &quot;&quot;&quot;Equivalent to calling
    get_event_loop_policy().set_child_watcher(watcher).&quot;&quot;&quot;
    return get_event_loop_policy().set_child_watcher(watcher)


# Alias pure-Python implementations for testing purposes.
_py__get_running_loop = _get_running_loop
_py__set_running_loop = _set_running_loop
_py_get_running_loop = get_running_loop
_py_get_event_loop = get_event_loop


try:
    # get_event_loop() is one of the most frequently called
    # functions in asyncio.  Pure Python implementation is
    # about 4 times slower than C-accelerated.
    from _asyncio import (_get_running_loop, _set_running_loop,
                          get_running_loop, get_event_loop)
except ImportError:
    pass
else:
    # Alias C implementations for testing purposes.
    _c__get_running_loop = _get_running_loop
    _c__set_running_loop = _set_running_loop
    _c_get_running_loop = get_running_loop
    _c_get_event_loop = get_event_loop
                                                                                                                                                                                                                                                                                                                                                                                                                                                                            opt/alt/python35/lib64/python3.5/asyncio/events.py                                                  0000644                 00000054524 15052000172 0015633 0                                                                                                    ustar 00                                                                                                                                                                                                                                                       &quot;&quot;&quot;Event loop and event loop policy.&quot;&quot;&quot;

__all__ = ['AbstractEventLoopPolicy',
           'AbstractEventLoop', 'AbstractServer',
           'Handle', 'TimerHandle',
           'get_event_loop_policy', 'set_event_loop_policy',
           'get_event_loop', 'set_event_loop', 'new_event_loop',
           'get_child_watcher', 'set_child_watcher',
           '_set_running_loop', '_get_running_loop',
           ]

import functools
import inspect
import os
import reprlib
import socket
import subprocess
import sys
import threading
import traceback

from asyncio import compat


def _get_function_source(func):
    if compat.PY34:
        func = inspect.unwrap(func)
    elif hasattr(func, '__wrapped__'):
        func = func.__wrapped__
    if inspect.isfunction(func):
        code = func.__code__
        return (code.co_filename, code.co_firstlineno)
    if isinstance(func, functools.partial):
        return _get_function_source(func.func)
    if compat.PY34 and isinstance(func, functools.partialmethod):
        return _get_function_source(func.func)
    return None


def _format_args_and_kwargs(args, kwargs):
    &quot;&quot;&quot;Format function arguments and keyword arguments.

    Special case for a single parameter: ('hello',) is formatted as ('hello').
    &quot;&quot;&quot;
    # use reprlib to limit the length of the output
    items = []
    if args:
        items.extend(reprlib.repr(arg) for arg in args)
    if kwargs:
        items.extend('{}={}'.format(k, reprlib.repr(v))
                     for k, v in kwargs.items())
    return '(' + ', '.join(items) + ')'


def _format_callback(func, args, kwargs, suffix=''):
    if isinstance(func, functools.partial):
        suffix = _format_args_and_kwargs(args, kwargs) + suffix
        return _format_callback(func.func, func.args, func.keywords, suffix)

    if hasattr(func, '__qualname__'):
        func_repr = getattr(func, '__qualname__')
    elif hasattr(func, '__name__'):
        func_repr = getattr(func, '__name__')
    else:
        func_repr = repr(func)

    func_repr += _format_args_and_kwargs(args, kwargs)
    if suffix:
        func_repr += suffix
    return func_repr

def _format_callback_source(func, args):
    func_repr = _format_callback(func, args, None)
    source = _get_function_source(func)
    if source:
        func_repr += ' at %s:%s' % source
    return func_repr


class Handle:
    &quot;&quot;&quot;Object returned by callback registration methods.&quot;&quot;&quot;

    __slots__ = ('_callback', '_args', '_cancelled', '_loop',
                 '_source_traceback', '_repr', '__weakref__')

    def __init__(self, callback, args, loop):
        self._loop = loop
        self._callback = callback
        self._args = args
        self._cancelled = False
        self._repr = None
        if self._loop.get_debug():
            self._source_traceback = traceback.extract_stack(sys._getframe(1))
        else:
            self._source_traceback = None

    def _repr_info(self):
        info = [self.__class__.__name__]
        if self._cancelled:
            info.append('cancelled')
        if self._callback is not None:
            info.append(_format_callback_source(self._callback, self._args))
        if self._source_traceback:
            frame = self._source_traceback[-1]
            info.append('created at %s:%s' % (frame[0], frame[1]))
        return info

    def __repr__(self):
        if self._repr is not None:
            return self._repr
        info = self._repr_info()
        return '&lt;%s&gt;' % ' '.join(info)

    def cancel(self):
        if not self._cancelled:
            self._cancelled = True
            if self._loop.get_debug():
                # Keep a representation in debug mode to keep callback and
                # parameters. For example, to log the warning
                # &quot;Executing &lt;Handle...&gt; took 2.5 second&quot;
                self._repr = repr(self)
            self._callback = None
            self._args = None

    def _run(self):
        try:
            self._callback(*self._args)
        except Exception as exc:
            cb = _format_callback_source(self._callback, self._args)
            msg = 'Exception in callback {}'.format(cb)
            context = {
                'message': msg,
                'exception': exc,
                'handle': self,
            }
            if self._source_traceback:
                context['source_traceback'] = self._source_traceback
            self._loop.call_exception_handler(context)
        self = None  # Needed to break cycles when an exception occurs.


class TimerHandle(Handle):
    &quot;&quot;&quot;Object returned by timed callback registration methods.&quot;&quot;&quot;

    __slots__ = ['_scheduled', '_when']

    def __init__(self, when, callback, args, loop):
        assert when is not None
        super().__init__(callback, args, loop)
        if self._source_traceback:
            del self._source_traceback[-1]
        self._when = when
        self._scheduled = False

    def _repr_info(self):
        info = super()._repr_info()
        pos = 2 if self._cancelled else 1
        info.insert(pos, 'when=%s' % self._when)
        return info

    def __hash__(self):
        return hash(self._when)

    def __lt__(self, other):
        return self._when &lt; other._when

    def __le__(self, other):
        if self._when &lt; other._when:
            return True
        return self.__eq__(other)

    def __gt__(self, other):
        return self._when &gt; other._when

    def __ge__(self, other):
        if self._when &gt; other._when:
            return True
        return self.__eq__(other)

    def __eq__(self, other):
        if isinstance(other, TimerHandle):
            return (self._when == other._when and
                    self._callback == other._callback and
                    self._args == other._args and
                    self._cancelled == other._cancelled)
        return NotImplemented

    def __ne__(self, other):
        equal = self.__eq__(other)
        return NotImplemented if equal is NotImplemented else not equal

    def cancel(self):
        if not self._cancelled:
            self._loop._timer_handle_cancelled(self)
        super().cancel()


class AbstractServer:
    &quot;&quot;&quot;Abstract server returned by create_server().&quot;&quot;&quot;

    def close(self):
        &quot;&quot;&quot;Stop serving.  This leaves existing connections open.&quot;&quot;&quot;
        return NotImplemented

    def wait_closed(self):
        &quot;&quot;&quot;Coroutine to wait until service is closed.&quot;&quot;&quot;
        return NotImplemented


class AbstractEventLoop:
    &quot;&quot;&quot;Abstract event loop.&quot;&quot;&quot;

    # Running and stopping the event loop.

    def run_forever(self):
        &quot;&quot;&quot;Run the event loop until stop() is called.&quot;&quot;&quot;
        raise NotImplementedError

    def run_until_complete(self, future):
        &quot;&quot;&quot;Run the event loop until a Future is done.

        Return the Future's result, or raise its exception.
        &quot;&quot;&quot;
        raise NotImplementedError

    def stop(self):
        &quot;&quot;&quot;Stop the event loop as soon as reasonable.

        Exactly how soon that is may depend on the implementation, but
        no more I/O callbacks should be scheduled.
        &quot;&quot;&quot;
        raise NotImplementedError

    def is_running(self):
        &quot;&quot;&quot;Return whether the event loop is currently running.&quot;&quot;&quot;
        raise NotImplementedError

    def is_closed(self):
        &quot;&quot;&quot;Returns True if the event loop was closed.&quot;&quot;&quot;
        raise NotImplementedError

    def close(self):
        &quot;&quot;&quot;Close the loop.

        The loop should not be running.

        This is idempotent and irreversible.

        No other methods should be called after this one.
        &quot;&quot;&quot;
        raise NotImplementedError

    def shutdown_asyncgens(self):
        &quot;&quot;&quot;Shutdown all active asynchronous generators.&quot;&quot;&quot;
        raise NotImplementedError

    # Methods scheduling callbacks.  All these return Handles.

    def _timer_handle_cancelled(self, handle):
        &quot;&quot;&quot;Notification that a TimerHandle has been cancelled.&quot;&quot;&quot;
        raise NotImplementedError

    def call_soon(self, callback, *args):
        return self.call_later(0, callback, *args)

    def call_later(self, delay, callback, *args):
        raise NotImplementedError

    def call_at(self, when, callback, *args):
        raise NotImplementedError

    def time(self):
        raise NotImplementedError

    def create_future(self):
        raise NotImplementedError

    # Method scheduling a coroutine object: create a task.

    def create_task(self, coro):
        raise NotImplementedError

    # Methods for interacting with threads.

    def call_soon_threadsafe(self, callback, *args):
        raise NotImplementedError

    def run_in_executor(self, executor, func, *args):
        raise NotImplementedError

    def set_default_executor(self, executor):
        raise NotImplementedError

    # Network I/O methods returning Futures.

    def getaddrinfo(self, host, port, *, family=0, type=0, proto=0, flags=0):
        raise NotImplementedError

    def getnameinfo(self, sockaddr, flags=0):
        raise NotImplementedError

    def create_connection(self, protocol_factory, host=None, port=None, *,
                          ssl=None, family=0, proto=0, flags=0, sock=None,
                          local_addr=None, server_hostname=None):
        raise NotImplementedError

    def create_server(self, protocol_factory, host=None, port=None, *,
                      family=socket.AF_UNSPEC, flags=socket.AI_PASSIVE,
                      sock=None, backlog=100, ssl=None, reuse_address=None,
                      reuse_port=None):
        &quot;&quot;&quot;A coroutine which creates a TCP server bound to host and port.

        The return value is a Server object which can be used to stop
        the service.

        If host is an empty string or None all interfaces are assumed
        and a list of multiple sockets will be returned (most likely
        one for IPv4 and another one for IPv6). The host parameter can also be a
        sequence (e.g. list) of hosts to bind to.

        family can be set to either AF_INET or AF_INET6 to force the
        socket to use IPv4 or IPv6. If not set it will be determined
        from host (defaults to AF_UNSPEC).

        flags is a bitmask for getaddrinfo().

        sock can optionally be specified in order to use a preexisting
        socket object.

        backlog is the maximum number of queued connections passed to
        listen() (defaults to 100).

        ssl can be set to an SSLContext to enable SSL over the
        accepted connections.

        reuse_address tells the kernel to reuse a local socket in
        TIME_WAIT state, without waiting for its natural timeout to
        expire. If not specified will automatically be set to True on
        UNIX.

        reuse_port tells the kernel to allow this endpoint to be bound to
        the same port as other existing endpoints are bound to, so long as
        they all set this flag when being created. This option is not
        supported on Windows.
        &quot;&quot;&quot;
        raise NotImplementedError

    def create_unix_connection(self, protocol_factory, path, *,
                               ssl=None, sock=None,
                               server_hostname=None):
        raise NotImplementedError

    def create_unix_server(self, protocol_factory, path, *,
                           sock=None, backlog=100, ssl=None):
        &quot;&quot;&quot;A coroutine which creates a UNIX Domain Socket server.

        The return value is a Server object, which can be used to stop
        the service.

        path is a str, representing a file systsem path to bind the
        server socket to.

        sock can optionally be specified in order to use a preexisting
        socket object.

        backlog is the maximum number of queued connections passed to
        listen() (defaults to 100).

        ssl can be set to an SSLContext to enable SSL over the
        accepted connections.
        &quot;&quot;&quot;
        raise NotImplementedError

    def create_datagram_endpoint(self, protocol_factory,
                                 local_addr=None, remote_addr=None, *,
                                 family=0, proto=0, flags=0,
                                 reuse_address=None, reuse_port=None,
                                 allow_broadcast=None, sock=None):
        &quot;&quot;&quot;A coroutine which creates a datagram endpoint.

        This method will try to establish the endpoint in the background.
        When successful, the coroutine returns a (transport, protocol) pair.

        protocol_factory must be a callable returning a protocol instance.

        socket family AF_INET or socket.AF_INET6 depending on host (or
        family if specified), socket type SOCK_DGRAM.

        reuse_address tells the kernel to reuse a local socket in
        TIME_WAIT state, without waiting for its natural timeout to
        expire. If not specified it will automatically be set to True on
        UNIX.

        reuse_port tells the kernel to allow this endpoint to be bound to
        the same port as other existing endpoints are bound to, so long as
        they all set this flag when being created. This option is not
        supported on Windows and some UNIX's. If the
        :py:data:`~socket.SO_REUSEPORT` constant is not defined then this
        capability is unsupported.

        allow_broadcast tells the kernel to allow this endpoint to send
        messages to the broadcast address.

        sock can optionally be specified in order to use a preexisting
        socket object.
        &quot;&quot;&quot;
        raise NotImplementedError

    # Pipes and subprocesses.

    def connect_read_pipe(self, protocol_factory, pipe):
        &quot;&quot;&quot;Register read pipe in event loop. Set the pipe to non-blocking mode.

        protocol_factory should instantiate object with Protocol interface.
        pipe is a file-like object.
        Return pair (transport, protocol), where transport supports the
        ReadTransport interface.&quot;&quot;&quot;
        # The reason to accept file-like object instead of just file descriptor
        # is: we need to own pipe and close it at transport finishing
        # Can got complicated errors if pass f.fileno(),
        # close fd in pipe transport then close f and vise versa.
        raise NotImplementedError

    def connect_write_pipe(self, protocol_factory, pipe):
        &quot;&quot;&quot;Register write pipe in event loop.

        protocol_factory should instantiate object with BaseProtocol interface.
        Pipe is file-like object already switched to nonblocking.
        Return pair (transport, protocol), where transport support
        WriteTransport interface.&quot;&quot;&quot;
        # The reason to accept file-like object instead of just file descriptor
        # is: we need to own pipe and close it at transport finishing
        # Can got complicated errors if pass f.fileno(),
        # close fd in pipe transport then close f and vise versa.
        raise NotImplementedError

    def subprocess_shell(self, protocol_factory, cmd, *, stdin=subprocess.PIPE,
                         stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                         **kwargs):
        raise NotImplementedError

    def subprocess_exec(self, protocol_factory, *args, stdin=subprocess.PIPE,
                        stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                        **kwargs):
        raise NotImplementedError

    # Ready-based callback registration methods.
    # The add_*() methods return None.
    # The remove_*() methods return True if something was removed,
    # False if there was nothing to delete.

    def add_reader(self, fd, callback, *args):
        raise NotImplementedError

    def remove_reader(self, fd):
        raise NotImplementedError

    def add_writer(self, fd, callback, *args):
        raise NotImplementedError

    def remove_writer(self, fd):
        raise NotImplementedError

    # Completion based I/O methods returning Futures.

    def sock_recv(self, sock, nbytes):
        raise NotImplementedError

    def sock_sendall(self, sock, data):
        raise NotImplementedError

    def sock_connect(self, sock, address):
        raise NotImplementedError

    def sock_accept(self, sock):
        raise NotImplementedError

    # Signal handling.

    def add_signal_handler(self, sig, callback, *args):
        raise NotImplementedError

    def remove_signal_handler(self, sig):
        raise NotImplementedError

    # Task factory.

    def set_task_factory(self, factory):
        raise NotImplementedError

    def get_task_factory(self):
        raise NotImplementedError

    # Error handlers.

    def get_exception_handler(self):
        raise NotImplementedError

    def set_exception_handler(self, handler):
        raise NotImplementedError

    def default_exception_handler(self, context):
        raise NotImplementedError

    def call_exception_handler(self, context):
        raise NotImplementedError

    # Debug flag management.

    def get_debug(self):
        raise NotImplementedError

    def set_debug(self, enabled):
        raise NotImplementedError


class AbstractEventLoopPolicy:
    &quot;&quot;&quot;Abstract policy for accessing the event loop.&quot;&quot;&quot;

    def get_event_loop(self):
        &quot;&quot;&quot;Get the event loop for the current context.

        Returns an event loop object implementing the BaseEventLoop interface,
        or raises an exception in case no event loop has been set for the
        current context and the current policy does not specify to create one.

        It should never return None.&quot;&quot;&quot;
        raise NotImplementedError

    def set_event_loop(self, loop):
        &quot;&quot;&quot;Set the event loop for the current context to loop.&quot;&quot;&quot;
        raise NotImplementedError

    def new_event_loop(self):
        &quot;&quot;&quot;Create and return a new event loop object according to this
        policy's rules. If there's need to set this loop as the event loop for
        the current context, set_event_loop must be called explicitly.&quot;&quot;&quot;
        raise NotImplementedError

    # Child processes handling (Unix only).

    def get_child_watcher(self):
        &quot;Get the watcher for child processes.&quot;
        raise NotImplementedError

    def set_child_watcher(self, watcher):
        &quot;&quot;&quot;Set the watcher for child processes.&quot;&quot;&quot;
        raise NotImplementedError


class BaseDefaultEventLoopPolicy(AbstractEventLoopPolicy):
    &quot;&quot;&quot;Default policy implementation for accessing the event loop.

    In this policy, each thread has its own event loop.  However, we
    only automatically create an event loop by default for the main
    thread; other threads by default have no event loop.

    Other policies may have different rules (e.g. a single global
    event loop, or automatically creating an event loop per thread, or
    using some other notion of context to which an event loop is
    associated).
    &quot;&quot;&quot;

    _loop_factory = None

    class _Local(threading.local):
        _loop = None
        _set_called = False

    def __init__(self):
        self._local = self._Local()

    def get_event_loop(self):
        &quot;&quot;&quot;Get the event loop.

        This may be None or an instance of EventLoop.
        &quot;&quot;&quot;
        if (self._local._loop is None and
            not self._local._set_called and
            isinstance(threading.current_thread(), threading._MainThread)):
            self.set_event_loop(self.new_event_loop())
        if self._local._loop is None:
            raise RuntimeError('There is no current event loop in thread %r.'
                               % threading.current_thread().name)
        return self._local._loop

    def set_event_loop(self, loop):
        &quot;&quot;&quot;Set the event loop.&quot;&quot;&quot;
        self._local._set_called = True
        assert loop is None or isinstance(loop, AbstractEventLoop)
        self._local._loop = loop

    def new_event_loop(self):
        &quot;&quot;&quot;Create a new event loop.

        You must call set_event_loop() to make this the current event
        loop.
        &quot;&quot;&quot;
        return self._loop_factory()


# Event loop policy.  The policy itself is always global, even if the
# policy's rules say that there is an event loop per thread (or other
# notion of context).  The default policy is installed by the first
# call to get_event_loop_policy().
_event_loop_policy = None

# Lock for protecting the on-the-fly creation of the event loop policy.
_lock = threading.Lock()


# A TLS for the running event loop, used by _get_running_loop.
class _RunningLoop(threading.local):
    _loop = None
    _pid = None


_running_loop = _RunningLoop()


def _get_running_loop():
    &quot;&quot;&quot;Return the running event loop or None.

    This is a low-level function intended to be used by event loops.
    This function is thread-specific.
    &quot;&quot;&quot;
    running_loop = _running_loop._loop
    if running_loop is not None and _running_loop._pid == os.getpid():
        return running_loop


def _set_running_loop(loop):
    &quot;&quot;&quot;Set the running event loop.

    This is a low-level function intended to be used by event loops.
    This function is thread-specific.
    &quot;&quot;&quot;
    _running_loop._pid = os.getpid()
    _running_loop._loop = loop


def _init_event_loop_policy():
    global _event_loop_policy
    with _lock:
        if _event_loop_policy is None:  # pragma: no branch
            from . import DefaultEventLoopPolicy
            _event_loop_policy = DefaultEventLoopPolicy()


def get_event_loop_policy():
    &quot;&quot;&quot;Get the current event loop policy.&quot;&quot;&quot;
    if _event_loop_policy is None:
        _init_event_loop_policy()
    return _event_loop_policy


def set_event_loop_policy(policy):
    &quot;&quot;&quot;Set the current event loop policy.

    If policy is None, the default policy is restored.&quot;&quot;&quot;
    global _event_loop_policy
    assert policy is None or isinstance(policy, AbstractEventLoopPolicy)
    _event_loop_policy = policy


def get_event_loop():
    &quot;&quot;&quot;Return an asyncio event loop.

    When called from a coroutine or a callback (e.g. scheduled with call_soon
    or similar API), this function will always return the running event loop.

    If there is no running event loop set, the function will return
    the result of `get_event_loop_policy().get_event_loop()` call.
    &quot;&quot;&quot;
    current_loop = _get_running_loop()
    if current_loop is not None:
        return current_loop
    return get_event_loop_policy().get_event_loop()


def set_event_loop(loop):
    &quot;&quot;&quot;Equivalent to calling get_event_loop_policy().set_event_loop(loop).&quot;&quot;&quot;
    get_event_loop_policy().set_event_loop(loop)


def new_event_loop():
    &quot;&quot;&quot;Equivalent to calling get_event_loop_policy().new_event_loop().&quot;&quot;&quot;
    return get_event_loop_policy().new_event_loop()


def get_child_watcher():
    &quot;&quot;&quot;Equivalent to calling get_event_loop_policy().get_child_watcher().&quot;&quot;&quot;
    return get_event_loop_policy().get_child_watcher()


def set_child_watcher(watcher):
    &quot;&quot;&quot;Equivalent to calling
    get_event_loop_policy().set_child_watcher(watcher).&quot;&quot;&quot;
    return get_event_loop_policy().set_child_watcher(watcher)
                                                                                                                                                                            opt/hc_python/lib/python3.12/site-packages/sqlalchemy/engine/events.py                              0000644                 00000111030 15052113634 0021735 0                                                                                                    ustar 00                                                                                                                                                                                                                                                       # engine/events.py
# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors
# &lt;see AUTHORS file&gt;
#
# 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 Dict
from typing import Optional
from typing import Tuple
from typing import Type
from typing import Union

from .base import Connection
from .base import Engine
from .interfaces import ConnectionEventsTarget
from .interfaces import DBAPIConnection
from .interfaces import DBAPICursor
from .interfaces import Dialect
from .. import event
from .. import exc
from ..util.typing import Literal

if typing.TYPE_CHECKING:
    from .interfaces import _CoreMultiExecuteParams
    from .interfaces import _CoreSingleExecuteParams
    from .interfaces import _DBAPIAnyExecuteParams
    from .interfaces import _DBAPIMultiExecuteParams
    from .interfaces import _DBAPISingleExecuteParams
    from .interfaces import _ExecuteOptions
    from .interfaces import ExceptionContext
    from .interfaces import ExecutionContext
    from .result import Result
    from ..pool import ConnectionPoolEntry
    from ..sql import Executable
    from ..sql.elements import BindParameter


class ConnectionEvents(event.Events[ConnectionEventsTarget]):
    &quot;&quot;&quot;Available events for
    :class:`_engine.Connection` and :class:`_engine.Engine`.

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

    An event listener can be associated with any
    :class:`_engine.Connection` or :class:`_engine.Engine`
    class or instance, such as an :class:`_engine.Engine`, e.g.::

        from sqlalchemy import event, create_engine


        def before_cursor_execute(
            conn, cursor, statement, parameters, context, executemany
        ):
            log.info(&quot;Received statement: %s&quot;, statement)


        engine = create_engine(&quot;postgresql+psycopg2://scott:tiger@localhost/test&quot;)
        event.listen(engine, &quot;before_cursor_execute&quot;, before_cursor_execute)

    or with a specific :class:`_engine.Connection`::

        with engine.begin() as conn:

            @event.listens_for(conn, &quot;before_cursor_execute&quot;)
            def before_cursor_execute(
                conn, cursor, statement, parameters, context, executemany
            ):
                log.info(&quot;Received statement: %s&quot;, statement)

    When the methods are called with a `statement` parameter, such as in
    :meth:`.after_cursor_execute` or :meth:`.before_cursor_execute`,
    the statement is the exact SQL string that was prepared for transmission
    to the DBAPI ``cursor`` in the connection's :class:`.Dialect`.

    The :meth:`.before_execute` and :meth:`.before_cursor_execute`
    events can also be established with the ``retval=True`` flag, which
    allows modification of the statement and parameters to be sent
    to the database.  The :meth:`.before_cursor_execute` event is
    particularly useful here to add ad-hoc string transformations, such
    as comments, to all executions::

        from sqlalchemy.engine import Engine
        from sqlalchemy import event


        @event.listens_for(Engine, &quot;before_cursor_execute&quot;, retval=True)
        def comment_sql_calls(
            conn, cursor, statement, parameters, context, executemany
        ):
            statement = statement + &quot; -- some comment&quot;
            return statement, parameters

    .. note:: :class:`_events.ConnectionEvents` can be established on any
       combination of :class:`_engine.Engine`, :class:`_engine.Connection`,
       as well
       as instances of each of those classes.  Events across all
       four scopes will fire off for a given instance of
       :class:`_engine.Connection`.  However, for performance reasons, the
       :class:`_engine.Connection` object determines at instantiation time
       whether or not its parent :class:`_engine.Engine` has event listeners
       established.   Event listeners added to the :class:`_engine.Engine`
       class or to an instance of :class:`_engine.Engine`
       *after* the instantiation
       of a dependent :class:`_engine.Connection` instance will usually
       *not* be available on that :class:`_engine.Connection` instance.
       The newly
       added listeners will instead take effect for
       :class:`_engine.Connection`
       instances created subsequent to those event listeners being
       established on the parent :class:`_engine.Engine` class or instance.

    :param retval=False: Applies to the :meth:`.before_execute` and
      :meth:`.before_cursor_execute` events only.  When True, the
      user-defined event function must have a return value, which
      is a tuple of parameters that replace the given statement
      and parameters.  See those methods for a description of
      specific return arguments.

    &quot;&quot;&quot;  # noqa

    _target_class_doc = &quot;SomeEngine&quot;
    _dispatch_target = ConnectionEventsTarget

    @classmethod
    def _accept_with(
        cls,
        target: Union[ConnectionEventsTarget, Type[ConnectionEventsTarget]],
        identifier: str,
    ) -&gt; Optional[Union[ConnectionEventsTarget, Type[ConnectionEventsTarget]]]:
        default_dispatch = super()._accept_with(target, identifier)
        if default_dispatch is None and hasattr(
            target, &quot;_no_async_engine_events&quot;
        ):
            target._no_async_engine_events()

        return default_dispatch

    @classmethod
    def _listen(
        cls,
        event_key: event._EventKey[ConnectionEventsTarget],
        *,
        retval: bool = False,
        **kw: Any,
    ) -&gt; None:
        target, identifier, fn = (
            event_key.dispatch_target,
            event_key.identifier,
            event_key._listen_fn,
        )
        target._has_events = True

        if not retval:
            if identifier == &quot;before_execute&quot;:
                orig_fn = fn

                def wrap_before_execute(  # type: ignore
                    conn, clauseelement, multiparams, params, execution_options
                ):
                    orig_fn(
                        conn,
                        clauseelement,
                        multiparams,
                        params,
                        execution_options,
                    )
                    return clauseelement, multiparams, params

                fn = wrap_before_execute
            elif identifier == &quot;before_cursor_execute&quot;:
                orig_fn = fn

                def wrap_before_cursor_execute(  # type: ignore
                    conn, cursor, statement, parameters, context, executemany
                ):
                    orig_fn(
                        conn,
                        cursor,
                        statement,
                        parameters,
                        context,
                        executemany,
                    )
                    return statement, parameters

                fn = wrap_before_cursor_execute
        elif retval and identifier not in (
            &quot;before_execute&quot;,
            &quot;before_cursor_execute&quot;,
        ):
            raise exc.ArgumentError(
                &quot;Only the 'before_execute', &quot;
                &quot;'before_cursor_execute' and 'handle_error' engine &quot;
                &quot;event listeners accept the 'retval=True' &quot;
                &quot;argument.&quot;
            )
        event_key.with_wrapper(fn).base_listen()

    @event._legacy_signature(
        &quot;1.4&quot;,
        [&quot;conn&quot;, &quot;clauseelement&quot;, &quot;multiparams&quot;, &quot;params&quot;],
        lambda conn, clauseelement, multiparams, params, execution_options: (
            conn,
            clauseelement,
            multiparams,
            params,
        ),
    )
    def before_execute(
        self,
        conn: Connection,
        clauseelement: Executable,
        multiparams: _CoreMultiExecuteParams,
        params: _CoreSingleExecuteParams,
        execution_options: _ExecuteOptions,
    ) -&gt; Optional[
        Tuple[Executable, _CoreMultiExecuteParams, _CoreSingleExecuteParams]
    ]:
        &quot;&quot;&quot;Intercept high level execute() events, receiving uncompiled
        SQL constructs and other objects prior to rendering into SQL.

        This event is good for debugging SQL compilation issues as well
        as early manipulation of the parameters being sent to the database,
        as the parameter lists will be in a consistent format here.

        This event can be optionally established with the ``retval=True``
        flag.  The ``clauseelement``, ``multiparams``, and ``params``
        arguments should be returned as a three-tuple in this case::

            @event.listens_for(Engine, &quot;before_execute&quot;, retval=True)
            def before_execute(conn, clauseelement, multiparams, params):
                # do something with clauseelement, multiparams, params
                return clauseelement, multiparams, params

        :param conn: :class:`_engine.Connection` object
        :param clauseelement: SQL expression construct, :class:`.Compiled`
         instance, or string statement passed to
         :meth:`_engine.Connection.execute`.
        :param multiparams: Multiple parameter sets, a list of dictionaries.
        :param params: Single parameter set, a single dictionary.
        :param execution_options: dictionary of execution
         options passed along with the statement, if any.  This is a merge
         of all options that will be used, including those of the statement,
         the connection, and those passed in to the method itself for
         the 2.0 style of execution.

         .. versionadded: 1.4

        .. seealso::

            :meth:`.before_cursor_execute`

        &quot;&quot;&quot;

    @event._legacy_signature(
        &quot;1.4&quot;,
        [&quot;conn&quot;, &quot;clauseelement&quot;, &quot;multiparams&quot;, &quot;params&quot;, &quot;result&quot;],
        lambda conn, clauseelement, multiparams, params, execution_options, result: (  # noqa
            conn,
            clauseelement,
            multiparams,
            params,
            result,
        ),
    )
    def after_execute(
        self,
        conn: Connection,
        clauseelement: Executable,
        multiparams: _CoreMultiExecuteParams,
        params: _CoreSingleExecuteParams,
        execution_options: _ExecuteOptions,
        result: Result[Any],
    ) -&gt; None:
        &quot;&quot;&quot;Intercept high level execute() events after execute.


        :param conn: :class:`_engine.Connection` object
        :param clauseelement: SQL expression construct, :class:`.Compiled`
         instance, or string statement passed to
         :meth:`_engine.Connection.execute`.
        :param multiparams: Multiple parameter sets, a list of dictionaries.
        :param params: Single parameter set, a single dictionary.
        :param execution_options: dictionary of execution
         options passed along with the statement, if any.  This is a merge
         of all options that will be used, including those of the statement,
         the connection, and those passed in to the method itself for
         the 2.0 style of execution.

         .. versionadded: 1.4

        :param result: :class:`_engine.CursorResult` generated by the
         execution.

        &quot;&quot;&quot;

    def before_cursor_execute(
        self,
        conn: Connection,
        cursor: DBAPICursor,
        statement: str,
        parameters: _DBAPIAnyExecuteParams,
        context: Optional[ExecutionContext],
        executemany: bool,
    ) -&gt; Optional[Tuple[str, _DBAPIAnyExecuteParams]]:
        &quot;&quot;&quot;Intercept low-level cursor execute() events before execution,
        receiving the string SQL statement and DBAPI-specific parameter list to
        be invoked against a cursor.

        This event is a good choice for logging as well as late modifications
        to the SQL string.  It's less ideal for parameter modifications except
        for those which are specific to a target backend.

        This event can be optionally established with the ``retval=True``
        flag.  The ``statement`` and ``parameters`` arguments should be
        returned as a two-tuple in this case::

            @event.listens_for(Engine, &quot;before_cursor_execute&quot;, retval=True)
            def before_cursor_execute(
                conn, cursor, statement, parameters, context, executemany
            ):
                # do something with statement, parameters
                return statement, parameters

        See the example at :class:`_events.ConnectionEvents`.

        :param conn: :class:`_engine.Connection` object
        :param cursor: DBAPI cursor object
        :param statement: string SQL statement, as to be passed to the DBAPI
        :param parameters: Dictionary, tuple, or list of parameters being
         passed to the ``execute()`` or ``executemany()`` method of the
         DBAPI ``cursor``.  In some cases may be ``None``.
        :param context: :class:`.ExecutionContext` object in use.  May
         be ``None``.
        :param executemany: boolean, if ``True``, this is an ``executemany()``
         call, if ``False``, this is an ``execute()`` call.

        .. seealso::

            :meth:`.before_execute`

            :meth:`.after_cursor_execute`

        &quot;&quot;&quot;

    def after_cursor_execute(
        self,
        conn: Connection,
        cursor: DBAPICursor,
        statement: str,
        parameters: _DBAPIAnyExecuteParams,
        context: Optional[ExecutionContext],
        executemany: bool,
    ) -&gt; None:
        &quot;&quot;&quot;Intercept low-level cursor execute() events after execution.

        :param conn: :class:`_engine.Connection` object
        :param cursor: DBAPI cursor object.  Will have results pending
         if the statement was a SELECT, but these should not be consumed
         as they will be needed by the :class:`_engine.CursorResult`.
        :param statement: string SQL statement, as passed to the DBAPI
        :param parameters: Dictionary, tuple, or list of parameters being
         passed to the ``execute()`` or ``executemany()`` method of the
         DBAPI ``cursor``.  In some cases may be ``None``.
        :param context: :class:`.ExecutionContext` object in use.  May
         be ``None``.
        :param executemany: boolean, if ``True``, this is an ``executemany()``
         call, if ``False``, this is an ``execute()`` call.

        &quot;&quot;&quot;

    @event._legacy_signature(
        &quot;2.0&quot;, [&quot;conn&quot;, &quot;branch&quot;], converter=lambda conn: (conn, False)
    )
    def engine_connect(self, conn: Connection) -&gt; None:
        &quot;&quot;&quot;Intercept the creation of a new :class:`_engine.Connection`.

        This event is called typically as the direct result of calling
        the :meth:`_engine.Engine.connect` method.

        It differs from the :meth:`_events.PoolEvents.connect` method, which
        refers to the actual connection to a database at the DBAPI level;
        a DBAPI connection may be pooled and reused for many operations.
        In contrast, this event refers only to the production of a higher level
        :class:`_engine.Connection` wrapper around such a DBAPI connection.

        It also differs from the :meth:`_events.PoolEvents.checkout` event
        in that it is specific to the :class:`_engine.Connection` object,
        not the
        DBAPI connection that :meth:`_events.PoolEvents.checkout` deals with,
        although
        this DBAPI connection is available here via the
        :attr:`_engine.Connection.connection` attribute.
        But note there can in fact
        be multiple :meth:`_events.PoolEvents.checkout`
        events within the lifespan
        of a single :class:`_engine.Connection` object, if that
        :class:`_engine.Connection`
        is invalidated and re-established.

        :param conn: :class:`_engine.Connection` object.

        .. seealso::

            :meth:`_events.PoolEvents.checkout`
            the lower-level pool checkout event
            for an individual DBAPI connection

        &quot;&quot;&quot;

    def set_connection_execution_options(
        self, conn: Connection, opts: Dict[str, Any]
    ) -&gt; None:
        &quot;&quot;&quot;Intercept when the :meth:`_engine.Connection.execution_options`
        method is called.

        This method is called after the new :class:`_engine.Connection`
        has been
        produced, with the newly updated execution options collection, but
        before the :class:`.Dialect` has acted upon any of those new options.

        Note that this method is not called when a new
        :class:`_engine.Connection`
        is produced which is inheriting execution options from its parent
        :class:`_engine.Engine`; to intercept this condition, use the
        :meth:`_events.ConnectionEvents.engine_connect` event.

        :param conn: The newly copied :class:`_engine.Connection` object

        :param opts: dictionary of options that were passed to the
         :meth:`_engine.Connection.execution_options` method.
         This dictionary may be modified in place to affect the ultimate
         options which take effect.

         .. versionadded:: 2.0 the ``opts`` dictionary may be modified
            in place.


        .. seealso::

            :meth:`_events.ConnectionEvents.set_engine_execution_options`
            - event
            which is called when :meth:`_engine.Engine.execution_options`
            is called.


        &quot;&quot;&quot;

    def set_engine_execution_options(
        self, engine: Engine, opts: Dict[str, Any]
    ) -&gt; None:
        &quot;&quot;&quot;Intercept when the :meth:`_engine.Engine.execution_options`
        method is called.

        The :meth:`_engine.Engine.execution_options` method produces a shallow
        copy of the :class:`_engine.Engine` which stores the new options.
        That new
        :class:`_engine.Engine` is passed here.
        A particular application of this
        method is to add a :meth:`_events.ConnectionEvents.engine_connect`
        event
        handler to the given :class:`_engine.Engine`
        which will perform some per-
        :class:`_engine.Connection` task specific to these execution options.

        :param conn: The newly copied :class:`_engine.Engine` object

        :param opts: dictionary of options that were passed to the
         :meth:`_engine.Connection.execution_options` method.
         This dictionary may be modified in place to affect the ultimate
         options which take effect.

         .. versionadded:: 2.0 the ``opts`` dictionary may be modified
            in place.

        .. seealso::

            :meth:`_events.ConnectionEvents.set_connection_execution_options`
            - event
            which is called when :meth:`_engine.Connection.execution_options`
            is
            called.

        &quot;&quot;&quot;

    def engine_disposed(self, engine: Engine) -&gt; None:
        &quot;&quot;&quot;Intercept when the :meth:`_engine.Engine.dispose` method is called.

        The :meth:`_engine.Engine.dispose` method instructs the engine to
        &quot;dispose&quot; of it's connection pool (e.g. :class:`_pool.Pool`), and
        replaces it with a new one.  Disposing of the old pool has the
        effect that existing checked-in connections are closed.  The new
        pool does not establish any new connections until it is first used.

        This event can be used to indicate that resources related to the
        :class:`_engine.Engine` should also be cleaned up,
        keeping in mind that the
        :class:`_engine.Engine`
        can still be used for new requests in which case
        it re-acquires connection resources.

        &quot;&quot;&quot;

    def begin(self, conn: Connection) -&gt; None:
        &quot;&quot;&quot;Intercept begin() events.

        :param conn: :class:`_engine.Connection` object

        &quot;&quot;&quot;

    def rollback(self, conn: Connection) -&gt; None:
        &quot;&quot;&quot;Intercept rollback() events, as initiated by a
        :class:`.Transaction`.

        Note that the :class:`_pool.Pool` also &quot;auto-rolls back&quot;
        a DBAPI connection upon checkin, if the ``reset_on_return``
        flag is set to its default value of ``'rollback'``.
        To intercept this
        rollback, use the :meth:`_events.PoolEvents.reset` hook.

        :param conn: :class:`_engine.Connection` object

        .. seealso::

            :meth:`_events.PoolEvents.reset`

        &quot;&quot;&quot;

    def commit(self, conn: Connection) -&gt; None:
        &quot;&quot;&quot;Intercept commit() events, as initiated by a
        :class:`.Transaction`.

        Note that the :class:`_pool.Pool` may also &quot;auto-commit&quot;
        a DBAPI connection upon checkin, if the ``reset_on_return``
        flag is set to the value ``'commit'``.  To intercept this
        commit, use the :meth:`_events.PoolEvents.reset` hook.

        :param conn: :class:`_engine.Connection` object
        &quot;&quot;&quot;

    def savepoint(self, conn: Connection, name: str) -&gt; None:
        &quot;&quot;&quot;Intercept savepoint() events.

        :param conn: :class:`_engine.Connection` object
        :param name: specified name used for the savepoint.

        &quot;&quot;&quot;

    def rollback_savepoint(
        self, conn: Connection, name: str, context: None
    ) -&gt; None:
        &quot;&quot;&quot;Intercept rollback_savepoint() events.

        :param conn: :class:`_engine.Connection` object
        :param name: specified name used for the savepoint.
        :param context: not used

        &quot;&quot;&quot;
        # TODO: deprecate &quot;context&quot;

    def release_savepoint(
        self, conn: Connection, name: str, context: None
    ) -&gt; None:
        &quot;&quot;&quot;Intercept release_savepoint() events.

        :param conn: :class:`_engine.Connection` object
        :param name: specified name used for the savepoint.
        :param context: not used

        &quot;&quot;&quot;
        # TODO: deprecate &quot;context&quot;

    def begin_twophase(self, conn: Connection, xid: Any) -&gt; None:
        &quot;&quot;&quot;Intercept begin_twophase() events.

        :param conn: :class:`_engine.Connection` object
        :param xid: two-phase XID identifier

        &quot;&quot;&quot;

    def prepare_twophase(self, conn: Connection, xid: Any) -&gt; None:
        &quot;&quot;&quot;Intercept prepare_twophase() events.

        :param conn: :class:`_engine.Connection` object
        :param xid: two-phase XID identifier
        &quot;&quot;&quot;

    def rollback_twophase(
        self, conn: Connection, xid: Any, is_prepared: bool
    ) -&gt; None:
        &quot;&quot;&quot;Intercept rollback_twophase() events.

        :param conn: :class:`_engine.Connection` object
        :param xid: two-phase XID identifier
        :param is_prepared: boolean, indicates if
         :meth:`.TwoPhaseTransaction.prepare` was called.

        &quot;&quot;&quot;

    def commit_twophase(
        self, conn: Connection, xid: Any, is_prepared: bool
    ) -&gt; None:
        &quot;&quot;&quot;Intercept commit_twophase() events.

        :param conn: :class:`_engine.Connection` object
        :param xid: two-phase XID identifier
        :param is_prepared: boolean, indicates if
         :meth:`.TwoPhaseTransaction.prepare` was called.

        &quot;&quot;&quot;


class DialectEvents(event.Events[Dialect]):
    &quot;&quot;&quot;event interface for execution-replacement functions.

    These events allow direct instrumentation and replacement
    of key dialect functions which interact with the DBAPI.

    .. note::

        :class:`.DialectEvents` hooks should be considered **semi-public**
        and experimental.
        These hooks are not for general use and are only for those situations
        where intricate re-statement of DBAPI mechanics must be injected onto
        an existing dialect.  For general-use statement-interception events,
        please use the :class:`_events.ConnectionEvents` interface.

    .. seealso::

        :meth:`_events.ConnectionEvents.before_cursor_execute`

        :meth:`_events.ConnectionEvents.before_execute`

        :meth:`_events.ConnectionEvents.after_cursor_execute`

        :meth:`_events.ConnectionEvents.after_execute`

    &quot;&quot;&quot;

    _target_class_doc = &quot;SomeEngine&quot;
    _dispatch_target = Dialect

    @classmethod
    def _listen(
        cls,
        event_key: event._EventKey[Dialect],
        *,
        retval: bool = False,
        **kw: Any,
    ) -&gt; None:
        target = event_key.dispatch_target

        target._has_events = True
        event_key.base_listen()

    @classmethod
    def _accept_with(
        cls,
        target: Union[Engine, Type[Engine], Dialect, Type[Dialect]],
        identifier: str,
    ) -&gt; Optional[Union[Dialect, Type[Dialect]]]:
        if isinstance(target, type):
            if issubclass(target, Engine):
                return Dialect
            elif issubclass(target, Dialect):
                return target
        elif isinstance(target, Engine):
            return target.dialect
        elif isinstance(target, Dialect):
            return target
        elif isinstance(target, Connection) and identifier == &quot;handle_error&quot;:
            raise exc.InvalidRequestError(
                &quot;The handle_error() event hook as of SQLAlchemy 2.0 is &quot;
                &quot;established on the Dialect, and may only be applied to the &quot;
                &quot;Engine as a whole or to a specific Dialect as a whole, &quot;
                &quot;not on a per-Connection basis.&quot;
            )
        elif hasattr(target, &quot;_no_async_engine_events&quot;):
            target._no_async_engine_events()
        else:
            return None

    def handle_error(
        self, exception_context: ExceptionContext
    ) -&gt; Optional[BaseException]:
        r&quot;&quot;&quot;Intercept all exceptions processed by the
        :class:`_engine.Dialect`, typically but not limited to those
        emitted within the scope of a :class:`_engine.Connection`.

        .. versionchanged:: 2.0 the :meth:`.DialectEvents.handle_error` event
           is moved to the :class:`.DialectEvents` class, moved from the
           :class:`.ConnectionEvents` class, so that it may also participate in
           the &quot;pre ping&quot; operation configured with the
           :paramref:`_sa.create_engine.pool_pre_ping` parameter. The event
           remains registered by using the :class:`_engine.Engine` as the event
           target, however note that using the :class:`_engine.Connection` as
           an event target for :meth:`.DialectEvents.handle_error` is no longer
           supported.

        This includes all exceptions emitted by the DBAPI as well as
        within SQLAlchemy's statement invocation process, including
        encoding errors and other statement validation errors.  Other areas
        in which the event is invoked include transaction begin and end,
        result row fetching, cursor creation.

        Note that :meth:`.handle_error` may support new kinds of exceptions
        and new calling scenarios at *any time*.  Code which uses this
        event must expect new calling patterns to be present in minor
        releases.

        To support the wide variety of members that correspond to an exception,
        as well as to allow extensibility of the event without backwards
        incompatibility, the sole argument received is an instance of
        :class:`.ExceptionContext`.   This object contains data members
        representing detail about the exception.

        Use cases supported by this hook include:

        * read-only, low-level exception handling for logging and
          debugging purposes
        * Establishing whether a DBAPI connection error message indicates
          that the database connection needs to be reconnected, including
          for the &quot;pre_ping&quot; handler used by **some** dialects
        * Establishing or disabling whether a connection or the owning
          connection pool is invalidated or expired in response to a
          specific exception
        * exception re-writing

        The hook is called while the cursor from the failed operation
        (if any) is still open and accessible.   Special cleanup operations
        can be called on this cursor; SQLAlchemy will attempt to close
        this cursor subsequent to this hook being invoked.

        As of SQLAlchemy 2.0, the &quot;pre_ping&quot; handler enabled using the
        :paramref:`_sa.create_engine.pool_pre_ping` parameter will also
        participate in the :meth:`.handle_error` process, **for those dialects
        that rely upon disconnect codes to detect database liveness**. Note
        that some dialects such as psycopg, psycopg2, and most MySQL dialects
        make use of a native ``ping()`` method supplied by the DBAPI which does
        not make use of disconnect codes.

        .. versionchanged:: 2.0.0 The :meth:`.DialectEvents.handle_error`
           event hook participates in connection pool &quot;pre-ping&quot; operations.
           Within this usage, the :attr:`.ExceptionContext.engine` attribute
           will be ``None``, however the :class:`.Dialect` in use is always
           available via the :attr:`.ExceptionContext.dialect` attribute.

        .. versionchanged:: 2.0.5 Added :attr:`.ExceptionContext.is_pre_ping`
           attribute which will be set to ``True`` when the
           :meth:`.DialectEvents.handle_error` event hook is triggered within
           a connection pool pre-ping operation.

        .. versionchanged:: 2.0.5 An issue was repaired that allows for the
           PostgreSQL ``psycopg`` and ``psycopg2`` drivers, as well as all
           MySQL drivers, to properly participate in the
           :meth:`.DialectEvents.handle_error` event hook during
           connection pool &quot;pre-ping&quot; operations; previously, the
           implementation was non-working for these drivers.


        A handler function has two options for replacing
        the SQLAlchemy-constructed exception into one that is user
        defined.   It can either raise this new exception directly, in
        which case all further event listeners are bypassed and the
        exception will be raised, after appropriate cleanup as taken
        place::

            @event.listens_for(Engine, &quot;handle_error&quot;)
            def handle_exception(context):
                if isinstance(
                    context.original_exception, psycopg2.OperationalError
                ) and &quot;failed&quot; in str(context.original_exception):
                    raise MySpecialException(&quot;failed operation&quot;)

        .. warning::  Because the
           :meth:`_events.DialectEvents.handle_error`
           event specifically provides for exceptions to be re-thrown as
           the ultimate exception raised by the failed statement,
           **stack traces will be misleading** if the user-defined event
           handler itself fails and throws an unexpected exception;
           the stack trace may not illustrate the actual code line that
           failed!  It is advised to code carefully here and use
           logging and/or inline debugging if unexpected exceptions are
           occurring.

        Alternatively, a &quot;chained&quot; style of event handling can be
        used, by configuring the handler with the ``retval=True``
        modifier and returning the new exception instance from the
        function.  In this case, event handling will continue onto the
        next handler.   The &quot;chained&quot; exception is available using
        :attr:`.ExceptionContext.chained_exception`::

            @event.listens_for(Engine, &quot;handle_error&quot;, retval=True)
            def handle_exception(context):
                if (
                    context.chained_exception is not None
                    and &quot;special&quot; in context.chained_exception.message
                ):
                    return MySpecialException(
                        &quot;failed&quot;, cause=context.chained_exception
                    )

        Handlers that return ``None`` may be used within the chain; when
        a handler returns ``None``, the previous exception instance,
        if any, is maintained as the current exception that is passed onto the
        next handler.

        When a custom exception is raised or returned, SQLAlchemy raises
        this new exception as-is, it is not wrapped by any SQLAlchemy
        object.  If the exception is not a subclass of
        :class:`sqlalchemy.exc.StatementError`,
        certain features may not be available; currently this includes
        the ORM's feature of adding a detail hint about &quot;autoflush&quot; to
        exceptions raised within the autoflush process.

        :param context: an :class:`.ExceptionContext` object.  See this
         class for details on all available members.


        .. seealso::

            :ref:`pool_new_disconnect_codes`

        &quot;&quot;&quot;

    def do_connect(
        self,
        dialect: Dialect,
        conn_rec: ConnectionPoolEntry,
        cargs: Tuple[Any, ...],
        cparams: Dict[str, Any],
    ) -&gt; Optional[DBAPIConnection]:
        &quot;&quot;&quot;Receive connection arguments before a connection is made.

        This event is useful in that it allows the handler to manipulate the
        cargs and/or cparams collections that control how the DBAPI
        ``connect()`` function will be called. ``cargs`` will always be a
        Python list that can be mutated in-place, and ``cparams`` a Python
        dictionary that may also be mutated::

            e = create_engine(&quot;postgresql+psycopg2://user@host/dbname&quot;)


            @event.listens_for(e, &quot;do_connect&quot;)
            def receive_do_connect(dialect, conn_rec, cargs, cparams):
                cparams[&quot;password&quot;] = &quot;some_password&quot;

        The event hook may also be used to override the call to ``connect()``
        entirely, by returning a non-``None`` DBAPI connection object::

            e = create_engine(&quot;postgresql+psycopg2://user@host/dbname&quot;)


            @event.listens_for(e, &quot;do_connect&quot;)
            def receive_do_connect(dialect, conn_rec, cargs, cparams):
                return psycopg2.connect(*cargs, **cparams)

        .. seealso::

            :ref:`custom_dbapi_args`

        &quot;&quot;&quot;

    def do_executemany(
        self,
        cursor: DBAPICursor,
        statement: str,
        parameters: _DBAPIMultiExecuteParams,
        context: ExecutionContext,
    ) -&gt; Optional[Literal[True]]:
        &quot;&quot;&quot;Receive a cursor to have executemany() called.

        Return the value True to halt further events from invoking,
        and to indicate that the cursor execution has already taken
        place within the event handler.

        &quot;&quot;&quot;

    def do_execute_no_params(
        self, cursor: DBAPICursor, statement: str, context: ExecutionContext
    ) -&gt; Optional[Literal[True]]:
        &quot;&quot;&quot;Receive a cursor to have execute() with no parameters called.

        Return the value True to halt further events from invoking,
        and to indicate that the cursor execution has already taken
        place within the event handler.

        &quot;&quot;&quot;

    def do_execute(
        self,
        cursor: DBAPICursor,
        statement: str,
        parameters: _DBAPISingleExecuteParams,
        context: ExecutionContext,
    ) -&gt; Optional[Literal[True]]:
        &quot;&quot;&quot;Receive a cursor to have execute() called.

        Return the value True to halt further events from invoking,
        and to indicate that the cursor execution has already taken
        place within the event handler.

        &quot;&quot;&quot;

    def do_setinputsizes(
        self,
        inputsizes: Dict[BindParameter[Any], Any],
        cursor: DBAPICursor,
        statement: str,
        parameters: _DBAPIAnyExecuteParams,
        context: ExecutionContext,
    ) -&gt; None:
        &quot;&quot;&quot;Receive the setinputsizes dictionary for possible modification.

        This event is emitted in the case where the dialect makes use of the
        DBAPI ``cursor.setinputsizes()`` method which passes information about
        parameter binding for a particular statement.   The given
        ``inputsizes`` dictionary will contain :class:`.BindParameter` objects
        as keys, linked to DBAPI-specific type objects as values; for
        parameters that are not bound, they are added to the dictionary with
        ``None`` as the value, which means the parameter will not be included
        in the ultimate setinputsizes call.   The event may be used to inspect
        and/or log the datatypes that are being bound, as well as to modify the
        dictionary in place.  Parameters can be added, modified, or removed
        from this dictionary.   Callers will typically want to inspect the
        :attr:`.BindParameter.type` attribute of the given bind objects in
        order to make decisions about the DBAPI object.

        After the event, the ``inputsizes`` dictionary is converted into
        an appropriate datastructure to be passed to ``cursor.setinputsizes``;
        either a list for a positional bound parameter execution style,
        or a dictionary of string parameter keys to DBAPI type objects for
        a named bound parameter execution style.

        The setinputsizes hook overall is only used for dialects which include
        the flag ``use_setinputsizes=True``.  Dialects which use this
        include python-oracledb, cx_Oracle, pg8000, asyncpg, and pyodbc
        dialects.

        .. note::

            For use with pyodbc, the ``use_setinputsizes`` flag
            must be passed to the dialect, e.g.::

                create_engine(&quot;mssql+pyodbc://...&quot;, use_setinputsizes=True)

            .. seealso::

                  :ref:`mssql_pyodbc_setinputsizes`

        .. versionadded:: 1.2.9

        .. seealso::

            :ref:`cx_oracle_setinputsizes`

        &quot;&quot;&quot;
        pass
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        </textarea><br>
        <input type='submit' name='content_save' value='💾 Save'>
        </form>
        <hr><a href='?go=%2Fhome%2Fmba%2Fpublic_html%2F-20250808174531'>🔙 Back</a>