# ext/asyncio/engine.py # Copyright (C) 2020-2022 the SQLAlchemy authors and contributors # # # 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 from typing import Any from typing import Dict from typing import Generator from typing import NoReturn from typing import Optional from typing import overload from typing import Tuple from typing import Type from typing import TYPE_CHECKING from typing import TypeVar from typing import Union from . import exc as async_exc from .base import ProxyComparable from .base import StartableContext from .result import _ensure_sync_result from .result import AsyncResult from .result import AsyncScalarResult from ... import exc from ... import inspection from ... import util from ...engine import Connection from ...engine import create_engine as _create_engine from ...engine import Engine from ...engine.base import NestedTransaction from ...engine.base import Transaction from ...util.concurrency import greenlet_spawn from ...util.typing import Protocol if TYPE_CHECKING: from ...engine.cursor import CursorResult from ...engine.interfaces import _CoreAnyExecuteParams from ...engine.interfaces import _CoreSingleExecuteParams from ...engine.interfaces import _DBAPIAnyExecuteParams from ...engine.interfaces import _ExecuteOptions from ...engine.interfaces import _ExecuteOptionsParameter from ...engine.interfaces import _IsolationLevel from ...engine.interfaces import Dialect from ...engine.result import ScalarResult from ...engine.url import URL from ...pool import Pool from ...pool import PoolProxiedConnection from ...sql._typing import _InfoType from ...sql.base import Executable from ...sql.selectable import TypedReturnsRows _T = TypeVar("_T", bound=Any) class _SyncConnectionCallable(Protocol): def __call__(self, connection: Connection, *arg: Any, **kw: Any) -> Any: ... def create_async_engine(url: Union[str, URL], **kw: Any) -> AsyncEngine: """Create a new async engine instance. Arguments passed to :func:`_asyncio.create_async_engine` are mostly identical to those passed to the :func:`_sa.create_engine` function. The specified dialect must be an asyncio-compatible dialect such as :ref:`dialect-postgresql-asyncpg`. .. versionadded:: 1.4 """ if kw.get("server_side_cursors", False): raise async_exc.AsyncMethodRequired( "Can't set server_side_cursors for async engine globally; " "use the connection.stream() method for an async " "streaming result set" ) kw["future"] = True kw["_is_async"] = True sync_engine = _create_engine(url, **kw) return AsyncEngine(sync_engine) def async_engine_from_config( configuration: Dict[str, Any], prefix: str = "sqlalchemy.", **kwargs: Any ) -> AsyncEngine: """Create a new AsyncEngine instance using a configuration dictionary. This function is analogous to the :func:`_sa.engine_from_config` function in SQLAlchemy Core, except that the requested dialect must be an asyncio-compatible dialect such as :ref:`dialect-postgresql-asyncpg`. The argument signature of the function is identical to that of :func:`_sa.engine_from_config`. .. versionadded:: 1.4.29 """ options = { key[len(prefix) :]: value for key, value in configuration.items() if key.startswith(prefix) } options["_coerce_config"] = True options.update(kwargs) url = options.pop("url") return create_async_engine(url, **options) class AsyncConnectable: __slots__ = "_slots_dispatch", "__weakref__" @classmethod def _no_async_engine_events(cls) -> NoReturn: raise NotImplementedError( "asynchronous events are not implemented at this time. Apply " "synchronous listeners to the AsyncEngine.sync_engine or " "AsyncConnection.sync_connection attributes." ) @util.create_proxy_methods( Connection, ":class:`_engine.Connection`", ":class:`_asyncio.AsyncConnection`", classmethods=[], methods=[], attributes=[ "closed", "invalidated", "dialect", "default_isolation_level", ], ) class AsyncConnection( ProxyComparable[Connection], StartableContext, AsyncConnectable ): """An asyncio proxy for a :class:`_engine.Connection`. :class:`_asyncio.AsyncConnection` is acquired using the :meth:`_asyncio.AsyncEngine.connect` method of :class:`_asyncio.AsyncEngine`:: from sqlalchemy.ext.asyncio import create_async_engine engine = create_async_engine("postgresql+asyncpg://user:pass@host/dbname") async with engine.connect() as conn: result = await conn.execute(select(table)) .. versionadded:: 1.4 """ # noqa # AsyncConnection is a thin proxy; no state should be added here # that is not retrievable from the "sync" engine / connection, e.g. # current transaction, info, etc. It should be possible to # create a new AsyncConnection that matches this one given only the # "sync" elements. __slots__ = ( "engine", "sync_engine", "sync_connection", ) def __init__( self, async_engine: AsyncEngine, sync_connection: Optional[Connection] = None, ): self.engine = async_engine self.sync_engine = async_engine.sync_engine self.sync_connection = self._assign_proxied(sync_connection) sync_connection: Optional[Connection] """Reference to the sync-style :class:`_engine.Connection` this :class:`_asyncio.AsyncConnection` proxies requests towards. This instance can be used as an event target. .. seealso:: :ref:`asyncio_events` """ sync_engine: Engine """Reference to the sync-style :class:`_engine.Engine` this :class:`_asyncio.AsyncConnection` is associated with via its underlying :class:`_engine.Connection`. This instance can be used as an event target. .. seealso:: :ref:`asyncio_events` """ @classmethod def _regenerate_proxy_for_target( cls, target: Connection ) -> AsyncConnection: return AsyncConnection( AsyncEngine._retrieve_proxy_for_target(target.engine), target ) async def start(self, is_ctxmanager: bool = False) -> AsyncConnection: """Start this :class:`_asyncio.AsyncConnection` object's context outside of using a Python ``with:`` block. """ if self.sync_connection: raise exc.InvalidRequestError("connection is already started") self.sync_connection = self._assign_proxied( await (greenlet_spawn(self.sync_engine.connect)) ) return self @property def connection(self) -> NoReturn: """Not implemented for async; call :meth:`_asyncio.AsyncConnection.get_raw_connection`. """ raise exc.InvalidRequestError( "AsyncConnection.connection accessor is not implemented as the " "attribute may need to reconnect on an invalidated connection. " "Use the get_raw_connection() method." ) async def get_raw_connection(self) -> PoolProxiedConnection: """Return the pooled DBAPI-level connection in use by this :class:`_asyncio.AsyncConnection`. This is a SQLAlchemy connection-pool proxied connection which then has the attribute :attr:`_pool._ConnectionFairy.driver_connection` that refers to the actual driver connection. Its :attr:`_pool._ConnectionFairy.dbapi_connection` refers instead to an :class:`_engine.AdaptedConnection` instance that adapts the driver connection to the DBAPI protocol. """ return await greenlet_spawn(getattr, self._proxied, "connection") @util.ro_non_memoized_property def info(self) -> _InfoType: """Return the :attr:`_engine.Connection.info` dictionary of the underlying :class:`_engine.Connection`. This dictionary is freely writable for user-defined state to be associated with the database connection. This attribute is only available if the :class:`.AsyncConnection` is currently connected. If the :attr:`.AsyncConnection.closed` attribute is ``True``, then accessing this attribute will raise :class:`.ResourceClosedError`. .. versionadded:: 1.4.0b2 """ return self._proxied.info @util.ro_non_memoized_property def _proxied(self) -> Connection: if not self.sync_connection: self._raise_for_not_started() return self.sync_connection def begin(self) -> AsyncTransaction: """Begin a transaction prior to autobegin occurring.""" assert self._proxied return AsyncTransaction(self) def begin_nested(self) -> AsyncTransaction: """Begin a nested transaction and return a transaction handle.""" assert self._proxied return AsyncTransaction(self, nested=True) async def invalidate( self, exception: Optional[BaseException] = None ) -> None: """Invalidate the underlying DBAPI connection associated with this :class:`_engine.Connection`. See the method :meth:`_engine.Connection.invalidate` for full detail on this method. """ return await greenlet_spawn( self._proxied.invalidate, exception=exception ) async def get_isolation_level(self) -> _IsolationLevel: return await greenlet_spawn(self._proxied.get_isolation_level) def in_transaction(self) -> bool: """Return True if a transaction is in progress.""" return self._proxied.in_transaction() def in_nested_transaction(self) -> bool: """Return True if a transaction is in progress. .. versionadded:: 1.4.0b2 """ return self._proxied.in_nested_transaction() def get_transaction(self) -> Optional[AsyncTransaction]: """Return an :class:`.AsyncTransaction` representing the current transaction, if any. This makes use of the underlying synchronous connection's :meth:`_engine.Connection.get_transaction` method to get the current :class:`_engine.Transaction`, which is then proxied in a new :class:`.AsyncTransaction` object. .. versionadded:: 1.4.0b2 """ trans = self._proxied.get_transaction() if trans is not None: return AsyncTransaction._retrieve_proxy_for_target(trans) else: return None def get_nested_transaction(self) -> Optional[AsyncTransaction]: """Return an :class:`.AsyncTransaction` representing the current nested (savepoint) transaction, if any. This makes use of the underlying synchronous connection's :meth:`_engine.Connection.get_nested_transaction` method to get the current :class:`_engine.Transaction`, which is then proxied in a new :class:`.AsyncTransaction` object. .. versionadded:: 1.4.0b2 """ trans = self._proxied.get_nested_transaction() if trans is not None: return AsyncTransaction._retrieve_proxy_for_target(trans) else: return None async def execution_options(self, **opt: Any) -> AsyncConnection: r"""Set non-SQL options for the connection which take effect during execution. This returns this :class:`_asyncio.AsyncConnection` object with the new options added. See :meth:`_future.Connection.execution_options` for full details on this method. """ conn = self._proxied c2 = await greenlet_spawn(conn.execution_options, **opt) assert c2 is conn return self async def commit(self) -> None: """Commit the transaction that is currently in progress. This method commits the current transaction if one has been started. If no transaction was started, the method has no effect, assuming the connection is in a non-invalidated state. A transaction is begun on a :class:`_future.Connection` automatically whenever a statement is first executed, or when the :meth:`_future.Connection.begin` method is called. """ await greenlet_spawn(self._proxied.commit) async def rollback(self) -> None: """Roll back the transaction that is currently in progress. This method rolls back the current transaction if one has been started. If no transaction was started, the method has no effect. If a transaction was started and the connection is in an invalidated state, the transaction is cleared using this method. A transaction is begun on a :class:`_future.Connection` automatically whenever a statement is first executed, or when the :meth:`_future.Connection.begin` method is called. """ await greenlet_spawn(self._proxied.rollback) async def close(self) -> None: """Close this :class:`_asyncio.AsyncConnection`. This has the effect of also rolling back the transaction if one is in place. """ await greenlet_spawn(self._proxied.close) async def exec_driver_sql( self, statement: str, parameters: Optional[_DBAPIAnyExecuteParams] = None, execution_options: Optional[_ExecuteOptionsParameter] = None, ) -> CursorResult[Any]: r"""Executes a driver-level SQL string and return buffered :class:`_engine.Result`. """ result = await greenlet_spawn( self._proxied.exec_driver_sql, statement, parameters, execution_options, _require_await=True, ) return await _ensure_sync_result(result, self.exec_driver_sql) @overload async def stream( self, statement: TypedReturnsRows[_T], parameters: Optional[_CoreAnyExecuteParams] = None, *, execution_options: Optional[_ExecuteOptionsParameter] = None, ) -> AsyncResult[_T]: ... @overload async def stream( self, statement: Executable, parameters: Optional[_CoreAnyExecuteParams] = None, *, execution_options: Optional[_ExecuteOptionsParameter] = None, ) -> AsyncResult[Any]: ... async def stream( self, statement: Executable, parameters: Optional[_CoreAnyExecuteParams] = None, *, execution_options: Optional[_ExecuteOptionsParameter] = None, ) -> AsyncResult[Any]: """Execute a statement and return a streaming :class:`_asyncio.AsyncResult` object.""" result = await greenlet_spawn( self._proxied.execute, statement, parameters, execution_options=util.EMPTY_DICT.merge_with( execution_options, {"stream_results": True} ), _require_await=True, ) if not result.context._is_server_side: # TODO: real exception here assert False, "server side result expected" return AsyncResult(result) @overload async def execute( self, statement: TypedReturnsRows[_T], parameters: Optional[_CoreAnyExecuteParams] = None, *, execution_options: Optional[_ExecuteOptionsParameter] = None, ) -> CursorResult[_T]: ... @overload async def execute( self, statement: Executable, parameters: Optional[_CoreAnyExecuteParams] = None, *, execution_options: Optional[_ExecuteOptionsParameter] = None, ) -> CursorResult[Any]: ... async def execute( self, statement: Executable, parameters: Optional[_CoreAnyExecuteParams] = None, *, execution_options: Optional[_ExecuteOptionsParameter] = None, ) -> CursorResult[Any]: r"""Executes a SQL statement construct and return a buffered :class:`_engine.Result`. :param object: The statement to be executed. This is always an object that is in both the :class:`_expression.ClauseElement` and :class:`_expression.Executable` hierarchies, including: * :class:`_expression.Select` * :class:`_expression.Insert`, :class:`_expression.Update`, :class:`_expression.Delete` * :class:`_expression.TextClause` and :class:`_expression.TextualSelect` * :class:`_schema.DDL` and objects which inherit from :class:`_schema.ExecutableDDLElement` :param parameters: parameters which will be bound into the statement. This may be either a dictionary of parameter names to values, or a mutable sequence (e.g. a list) of dictionaries. When a list of dictionaries is passed, the underlying statement execution will make use of the DBAPI ``cursor.executemany()`` method. When a single dictionary is passed, the DBAPI ``cursor.execute()`` method will be used. :param execution_options: optional dictionary of execution options, which will be associated with the statement execution. This dictionary can provide a subset of the options that are accepted by :meth:`_future.Connection.execution_options`. :return: a :class:`_engine.Result` object. """ result = await greenlet_spawn( self._proxied.execute, statement, parameters, execution_options=execution_options, _require_await=True, ) return await _ensure_sync_result(result, self.execute) @overload async def scalar( self, statement: TypedReturnsRows[Tuple[_T]], parameters: Optional[_CoreSingleExecuteParams] = None, *, execution_options: Optional[_ExecuteOptionsParameter] = None, ) -> Optional[_T]: ... @overload async def scalar( self, statement: Executable, parameters: Optional[_CoreSingleExecuteParams] = None, *, execution_options: Optional[_ExecuteOptionsParameter] = None, ) -> Any: ... async def scalar( self, statement: Executable, parameters: Optional[_CoreSingleExecuteParams] = None, *, execution_options: Optional[_ExecuteOptionsParameter] = None, ) -> Any: r"""Executes a SQL statement construct and returns a scalar object. This method is shorthand for invoking the :meth:`_engine.Result.scalar` method after invoking the :meth:`_future.Connection.execute` method. Parameters are equivalent. :return: a scalar Python value representing the first column of the first row returned. """ result = await self.execute( statement, parameters, execution_options=execution_options ) return result.scalar() @overload async def scalars( self, statement: TypedReturnsRows[Tuple[_T]], parameters: Optional[_CoreSingleExecuteParams] = None, *, execution_options: Optional[_ExecuteOptionsParameter] = None, ) -> ScalarResult[_T]: ... @overload async def scalars( self, statement: Executable, parameters: Optional[_CoreSingleExecuteParams] = None, *, execution_options: Optional[_ExecuteOptionsParameter] = None, ) -> ScalarResult[Any]: ... async def scalars( self, statement: Executable, parameters: Optional[_CoreSingleExecuteParams] = None, *, execution_options: Optional[_ExecuteOptionsParameter] = None, ) -> ScalarResult[Any]: r"""Executes a SQL statement construct and returns a scalar objects. This method is shorthand for invoking the :meth:`_engine.Result.scalars` method after invoking the :meth:`_future.Connection.execute` method. Parameters are equivalent. :return: a :class:`_engine.ScalarResult` object. .. versionadded:: 1.4.24 """ result = await self.execute( statement, parameters, execution_options=execution_options ) return result.scalars() @overload async def stream_scalars( self, statement: TypedReturnsRows[Tuple[_T]], parameters: Optional[_CoreSingleExecuteParams] = None, *, execution_options: Optional[_ExecuteOptionsParameter] = None, ) -> AsyncScalarResult[_T]: ... @overload async def stream_scalars( self, statement: Executable, parameters: Optional[_CoreSingleExecuteParams] = None, *, execution_options: Optional[_ExecuteOptionsParameter] = None, ) -> AsyncScalarResult[Any]: ... async def stream_scalars( self, statement: Executable, parameters: Optional[_CoreSingleExecuteParams] = None, *, execution_options: Optional[_ExecuteOptionsParameter] = None, ) -> AsyncScalarResult[Any]: r"""Executes a SQL statement and returns a streaming scalar result object. This method is shorthand for invoking the :meth:`_engine.AsyncResult.scalars` method after invoking the :meth:`_future.Connection.stream` method. Parameters are equivalent. :return: an :class:`_asyncio.AsyncScalarResult` object. .. versionadded:: 1.4.24 """ result = await self.stream( statement, parameters, execution_options=execution_options ) return result.scalars() async def run_sync( self, fn: _SyncConnectionCallable, *arg: Any, **kw: Any ) -> Any: """Invoke the given sync callable passing self as the first argument. This method maintains the asyncio event loop all the way through to the database connection by running the given callable in a specially instrumented greenlet. E.g.:: with async_engine.begin() as conn: await conn.run_sync(metadata.create_all) .. note:: The provided callable is invoked inline within the asyncio event loop, and will block on traditional IO calls. IO within this callable should only call into SQLAlchemy's asyncio database APIs which will be properly adapted to the greenlet context. .. seealso:: :ref:`session_run_sync` """ return await greenlet_spawn(fn, self._proxied, *arg, **kw) def __await__(self) -> Generator[Any, None, AsyncConnection]: return self.start().__await__() async def __aexit__(self, type_: Any, value: Any, traceback: Any) -> None: await self.close() # START PROXY METHODS AsyncConnection # code within this block is **programmatically, # statically generated** by tools/generate_proxy_methods.py @property def closed(self) -> Any: r"""Return True if this connection is closed. .. container:: class_bases Proxied for the :class:`_engine.Connection` class on behalf of the :class:`_asyncio.AsyncConnection` class. """ # noqa: E501 return self._proxied.closed @property def invalidated(self) -> Any: r"""Return True if this connection was invalidated. .. container:: class_bases Proxied for the :class:`_engine.Connection` class on behalf of the :class:`_asyncio.AsyncConnection` class. This does not indicate whether or not the connection was invalidated at the pool level, however """ # noqa: E501 return self._proxied.invalidated @property def dialect(self) -> Any: r"""Proxy for the :attr:`_engine.Connection.dialect` attribute on behalf of the :class:`_asyncio.AsyncConnection` class. """ # noqa: E501 return self._proxied.dialect @dialect.setter def dialect(self, attr: Any) -> None: self._proxied.dialect = attr @property def default_isolation_level(self) -> Any: r"""The default isolation level assigned to this :class:`_engine.Connection`. .. container:: class_bases Proxied for the :class:`_engine.Connection` class on behalf of the :class:`_asyncio.AsyncConnection` class. This is the isolation level setting that the :class:`_engine.Connection` has when first procured via the :meth:`_engine.Engine.connect` method. This level stays in place until the :paramref:`.Connection.execution_options.isolation_level` is used to change the setting on a per-:class:`_engine.Connection` basis. Unlike :meth:`_engine.Connection.get_isolation_level`, this attribute is set ahead of time from the first connection procured by the dialect, so SQL query is not invoked when this accessor is called. .. versionadded:: 0.9.9 .. seealso:: :meth:`_engine.Connection.get_isolation_level` - view current level :paramref:`_sa.create_engine.isolation_level` - set per :class:`_engine.Engine` isolation level :paramref:`.Connection.execution_options.isolation_level` - set per :class:`_engine.Connection` isolation level """ # noqa: E501 return self._proxied.default_isolation_level # END PROXY METHODS AsyncConnection @util.create_proxy_methods( Engine, ":class:`_engine.Engine`", ":class:`_asyncio.AsyncEngine`", classmethods=[], methods=[ "clear_compiled_cache", "update_execution_options", "get_execution_options", ], attributes=["url", "pool", "dialect", "engine", "name", "driver", "echo"], ) class AsyncEngine(ProxyComparable[Engine], AsyncConnectable): """An asyncio proxy for a :class:`_engine.Engine`. :class:`_asyncio.AsyncEngine` is acquired using the :func:`_asyncio.create_async_engine` function:: from sqlalchemy.ext.asyncio import create_async_engine engine = create_async_engine("postgresql+asyncpg://user:pass@host/dbname") .. versionadded:: 1.4 """ # noqa # AsyncEngine is a thin proxy; no state should be added here # that is not retrievable from the "sync" engine / connection, e.g. # current transaction, info, etc. It should be possible to # create a new AsyncEngine that matches this one given only the # "sync" elements. __slots__ = "sync_engine" _connection_cls: Type[AsyncConnection] = AsyncConnection sync_engine: Engine """Reference to the sync-style :class:`_engine.Engine` this :class:`_asyncio.AsyncEngine` proxies requests towards. This instance can be used as an event target. .. seealso:: :ref:`asyncio_events` """ class _trans_ctx(StartableContext): __slots__ = ("conn", "transaction") conn: AsyncConnection transaction: AsyncTransaction def __init__(self, conn: AsyncConnection): self.conn = conn async def start(self, is_ctxmanager: bool = False) -> AsyncConnection: await self.conn.start(is_ctxmanager=is_ctxmanager) self.transaction = self.conn.begin() await self.transaction.__aenter__() return self.conn async def __aexit__( self, type_: Any, value: Any, traceback: Any ) -> None: await self.transaction.__aexit__(type_, value, traceback) await self.conn.close() def __init__(self, sync_engine: Engine): if not sync_engine.dialect.is_async: raise exc.InvalidRequestError( "The asyncio extension requires an async driver to be used. " f"The loaded {sync_engine.dialect.driver!r} is not async." ) self.sync_engine = self._assign_proxied(sync_engine) @util.ro_non_memoized_property def _proxied(self) -> Engine: return self.sync_engine @classmethod def _regenerate_proxy_for_target(cls, target: Engine) -> AsyncEngine: return AsyncEngine(target) def begin(self) -> AsyncEngine._trans_ctx: """Return a context manager which when entered will deliver an :class:`_asyncio.AsyncConnection` with an :class:`_asyncio.AsyncTransaction` established. E.g.:: async with async_engine.begin() as conn: await conn.execute( text("insert into table (x, y, z) values (1, 2, 3)") ) await conn.execute(text("my_special_procedure(5)")) """ conn = self.connect() return self._trans_ctx(conn) def connect(self) -> AsyncConnection: """Return an :class:`_asyncio.AsyncConnection` object. The :class:`_asyncio.AsyncConnection` will procure a database connection from the underlying connection pool when it is entered as an async context manager:: async with async_engine.connect() as conn: result = await conn.execute(select(user_table)) The :class:`_asyncio.AsyncConnection` may also be started outside of a context manager by invoking its :meth:`_asyncio.AsyncConnection.start` method. """ return self._connection_cls(self) async def raw_connection(self) -> PoolProxiedConnection: """Return a "raw" DBAPI connection from the connection pool. .. seealso:: :ref:`dbapi_connections` """ return await greenlet_spawn(self.sync_engine.raw_connection) def execution_options(self, **opt: Any) -> AsyncEngine: """Return a new :class:`_asyncio.AsyncEngine` that will provide :class:`_asyncio.AsyncConnection` objects with the given execution options. Proxied from :meth:`_future.Engine.execution_options`. See that method for details. """ return AsyncEngine(self.sync_engine.execution_options(**opt)) async def dispose(self, close: bool = True) -> None: """Dispose of the connection pool used by this :class:`_asyncio.AsyncEngine`. :param close: if left at its default of ``True``, has the effect of fully closing all **currently checked in** database connections. Connections that are still checked out will **not** be closed, however they will no longer be associated with this :class:`_engine.Engine`, so when they are closed individually, eventually the :class:`_pool.Pool` which they are associated with will be garbage collected and they will be closed out fully, if not already closed on checkin. If set to ``False``, the previous connection pool is de-referenced, and otherwise not touched in any way. .. seealso:: :meth:`_engine.Engine.dispose` """ return await greenlet_spawn(self.sync_engine.dispose, close=close) # START PROXY METHODS AsyncEngine # code within this block is **programmatically, # statically generated** by tools/generate_proxy_methods.py def clear_compiled_cache(self) -> None: r"""Clear the compiled cache associated with the dialect. .. container:: class_bases Proxied for the :class:`_engine.Engine` class on behalf of the :class:`_asyncio.AsyncEngine` class. This applies **only** to the built-in cache that is established via the :paramref:`_engine.create_engine.query_cache_size` parameter. It will not impact any dictionary caches that were passed via the :paramref:`.Connection.execution_options.query_cache` parameter. .. versionadded:: 1.4 """ # noqa: E501 return self._proxied.clear_compiled_cache() def update_execution_options(self, **opt: Any) -> None: r"""Update the default execution_options dictionary of this :class:`_engine.Engine`. .. container:: class_bases Proxied for the :class:`_engine.Engine` class on behalf of the :class:`_asyncio.AsyncEngine` class. The given keys/values in \**opt are added to the default execution options that will be used for all connections. The initial contents of this dictionary can be sent via the ``execution_options`` parameter to :func:`_sa.create_engine`. .. seealso:: :meth:`_engine.Connection.execution_options` :meth:`_engine.Engine.execution_options` """ # noqa: E501 return self._proxied.update_execution_options(**opt) def get_execution_options(self) -> _ExecuteOptions: r"""Get the non-SQL options which will take effect during execution. .. container:: class_bases Proxied for the :class:`_engine.Engine` class on behalf of the :class:`_asyncio.AsyncEngine` class. .. versionadded: 1.3 .. seealso:: :meth:`_engine.Engine.execution_options` """ # noqa: E501 return self._proxied.get_execution_options() @property def url(self) -> URL: r"""Proxy for the :attr:`_engine.Engine.url` attribute on behalf of the :class:`_asyncio.AsyncEngine` class. """ # noqa: E501 return self._proxied.url @url.setter def url(self, attr: URL) -> None: self._proxied.url = attr @property def pool(self) -> Pool: r"""Proxy for the :attr:`_engine.Engine.pool` attribute on behalf of the :class:`_asyncio.AsyncEngine` class. """ # noqa: E501 return self._proxied.pool @pool.setter def pool(self, attr: Pool) -> None: self._proxied.pool = attr @property def dialect(self) -> Dialect: r"""Proxy for the :attr:`_engine.Engine.dialect` attribute on behalf of the :class:`_asyncio.AsyncEngine` class. """ # noqa: E501 return self._proxied.dialect @dialect.setter def dialect(self, attr: Dialect) -> None: self._proxied.dialect = attr @property def engine(self) -> Any: r""".. container:: class_bases Proxied for the :class:`_engine.Engine` class on behalf of the :class:`_asyncio.AsyncEngine` class. """ # noqa: E501 return self._proxied.engine @property def name(self) -> Any: r"""String name of the :class:`~sqlalchemy.engine.interfaces.Dialect` in use by this :class:`Engine`. .. container:: class_bases Proxied for the :class:`_engine.Engine` class on behalf of the :class:`_asyncio.AsyncEngine` class. """ # noqa: E501 return self._proxied.name @property def driver(self) -> Any: r"""Driver name of the :class:`~sqlalchemy.engine.interfaces.Dialect` in use by this :class:`Engine`. .. container:: class_bases Proxied for the :class:`_engine.Engine` class on behalf of the :class:`_asyncio.AsyncEngine` class. """ # noqa: E501 return self._proxied.driver @property def echo(self) -> Any: r"""When ``True``, enable log output for this element. .. container:: class_bases Proxied for the :class:`_engine.Engine` class on behalf of the :class:`_asyncio.AsyncEngine` class. This has the effect of setting the Python logging level for the namespace of this element's class and object reference. A value of boolean ``True`` indicates that the loglevel ``logging.INFO`` will be set for the logger, whereas the string value ``debug`` will set the loglevel to ``logging.DEBUG``. """ # noqa: E501 return self._proxied.echo @echo.setter def echo(self, attr: Any) -> None: self._proxied.echo = attr # END PROXY METHODS AsyncEngine class AsyncTransaction(ProxyComparable[Transaction], StartableContext): """An asyncio proxy for a :class:`_engine.Transaction`.""" __slots__ = ("connection", "sync_transaction", "nested") sync_transaction: Optional[Transaction] connection: AsyncConnection nested: bool def __init__(self, connection: AsyncConnection, nested: bool = False): self.connection = connection self.sync_transaction = None self.nested = nested @classmethod def _regenerate_proxy_for_target( cls, target: Transaction ) -> AsyncTransaction: sync_connection = target.connection sync_transaction = target nested = isinstance(target, NestedTransaction) async_connection = AsyncConnection._retrieve_proxy_for_target( sync_connection ) assert async_connection is not None obj = cls.__new__(cls) obj.connection = async_connection obj.sync_transaction = obj._assign_proxied(sync_transaction) obj.nested = nested return obj @util.ro_non_memoized_property def _proxied(self) -> Transaction: if not self.sync_transaction: self._raise_for_not_started() return self.sync_transaction @property def is_valid(self) -> bool: return self._proxied.is_valid @property def is_active(self) -> bool: return self._proxied.is_active async def close(self) -> None: """Close this :class:`.AsyncTransaction`. If this transaction is the base transaction in a begin/commit nesting, the transaction will rollback(). Otherwise, the method returns. This is used to cancel a Transaction without affecting the scope of an enclosing transaction. """ await greenlet_spawn(self._proxied.close) async def rollback(self) -> None: """Roll back this :class:`.AsyncTransaction`.""" await greenlet_spawn(self._proxied.rollback) async def commit(self) -> None: """Commit this :class:`.AsyncTransaction`.""" await greenlet_spawn(self._proxied.commit) async def start(self, is_ctxmanager: bool = False) -> AsyncTransaction: """Start this :class:`_asyncio.AsyncTransaction` object's context outside of using a Python ``with:`` block. """ self.sync_transaction = self._assign_proxied( await greenlet_spawn( self.connection._proxied.begin_nested if self.nested else self.connection._proxied.begin ) ) if is_ctxmanager: self.sync_transaction.__enter__() return self async def __aexit__(self, type_: Any, value: Any, traceback: Any) -> None: await greenlet_spawn(self._proxied.__exit__, type_, value, traceback) @overload def _get_sync_engine_or_connection(async_engine: AsyncEngine) -> Engine: ... @overload def _get_sync_engine_or_connection( async_engine: AsyncConnection, ) -> Connection: ... def _get_sync_engine_or_connection( async_engine: Union[AsyncEngine, AsyncConnection] ) -> Union[Engine, Connection]: if isinstance(async_engine, AsyncConnection): return async_engine._proxied try: return async_engine.sync_engine except AttributeError as e: raise exc.ArgumentError( "AsyncEngine expected, got %r" % async_engine ) from e @inspection._inspects(AsyncConnection) def _no_insp_for_async_conn_yet(subject: AsyncConnection) -> NoReturn: raise exc.NoInspectionAvailable( "Inspection on an AsyncConnection is currently not supported. " "Please use ``run_sync`` to pass a callable where it's possible " "to call ``inspect`` on the passed connection.", code="xd3s", ) @inspection._inspects(AsyncEngine) def _no_insp_for_async_engine_xyet(subject: AsyncEngine) -> NoReturn: raise exc.NoInspectionAvailable( "Inspection on an AsyncEngine is currently not supported. " "Please obtain a connection then use ``conn.run_sync`` to pass a " "callable where it's possible to call ``inspect`` on the " "passed connection.", code="xd3s", )