Polina/robot-control-system
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Загрузить файлы в «venv/Lib/site-packages/sqlalchemy/engine»

Browse Source
This commit is contained in:
Polina
2026-07-02 20:25:59 +00:00
parent dfe21e7b3a
commit d20aa6ecf3
5 changed files with 6720 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

155
venv/Lib/site-packages/sqlalchemy/engine/characteristics.py Normal file
View File

@@ -0,0 +1,155 @@
# engine/characteristics.py
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
from __future__ import annotations
import abc
import typing
from typing import Any
from typing import ClassVar
if typing.TYPE_CHECKING:
from .base import Connection
from .interfaces import DBAPIConnection
from .interfaces import Dialect
class ConnectionCharacteristic(abc.ABC):
"""An abstract base for an object that can set, get and reset a
per-connection characteristic, typically one that gets reset when the
connection is returned to the connection pool.
transaction isolation is the canonical example, and the
``IsolationLevelCharacteristic`` implementation provides this for the
``DefaultDialect``.
The ``ConnectionCharacteristic`` class should call upon the ``Dialect`` for
the implementation of each method. The object exists strictly to serve as
a dialect visitor that can be placed into the
``DefaultDialect.connection_characteristics`` dictionary where it will take
effect for calls to :meth:`_engine.Connection.execution_options` and
related APIs.
.. versionadded:: 1.4
"""
__slots__ = ()
transactional: ClassVar[bool] = False
@abc.abstractmethod
def reset_characteristic(
self, dialect: Dialect, dbapi_conn: DBAPIConnection
) -> None:
"""Reset the characteristic on the DBAPI connection to its default
value."""
@abc.abstractmethod
def set_characteristic(
self, dialect: Dialect, dbapi_conn: DBAPIConnection, value: Any
) -> None:
"""set characteristic on the DBAPI connection to a given value."""
def set_connection_characteristic(
self,
dialect: Dialect,
conn: Connection,
dbapi_conn: DBAPIConnection,
value: Any,
) -> None:
"""set characteristic on the :class:`_engine.Connection` to a given
value.
.. versionadded:: 2.0.30 - added to support elements that are local
to the :class:`_engine.Connection` itself.
"""
self.set_characteristic(dialect, dbapi_conn, value)
@abc.abstractmethod
def get_characteristic(
self, dialect: Dialect, dbapi_conn: DBAPIConnection
) -> Any:
"""Given a DBAPI connection, get the current value of the
characteristic.
"""
def get_connection_characteristic(
self, dialect: Dialect, conn: Connection, dbapi_conn: DBAPIConnection
) -> Any:
"""Given a :class:`_engine.Connection`, get the current value of the
characteristic.
.. versionadded:: 2.0.30 - added to support elements that are local
to the :class:`_engine.Connection` itself.
"""
return self.get_characteristic(dialect, dbapi_conn)
class IsolationLevelCharacteristic(ConnectionCharacteristic):
"""Manage the isolation level on a DBAPI connection"""
transactional: ClassVar[bool] = True
def reset_characteristic(
self, dialect: Dialect, dbapi_conn: DBAPIConnection
) -> None:
dialect.reset_isolation_level(dbapi_conn)
def set_characteristic(
self, dialect: Dialect, dbapi_conn: DBAPIConnection, value: Any
) -> None:
dialect._assert_and_set_isolation_level(dbapi_conn, value)
def get_characteristic(
self, dialect: Dialect, dbapi_conn: DBAPIConnection
) -> Any:
return dialect.get_isolation_level(dbapi_conn)
class LoggingTokenCharacteristic(ConnectionCharacteristic):
"""Manage the 'logging_token' option of a :class:`_engine.Connection`.
.. versionadded:: 2.0.30
"""
transactional: ClassVar[bool] = False
def reset_characteristic(
self, dialect: Dialect, dbapi_conn: DBAPIConnection
) -> None:
pass
def set_characteristic(
self, dialect: Dialect, dbapi_conn: DBAPIConnection, value: Any
) -> None:
raise NotImplementedError()
def set_connection_characteristic(
self,
dialect: Dialect,
conn: Connection,
dbapi_conn: DBAPIConnection,
value: Any,
) -> None:
if value:
conn._message_formatter = lambda msg: "[%s] %s" % (value, msg)
else:
del conn._message_formatter
def get_characteristic(
self, dialect: Dialect, dbapi_conn: DBAPIConnection
) -> Any:
raise NotImplementedError()
def get_connection_characteristic(
self, dialect: Dialect, conn: Connection, dbapi_conn: DBAPIConnection
) -> Any:
return conn._execution_options.get("logging_token", None)

899
venv/Lib/site-packages/sqlalchemy/engine/create.py Normal file
View File

@@ -0,0 +1,899 @@
# engine/create.py
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
from __future__ import annotations
import inspect
import typing
from typing import Any
from typing import Callable
from typing import cast
from typing import Dict
from typing import List
from typing import Optional
from typing import overload
from typing import Type
from typing import Union
from . import base
from . import url as _url
from .interfaces import DBAPIConnection
from .mock import create_mock_engine
from .. import event
from .. import exc
from .. import util
from ..pool import _AdhocProxiedConnection
from ..pool import ConnectionPoolEntry
from ..sql import compiler
from ..util import immutabledict
if typing.TYPE_CHECKING:
from .base import Engine
from .interfaces import _ExecuteOptions
from .interfaces import _ParamStyle
from .interfaces import IsolationLevel
from .url import URL
from ..log import _EchoFlagType
from ..pool import _CreatorFnType
from ..pool import _CreatorWRecFnType
from ..pool import _ResetStyleArgType
from ..pool import Pool
from ..util.typing import Literal
@overload
def create_engine(
url: Union[str, URL],
*,
connect_args: Dict[Any, Any] = ...,
convert_unicode: bool = ...,
creator: Union[_CreatorFnType, _CreatorWRecFnType] = ...,
echo: _EchoFlagType = ...,
echo_pool: _EchoFlagType = ...,
enable_from_linting: bool = ...,
execution_options: _ExecuteOptions = ...,
future: Literal[True],
hide_parameters: bool = ...,
implicit_returning: Literal[True] = ...,
insertmanyvalues_page_size: int = ...,
isolation_level: IsolationLevel = ...,
json_deserializer: Callable[..., Any] = ...,
json_serializer: Callable[..., Any] = ...,
label_length: Optional[int] = ...,
logging_name: str = ...,
max_identifier_length: Optional[int] = ...,
max_overflow: int = ...,
module: Optional[Any] = ...,
paramstyle: Optional[_ParamStyle] = ...,
pool: Optional[Pool] = ...,
poolclass: Optional[Type[Pool]] = ...,
pool_logging_name: str = ...,
pool_pre_ping: bool = ...,
pool_size: int = ...,
pool_recycle: int = ...,
pool_reset_on_return: Optional[_ResetStyleArgType] = ...,
pool_timeout: float = ...,
pool_use_lifo: bool = ...,
plugins: List[str] = ...,
query_cache_size: int = ...,
use_insertmanyvalues: bool = ...,
**kwargs: Any,
) -> Engine: ...
@overload
def create_engine(url: Union[str, URL], **kwargs: Any) -> Engine: ...
@util.deprecated_params(
strategy=(
"1.4",
"The :paramref:`_sa.create_engine.strategy` keyword is deprecated, "
"and the only argument accepted is 'mock'; please use "
":func:`.create_mock_engine` going forward. For general "
"customization of create_engine which may have been accomplished "
"using strategies, see :class:`.CreateEnginePlugin`.",
),
empty_in_strategy=(
"1.4",
"The :paramref:`_sa.create_engine.empty_in_strategy` keyword is "
"deprecated, and no longer has any effect. All IN expressions "
"are now rendered using "
'the "expanding parameter" strategy which renders a set of bound'
'expressions, or an "empty set" SELECT, at statement execution'
"time.",
),
implicit_returning=(
"2.0",
"The :paramref:`_sa.create_engine.implicit_returning` parameter "
"is deprecated and will be removed in a future release. ",
),
)
def create_engine(url: Union[str, _url.URL], **kwargs: Any) -> Engine:
"""Create a new :class:`_engine.Engine` instance.
The standard calling form is to send the :ref:`URL <database_urls>` as the
first positional argument, usually a string
that indicates database dialect and connection arguments::
engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
.. note::
Please review :ref:`database_urls` for general guidelines in composing
URL strings. In particular, special characters, such as those often
part of passwords, must be URL encoded to be properly parsed.
Additional keyword arguments may then follow it which
establish various options on the resulting :class:`_engine.Engine`
and its underlying :class:`.Dialect` and :class:`_pool.Pool`
constructs::
engine = create_engine(
"mysql+mysqldb://scott:tiger@hostname/dbname",
pool_recycle=3600,
echo=True,
)
The string form of the URL is
``dialect[+driver]://user:password@host/dbname[?key=value..]``, where
``dialect`` is a database name such as ``mysql``, ``oracle``,
``postgresql``, etc., and ``driver`` the name of a DBAPI, such as
``psycopg2``, ``pyodbc``, ``cx_oracle``, etc. Alternatively,
the URL can be an instance of :class:`~sqlalchemy.engine.url.URL`.
``**kwargs`` takes a wide variety of options which are routed
towards their appropriate components. Arguments may be specific to
the :class:`_engine.Engine`, the underlying :class:`.Dialect`,
as well as the
:class:`_pool.Pool`. Specific dialects also accept keyword arguments that
are unique to that dialect. Here, we describe the parameters
that are common to most :func:`_sa.create_engine()` usage.
Once established, the newly resulting :class:`_engine.Engine` will
request a connection from the underlying :class:`_pool.Pool` once
:meth:`_engine.Engine.connect` is called, or a method which depends on it
such as :meth:`_engine.Engine.execute` is invoked. The
:class:`_pool.Pool` in turn
will establish the first actual DBAPI connection when this request
is received. The :func:`_sa.create_engine` call itself does **not**
establish any actual DBAPI connections directly.
.. seealso::
:doc:`/core/engines`
:doc:`/dialects/index`
:ref:`connections_toplevel`
:param connect_args: a dictionary of options which will be
passed directly to the DBAPI's ``connect()`` method as
additional keyword arguments. See the example
at :ref:`custom_dbapi_args`.
:param creator: a callable which returns a DBAPI connection.
This creation function will be passed to the underlying
connection pool and will be used to create all new database
connections. Usage of this function causes connection
parameters specified in the URL argument to be bypassed.
This hook is not as flexible as the newer
:meth:`_events.DialectEvents.do_connect` hook which allows complete
control over how a connection is made to the database, given the full
set of URL arguments and state beforehand.
.. seealso::
:meth:`_events.DialectEvents.do_connect` - event hook that allows
full control over DBAPI connection mechanics.
:ref:`custom_dbapi_args`
:param echo=False: if True, the Engine will log all statements
as well as a ``repr()`` of their parameter lists to the default log
handler, which defaults to ``sys.stdout`` for output. If set to the
string ``"debug"``, result rows will be printed to the standard output
as well. The ``echo`` attribute of ``Engine`` can be modified at any
time to turn logging on and off; direct control of logging is also
available using the standard Python ``logging`` module.
.. seealso::
:ref:`dbengine_logging` - further detail on how to configure
logging.
:param echo_pool=False: if True, the connection pool will log
informational output such as when connections are invalidated
as well as when connections are recycled to the default log handler,
which defaults to ``sys.stdout`` for output. If set to the string
``"debug"``, the logging will include pool checkouts and checkins.
Direct control of logging is also available using the standard Python
``logging`` module.
.. seealso::
:ref:`dbengine_logging` - further detail on how to configure
logging.
:param empty_in_strategy: No longer used; SQLAlchemy now uses
"empty set" behavior for IN in all cases.
:param enable_from_linting: defaults to True. Will emit a warning
if a given SELECT statement is found to have un-linked FROM elements
which would cause a cartesian product.
.. versionadded:: 1.4
.. seealso::
:ref:`change_4737`
:param execution_options: Dictionary execution options which will
be applied to all connections. See
:meth:`~sqlalchemy.engine.Connection.execution_options`
:param future: Use the 2.0 style :class:`_engine.Engine` and
:class:`_engine.Connection` API.
As of SQLAlchemy 2.0, this parameter is present for backwards
compatibility only and must remain at its default value of ``True``.
The :paramref:`_sa.create_engine.future` parameter will be
deprecated in a subsequent 2.x release and eventually removed.
.. versionadded:: 1.4
.. versionchanged:: 2.0 All :class:`_engine.Engine` objects are
"future" style engines and there is no longer a ``future=False``
mode of operation.
.. seealso::
:ref:`migration_20_toplevel`
:param hide_parameters: Boolean, when set to True, SQL statement parameters
will not be displayed in INFO logging nor will they be formatted into
the string representation of :class:`.StatementError` objects.
.. versionadded:: 1.3.8
.. seealso::
:ref:`dbengine_logging` - further detail on how to configure
logging.
:param implicit_returning=True: Legacy parameter that may only be set
to True. In SQLAlchemy 2.0, this parameter does nothing. In order to
disable "implicit returning" for statements invoked by the ORM,
configure this on a per-table basis using the
:paramref:`.Table.implicit_returning` parameter.
:param insertmanyvalues_page_size: number of rows to format into an
INSERT statement when the statement uses "insertmanyvalues" mode, which is
a paged form of bulk insert that is used for many backends when using
:term:`executemany` execution typically in conjunction with RETURNING.
Defaults to 1000, but may also be subject to dialect-specific limiting
factors which may override this value on a per-statement basis.
.. versionadded:: 2.0
.. seealso::
:ref:`engine_insertmanyvalues`
:ref:`engine_insertmanyvalues_page_size`
:paramref:`_engine.Connection.execution_options.insertmanyvalues_page_size`
:param isolation_level: optional string name of an isolation level
which will be set on all new connections unconditionally.
Isolation levels are typically some subset of the string names
``"SERIALIZABLE"``, ``"REPEATABLE READ"``,
``"READ COMMITTED"``, ``"READ UNCOMMITTED"`` and ``"AUTOCOMMIT"``
based on backend.
The :paramref:`_sa.create_engine.isolation_level` parameter is
in contrast to the
:paramref:`.Connection.execution_options.isolation_level`
execution option, which may be set on an individual
:class:`.Connection`, as well as the same parameter passed to
:meth:`.Engine.execution_options`, where it may be used to create
multiple engines with different isolation levels that share a common
connection pool and dialect.
.. versionchanged:: 2.0 The
:paramref:`_sa.create_engine.isolation_level`
parameter has been generalized to work on all dialects which support
the concept of isolation level, and is provided as a more succinct,
up front configuration switch in contrast to the execution option
which is more of an ad-hoc programmatic option.
.. seealso::
:ref:`dbapi_autocommit`
:param json_deserializer: for dialects that support the
:class:`_types.JSON`
datatype, this is a Python callable that will convert a JSON string
to a Python object. By default, the Python ``json.loads`` function is
used.
.. versionchanged:: 1.3.7 The SQLite dialect renamed this from
``_json_deserializer``.
:param json_serializer: for dialects that support the :class:`_types.JSON`
datatype, this is a Python callable that will render a given object
as JSON. By default, the Python ``json.dumps`` function is used.
.. versionchanged:: 1.3.7 The SQLite dialect renamed this from
``_json_serializer``.
:param label_length=None: optional integer value which limits
the size of dynamically generated column labels to that many
characters. If less than 6, labels are generated as
"_(counter)". If ``None``, the value of
``dialect.max_identifier_length``, which may be affected via the
:paramref:`_sa.create_engine.max_identifier_length` parameter,
is used instead. The value of
:paramref:`_sa.create_engine.label_length`
may not be larger than that of
:paramref:`_sa.create_engine.max_identfier_length`.
.. seealso::
:paramref:`_sa.create_engine.max_identifier_length`
:param logging_name: String identifier which will be used within
the "name" field of logging records generated within the
"sqlalchemy.engine" logger. Defaults to a hexstring of the
object's id.
.. seealso::
:ref:`dbengine_logging` - further detail on how to configure
logging.
:paramref:`_engine.Connection.execution_options.logging_token`
:param max_identifier_length: integer; override the max_identifier_length
determined by the dialect. if ``None`` or zero, has no effect. This
is the database's configured maximum number of characters that may be
used in a SQL identifier such as a table name, column name, or label
name. All dialects determine this value automatically, however in the
case of a new database version for which this value has changed but
SQLAlchemy's dialect has not been adjusted, the value may be passed
here.
.. versionadded:: 1.3.9
.. seealso::
:paramref:`_sa.create_engine.label_length`
:param max_overflow=10: the number of connections to allow in
connection pool "overflow", that is connections that can be
opened above and beyond the pool_size setting, which defaults
to five. this is only used with :class:`~sqlalchemy.pool.QueuePool`.
:param module=None: reference to a Python module object (the module
itself, not its string name). Specifies an alternate DBAPI module to
be used by the engine's dialect. Each sub-dialect references a
specific DBAPI which will be imported before first connect. This
parameter causes the import to be bypassed, and the given module to
be used instead. Can be used for testing of DBAPIs as well as to
inject "mock" DBAPI implementations into the :class:`_engine.Engine`.
:param paramstyle=None: The `paramstyle <https://legacy.python.org/dev/peps/pep-0249/#paramstyle>`_
to use when rendering bound parameters. This style defaults to the
one recommended by the DBAPI itself, which is retrieved from the
``.paramstyle`` attribute of the DBAPI. However, most DBAPIs accept
more than one paramstyle, and in particular it may be desirable
to change a "named" paramstyle into a "positional" one, or vice versa.
When this attribute is passed, it should be one of the values
``"qmark"``, ``"numeric"``, ``"named"``, ``"format"`` or
``"pyformat"``, and should correspond to a parameter style known
to be supported by the DBAPI in use.
:param pool=None: an already-constructed instance of
:class:`~sqlalchemy.pool.Pool`, such as a
:class:`~sqlalchemy.pool.QueuePool` instance. If non-None, this
pool will be used directly as the underlying connection pool
for the engine, bypassing whatever connection parameters are
present in the URL argument. For information on constructing
connection pools manually, see :ref:`pooling_toplevel`.
:param poolclass=None: a :class:`~sqlalchemy.pool.Pool`
subclass, which will be used to create a connection pool
instance using the connection parameters given in the URL. Note
this differs from ``pool`` in that you don't actually
instantiate the pool in this case, you just indicate what type
of pool to be used.
:param pool_logging_name: String identifier which will be used within
the "name" field of logging records generated within the
"sqlalchemy.pool" logger. Defaults to a hexstring of the object's
id.
.. seealso::
:ref:`dbengine_logging` - further detail on how to configure
logging.
:param pool_pre_ping: boolean, if True will enable the connection pool
"pre-ping" feature that tests connections for liveness upon
each checkout.
.. versionadded:: 1.2
.. seealso::
:ref:`pool_disconnects_pessimistic`
:param pool_size=5: the number of connections to keep open
inside the connection pool. This used with
:class:`~sqlalchemy.pool.QueuePool` as
well as :class:`~sqlalchemy.pool.SingletonThreadPool`. With
:class:`~sqlalchemy.pool.QueuePool`, a ``pool_size`` setting
of 0 indicates no limit; to disable pooling, set ``poolclass`` to
:class:`~sqlalchemy.pool.NullPool` instead.
:param pool_recycle=-1: this setting causes the pool to recycle
connections after the given number of seconds has passed. It
defaults to -1, or no timeout. For example, setting to 3600
means connections will be recycled after one hour. Note that
MySQL in particular will disconnect automatically if no
activity is detected on a connection for eight hours (although
this is configurable with the MySQLDB connection itself and the
server configuration as well).
.. seealso::
:ref:`pool_setting_recycle`
:param pool_reset_on_return='rollback': set the
:paramref:`_pool.Pool.reset_on_return` parameter of the underlying
:class:`_pool.Pool` object, which can be set to the values
``"rollback"``, ``"commit"``, or ``None``.
.. seealso::
:ref:`pool_reset_on_return`
:ref:`dbapi_autocommit_skip_rollback` - a more modern approach
to using connections with no transactional instructions
:param pool_timeout=30: number of seconds to wait before giving
up on getting a connection from the pool. This is only used
with :class:`~sqlalchemy.pool.QueuePool`. This can be a float but is
subject to the limitations of Python time functions which may not be
reliable in the tens of milliseconds.
.. note: don't use 30.0 above, it seems to break with the :param tag
:param pool_use_lifo=False: use LIFO (last-in-first-out) when retrieving
connections from :class:`.QueuePool` instead of FIFO
(first-in-first-out). Using LIFO, a server-side timeout scheme can
reduce the number of connections used during non- peak periods of
use. When planning for server-side timeouts, ensure that a recycle or
pre-ping strategy is in use to gracefully handle stale connections.
.. versionadded:: 1.3
.. seealso::
:ref:`pool_use_lifo`
:ref:`pool_disconnects`
:param plugins: string list of plugin names to load. See
:class:`.CreateEnginePlugin` for background.
.. versionadded:: 1.2.3
:param query_cache_size: size of the cache used to cache the SQL string
form of queries. Set to zero to disable caching.
The cache is pruned of its least recently used items when its size reaches
N * 1.5. Defaults to 500, meaning the cache will always store at least
500 SQL statements when filled, and will grow up to 750 items at which
point it is pruned back down to 500 by removing the 250 least recently
used items.
Caching is accomplished on a per-statement basis by generating a
cache key that represents the statement's structure, then generating
string SQL for the current dialect only if that key is not present
in the cache. All statements support caching, however some features
such as an INSERT with a large set of parameters will intentionally
bypass the cache. SQL logging will indicate statistics for each
statement whether or not it were pull from the cache.
.. note:: some ORM functions related to unit-of-work persistence as well
as some attribute loading strategies will make use of individual
per-mapper caches outside of the main cache.
.. seealso::
:ref:`sql_caching`
.. versionadded:: 1.4
:param skip_autocommit_rollback: When True, the dialect will
unconditionally skip all calls to the DBAPI ``connection.rollback()``
method if the DBAPI connection is confirmed to be in "autocommit" mode.
The availability of this feature is dialect specific; if not available,
a ``NotImplementedError`` is raised by the dialect when rollback occurs.
.. seealso::
:ref:`dbapi_autocommit_skip_rollback`
.. versionadded:: 2.0.43
:param use_insertmanyvalues: True by default, use the "insertmanyvalues"
execution style for INSERT..RETURNING statements by default.
.. versionadded:: 2.0
.. seealso::
:ref:`engine_insertmanyvalues`
""" # noqa
if "strategy" in kwargs:
strat = kwargs.pop("strategy")
if strat == "mock":
# this case is deprecated
return create_mock_engine(url, **kwargs) # type: ignore
else:
raise exc.ArgumentError("unknown strategy: %r" % strat)
kwargs.pop("empty_in_strategy", None)
# create url.URL object
u = _url.make_url(url)
u, plugins, kwargs = u._instantiate_plugins(kwargs)
entrypoint = u._get_entrypoint()
_is_async = kwargs.pop("_is_async", False)
if _is_async:
dialect_cls = entrypoint.get_async_dialect_cls(u)
else:
dialect_cls = entrypoint.get_dialect_cls(u)
if kwargs.pop("_coerce_config", False):
def pop_kwarg(key: str, default: Optional[Any] = None) -> Any:
value = kwargs.pop(key, default)
if key in dialect_cls.engine_config_types:
value = dialect_cls.engine_config_types[key](value)
return value
else:
pop_kwarg = kwargs.pop # type: ignore
dialect_args = {}
# consume dialect arguments from kwargs
for k in util.get_cls_kwargs(dialect_cls):
if k in kwargs:
dialect_args[k] = pop_kwarg(k)
dbapi = kwargs.pop("module", None)
if dbapi is None:
dbapi_args = {}
if "import_dbapi" in dialect_cls.__dict__:
dbapi_meth = dialect_cls.import_dbapi
elif hasattr(dialect_cls, "dbapi") and inspect.ismethod(
dialect_cls.dbapi
):
util.warn_deprecated(
"The dbapi() classmethod on dialect classes has been "
"renamed to import_dbapi(). Implement an import_dbapi() "
f"classmethod directly on class {dialect_cls} to remove this "
"warning; the old .dbapi() classmethod may be maintained for "
"backwards compatibility.",
"2.0",
)
dbapi_meth = dialect_cls.dbapi
else:
dbapi_meth = dialect_cls.import_dbapi
for k in util.get_func_kwargs(dbapi_meth):
if k in kwargs:
dbapi_args[k] = pop_kwarg(k)
dbapi = dbapi_meth(**dbapi_args)
dialect_args["dbapi"] = dbapi
dialect_args.setdefault("compiler_linting", compiler.NO_LINTING)
enable_from_linting = kwargs.pop("enable_from_linting", True)
if enable_from_linting:
dialect_args["compiler_linting"] ^= compiler.COLLECT_CARTESIAN_PRODUCTS
for plugin in plugins:
plugin.handle_dialect_kwargs(dialect_cls, dialect_args)
# create dialect
dialect = dialect_cls(**dialect_args)
# assemble connection arguments
cargs_tup, _cparams = dialect.create_connect_args(u)
cparams = util.immutabledict(_cparams).union(pop_kwarg("connect_args", {}))
if "async_fallback" in cparams and util.asbool(cparams["async_fallback"]):
util.warn_deprecated(
"The async_fallback dialect argument is deprecated and will be "
"removed in SQLAlchemy 2.1.",
"2.0",
)
# look for existing pool or create
pool = pop_kwarg("pool", None)
if pool is None:
def connect(
connection_record: Optional[ConnectionPoolEntry] = None,
) -> DBAPIConnection:
if dialect._has_events:
mutable_cargs = list(cargs_tup)
mutable_cparams = dict(cparams)
for fn in dialect.dispatch.do_connect:
connection = cast(
DBAPIConnection,
fn(
dialect,
connection_record,
mutable_cargs,
mutable_cparams,
),
)
if connection is not None:
return connection
return dialect.connect(*mutable_cargs, **mutable_cparams)
else:
return dialect.connect(*cargs_tup, **cparams)
creator = pop_kwarg("creator", connect)
poolclass = pop_kwarg("poolclass", None)
if poolclass is None:
poolclass = dialect.get_dialect_pool_class(u)
pool_args = {"dialect": dialect}
# consume pool arguments from kwargs, translating a few of
# the arguments
for k in util.get_cls_kwargs(poolclass):
tk = _pool_translate_kwargs.get(k, k)
if tk in kwargs:
pool_args[k] = pop_kwarg(tk)
for plugin in plugins:
plugin.handle_pool_kwargs(poolclass, pool_args)
pool = poolclass(creator, **pool_args)
else:
pool._dialect = dialect
if (
hasattr(pool, "_is_asyncio")
and pool._is_asyncio is not dialect.is_async
):
raise exc.ArgumentError(
f"Pool class {pool.__class__.__name__} cannot be "
f"used with {'non-' if not dialect.is_async else ''}"
"asyncio engine",
code="pcls",
)
# create engine.
if not pop_kwarg("future", True):
raise exc.ArgumentError(
"The 'future' parameter passed to "
"create_engine() may only be set to True."
)
engineclass = base.Engine
engine_args = {}
for k in util.get_cls_kwargs(engineclass):
if k in kwargs:
engine_args[k] = pop_kwarg(k)
# internal flags used by the test suite for instrumenting / proxying
# engines with mocks etc.
_initialize = kwargs.pop("_initialize", True)
# all kwargs should be consumed
if kwargs:
raise TypeError(
"Invalid argument(s) %s sent to create_engine(), "
"using configuration %s/%s/%s. Please check that the "
"keyword arguments are appropriate for this combination "
"of components."
% (
",".join("'%s'" % k for k in kwargs),
dialect.__class__.__name__,
pool.__class__.__name__,
engineclass.__name__,
)
)
engine = engineclass(pool, dialect, u, **engine_args)
if _initialize:
do_on_connect = dialect.on_connect_url(u)
if do_on_connect:
def on_connect(
dbapi_connection: DBAPIConnection,
connection_record: ConnectionPoolEntry,
) -> None:
assert do_on_connect is not None
do_on_connect(dbapi_connection)
event.listen(pool, "connect", on_connect)
builtin_on_connect = dialect._builtin_onconnect()
if builtin_on_connect:
event.listen(pool, "connect", builtin_on_connect)
def first_connect(
dbapi_connection: DBAPIConnection,
connection_record: ConnectionPoolEntry,
) -> None:
c = base.Connection(
engine,
connection=_AdhocProxiedConnection(
dbapi_connection, connection_record
),
_has_events=False,
# reconnecting will be a reentrant condition, so if the
# connection goes away, Connection is then closed
_allow_revalidate=False,
# dont trigger the autobegin sequence
# within the up front dialect checks
_allow_autobegin=False,
)
c._execution_options = util.EMPTY_DICT
try:
dialect.initialize(c)
finally:
# note that "invalidated" and "closed" are mutually
# exclusive in 1.4 Connection.
if not c.invalidated and not c.closed:
# transaction is rolled back otherwise, tested by
# test/dialect/postgresql/test_dialect.py
# ::MiscBackendTest::test_initial_transaction_state
dialect.do_rollback(c.connection)
# previously, the "first_connect" event was used here, which was then
# scaled back if the "on_connect" handler were present. now,
# since "on_connect" is virtually always present, just use
# "connect" event with once_unless_exception in all cases so that
# the connection event flow is consistent in all cases.
event.listen(
pool, "connect", first_connect, _once_unless_exception=True
)
dialect_cls.engine_created(engine)
if entrypoint is not dialect_cls:
entrypoint.engine_created(engine)
for plugin in plugins:
plugin.engine_created(engine)
return engine
def engine_from_config(
configuration: Dict[str, Any], prefix: str = "sqlalchemy.", **kwargs: Any
) -> Engine:
"""Create a new Engine instance using a configuration dictionary.
The dictionary is typically produced from a config file.
The keys of interest to ``engine_from_config()`` should be prefixed, e.g.
``sqlalchemy.url``, ``sqlalchemy.echo``, etc. The 'prefix' argument
indicates the prefix to be searched for. Each matching key (after the
prefix is stripped) is treated as though it were the corresponding keyword
argument to a :func:`_sa.create_engine` call.
The only required key is (assuming the default prefix) ``sqlalchemy.url``,
which provides the :ref:`database URL <database_urls>`.
A select set of keyword arguments will be "coerced" to their
expected type based on string values. The set of arguments
is extensible per-dialect using the ``engine_config_types`` accessor.
:param configuration: A dictionary (typically produced from a config file,
but this is not a requirement). Items whose keys start with the value
of 'prefix' will have that prefix stripped, and will then be passed to
:func:`_sa.create_engine`.
:param prefix: Prefix to match and then strip from keys
in 'configuration'.
:param kwargs: Each keyword argument to ``engine_from_config()`` itself
overrides the corresponding item taken from the 'configuration'
dictionary. Keyword arguments should *not* be prefixed.
"""
options = {
key[len(prefix) :]: configuration[key]
for key in configuration
if key.startswith(prefix)
}
options["_coerce_config"] = True
options.update(kwargs)
url = options.pop("url")
return create_engine(url, **options)
@overload
def create_pool_from_url(
url: Union[str, URL],
*,
poolclass: Optional[Type[Pool]] = ...,
logging_name: str = ...,
pre_ping: bool = ...,
size: int = ...,
recycle: int = ...,
reset_on_return: Optional[_ResetStyleArgType] = ...,
timeout: float = ...,
use_lifo: bool = ...,
**kwargs: Any,
) -> Pool: ...
@overload
def create_pool_from_url(url: Union[str, URL], **kwargs: Any) -> Pool: ...
def create_pool_from_url(url: Union[str, URL], **kwargs: Any) -> Pool:
"""Create a pool instance from the given url.
If ``poolclass`` is not provided the pool class used
is selected using the dialect specified in the URL.
The arguments passed to :func:`_sa.create_pool_from_url` are
identical to the pool argument passed to the :func:`_sa.create_engine`
function.
.. versionadded:: 2.0.10
"""
for key in _pool_translate_kwargs:
if key in kwargs:
kwargs[_pool_translate_kwargs[key]] = kwargs.pop(key)
engine = create_engine(url, **kwargs, _initialize=False)
return engine.pool
_pool_translate_kwargs = immutabledict(
{
"logging_name": "pool_logging_name",
"echo": "echo_pool",
"timeout": "pool_timeout",
"recycle": "pool_recycle",
"events": "pool_events", # deprecated
"reset_on_return": "pool_reset_on_return",
"pre_ping": "pool_pre_ping",
"use_lifo": "pool_use_lifo",
}
)

2307
venv/Lib/site-packages/sqlalchemy/engine/cursor.py Normal file
View File

File diff suppressed because it is too large Load Diff

2394
venv/Lib/site-packages/sqlalchemy/engine/default.py Normal file
View File

File diff suppressed because it is too large Load Diff

965
venv/Lib/site-packages/sqlalchemy/engine/events.py Normal file
View File

@@ -0,0 +1,965 @@
# engine/events.py
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
from __future__ import annotations
import typing
from typing import Any
from typing import 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]):
"""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("Received statement: %s", statement)
engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
event.listen(engine, "before_cursor_execute", before_cursor_execute)
or with a specific :class:`_engine.Connection`::
with engine.begin() as conn:
@event.listens_for(conn, "before_cursor_execute")
def before_cursor_execute(
conn, cursor, statement, parameters, context, executemany
):
log.info("Received statement: %s", 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, "before_cursor_execute", retval=True)
def comment_sql_calls(
conn, cursor, statement, parameters, context, executemany
):
statement = statement + " -- some comment"
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.
""" # noqa
_target_class_doc = "SomeEngine"
_dispatch_target = ConnectionEventsTarget
@classmethod
def _accept_with(
cls,
target: Union[ConnectionEventsTarget, Type[ConnectionEventsTarget]],
identifier: str,
) -> Optional[Union[ConnectionEventsTarget, Type[ConnectionEventsTarget]]]:
default_dispatch = super()._accept_with(target, identifier)
if default_dispatch is None and hasattr(
target, "_no_async_engine_events"
):
target._no_async_engine_events()
return default_dispatch
@classmethod
def _listen(
cls,
event_key: event._EventKey[ConnectionEventsTarget],
*,
retval: bool = False,
**kw: Any,
) -> None:
target, identifier, fn = (
event_key.dispatch_target,
event_key.identifier,
event_key._listen_fn,
)
target._has_events = True
if not retval:
if identifier == "before_execute":
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 == "before_cursor_execute":
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 (
"before_execute",
"before_cursor_execute",
):
raise exc.ArgumentError(
"Only the 'before_execute', "
"'before_cursor_execute' and 'handle_error' engine "
"event listeners accept the 'retval=True' "
"argument."
)
event_key.with_wrapper(fn).base_listen()
@event._legacy_signature(
"1.4",
["conn", "clauseelement", "multiparams", "params"],
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,
) -> Optional[
Tuple[Executable, _CoreMultiExecuteParams, _CoreSingleExecuteParams]
]:
"""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, "before_execute", 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`
"""
@event._legacy_signature(
"1.4",
["conn", "clauseelement", "multiparams", "params", "result"],
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],
) -> None:
"""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.
"""
def before_cursor_execute(
self,
conn: Connection,
cursor: DBAPICursor,
statement: str,
parameters: _DBAPIAnyExecuteParams,
context: Optional[ExecutionContext],
executemany: bool,
) -> Optional[Tuple[str, _DBAPIAnyExecuteParams]]:
"""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, "before_cursor_execute", 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`
"""
def after_cursor_execute(
self,
conn: Connection,
cursor: DBAPICursor,
statement: str,
parameters: _DBAPIAnyExecuteParams,
context: Optional[ExecutionContext],
executemany: bool,
) -> None:
"""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.
"""
@event._legacy_signature(
"2.0", ["conn", "branch"], converter=lambda conn: (conn, False)
)
def engine_connect(self, conn: Connection) -> None:
"""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
"""
def set_connection_execution_options(
self, conn: Connection, opts: Dict[str, Any]
) -> None:
"""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.
"""
def set_engine_execution_options(
self, engine: Engine, opts: Dict[str, Any]
) -> None:
"""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.
"""
def engine_disposed(self, engine: Engine) -> None:
"""Intercept when the :meth:`_engine.Engine.dispose` method is called.
The :meth:`_engine.Engine.dispose` method instructs the engine to
"dispose" 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.
"""
def begin(self, conn: Connection) -> None:
"""Intercept begin() events.
:param conn: :class:`_engine.Connection` object
"""
def rollback(self, conn: Connection) -> None:
"""Intercept rollback() events, as initiated by a
:class:`.Transaction`.
Note that the :class:`_pool.Pool` also "auto-rolls back"
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`
"""
def commit(self, conn: Connection) -> None:
"""Intercept commit() events, as initiated by a
:class:`.Transaction`.
Note that the :class:`_pool.Pool` may also "auto-commit"
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
"""
def savepoint(self, conn: Connection, name: str) -> None:
"""Intercept savepoint() events.
:param conn: :class:`_engine.Connection` object
:param name: specified name used for the savepoint.
"""
def rollback_savepoint(
self, conn: Connection, name: str, context: None
) -> None:
"""Intercept rollback_savepoint() events.
:param conn: :class:`_engine.Connection` object
:param name: specified name used for the savepoint.
:param context: not used
"""
# TODO: deprecate "context"
def release_savepoint(
self, conn: Connection, name: str, context: None
) -> None:
"""Intercept release_savepoint() events.
:param conn: :class:`_engine.Connection` object
:param name: specified name used for the savepoint.
:param context: not used
"""
# TODO: deprecate "context"
def begin_twophase(self, conn: Connection, xid: Any) -> None:
"""Intercept begin_twophase() events.
:param conn: :class:`_engine.Connection` object
:param xid: two-phase XID identifier
"""
def prepare_twophase(self, conn: Connection, xid: Any) -> None:
"""Intercept prepare_twophase() events.
:param conn: :class:`_engine.Connection` object
:param xid: two-phase XID identifier
"""
def rollback_twophase(
self, conn: Connection, xid: Any, is_prepared: bool
) -> None:
"""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.
"""
def commit_twophase(
self, conn: Connection, xid: Any, is_prepared: bool
) -> None:
"""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.
"""
class DialectEvents(event.Events[Dialect]):
"""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`
"""
_target_class_doc = "SomeEngine"
_dispatch_target = Dialect
@classmethod
def _listen(
cls,
event_key: event._EventKey[Dialect],
*,
retval: bool = False,
**kw: Any,
) -> 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,
) -> 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 == "handle_error":
raise exc.InvalidRequestError(
"The handle_error() event hook as of SQLAlchemy 2.0 is "
"established on the Dialect, and may only be applied to the "
"Engine as a whole or to a specific Dialect as a whole, "
"not on a per-Connection basis."
)
elif hasattr(target, "_no_async_engine_events"):
target._no_async_engine_events()
else:
return None
def handle_error(
self, exception_context: ExceptionContext
) -> Optional[BaseException]:
r"""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 "pre ping" 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 "pre_ping" 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 "pre_ping" 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 "pre-ping" 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 "pre-ping" 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, "handle_error")
def handle_exception(context):
if isinstance(
context.original_exception, psycopg2.OperationalError
) and "failed" in str(context.original_exception):
raise MySpecialException("failed operation")
.. 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 "chained" 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 "chained" exception is available using
:attr:`.ExceptionContext.chained_exception`::
@event.listens_for(Engine, "handle_error", retval=True)
def handle_exception(context):
if (
context.chained_exception is not None
and "special" in context.chained_exception.message
):
return MySpecialException(
"failed", 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 "autoflush" 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`
"""
def do_connect(
self,
dialect: Dialect,
conn_rec: ConnectionPoolEntry,
cargs: Tuple[Any, ...],
cparams: Dict[str, Any],
) -> Optional[DBAPIConnection]:
"""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("postgresql+psycopg2://user@host/dbname")
@event.listens_for(e, "do_connect")
def receive_do_connect(dialect, conn_rec, cargs, cparams):
cparams["password"] = "some_password"
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("postgresql+psycopg2://user@host/dbname")
@event.listens_for(e, "do_connect")
def receive_do_connect(dialect, conn_rec, cargs, cparams):
return psycopg2.connect(*cargs, **cparams)
.. seealso::
:ref:`custom_dbapi_args`
"""
def do_executemany(
self,
cursor: DBAPICursor,
statement: str,
parameters: _DBAPIMultiExecuteParams,
context: ExecutionContext,
) -> Optional[Literal[True]]:
"""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.
"""
def do_execute_no_params(
self, cursor: DBAPICursor, statement: str, context: ExecutionContext
) -> Optional[Literal[True]]:
"""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.
"""
def do_execute(
self,
cursor: DBAPICursor,
statement: str,
parameters: _DBAPISingleExecuteParams,
context: ExecutionContext,
) -> Optional[Literal[True]]:
"""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.
"""
def do_setinputsizes(
self,
inputsizes: Dict[BindParameter[Any], Any],
cursor: DBAPICursor,
statement: str,
parameters: _DBAPIAnyExecuteParams,
context: ExecutionContext,
) -> None:
"""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("mssql+pyodbc://...", use_setinputsizes=True)
.. seealso::
:ref:`mssql_pyodbc_setinputsizes`
.. versionadded:: 1.2.9
.. seealso::
:ref:`cx_oracle_setinputsizes`
"""
pass