Class AsyncSession

lib/sqlalchemy/ext/asyncio/session.py:210–1664

Asyncio version of :class:`_orm.Session`. The :class:`_asyncio.AsyncSession` is a proxy for a traditional :class:`_orm.Session` instance. The :class:`_asyncio.AsyncSession` is **not safe for use in concurrent tasks.**. See :ref:`session_faq_threadsafe` for background. .. vers

    ],
)
class AsyncSession(ReversibleProxy[Session]):
    """Asyncio version of :class:`_orm.Session`.

    The :class:`_asyncio.AsyncSession` is a proxy for a traditional
    :class:`_orm.Session` instance.

    The :class:`_asyncio.AsyncSession` is **not safe for use in concurrent
    tasks.**.  See :ref:`session_faq_threadsafe` for background.

    .. versionadded:: 1.4

    To use an :class:`_asyncio.AsyncSession` with custom :class:`_orm.Session`
    implementations, see the
    :paramref:`_asyncio.AsyncSession.sync_session_class` parameter.


    """

    _is_asyncio = True

    dispatch: dispatcher[Session]

    def __init__(
        self,
        bind: Optional[_AsyncSessionBind] = None,
        *,
        binds: Optional[Dict[_SessionBindKey, _AsyncSessionBind]] = None,
        sync_session_class: Optional[Type[Session]] = None,
        **kw: Any,
    ):
        r"""Construct a new :class:`_asyncio.AsyncSession`.

        All parameters other than ``sync_session_class`` are passed to the
        ``sync_session_class`` callable directly to instantiate a new
        :class:`_orm.Session`. Refer to :meth:`_orm.Session.__init__` for
        parameter documentation.

        :param sync_session_class:
          A :class:`_orm.Session` subclass or other callable which will be used
          to construct the :class:`_orm.Session` which will be proxied. This
          parameter may be used to provide custom :class:`_orm.Session`
          subclasses. Defaults to the
          :attr:`_asyncio.AsyncSession.sync_session_class` class-level
          attribute.

          .. versionadded:: 1.4.24

        """
        sync_bind = sync_binds = None

        if bind:
            self.bind = bind
            sync_bind = engine._get_sync_engine_or_connection(bind)

        if binds:
            self.binds = binds
            sync_binds = {
                key: engine._get_sync_engine_or_connection(b)