Class scoped_session

lib/sqlalchemy/orm/scoping.py:153–2229

Provides scoped management of :class:`.Session` objects. See :ref:`unitofwork_contextual` for a tutorial. .. note:: When using :ref:`asyncio_toplevel`, the async-compatible :class:`_asyncio.async_scoped_session` class should be used in place of :class:`.scoped_session

    ],
)
class scoped_session(Generic[_S]):
    class="st">"""Provides scoped management of :class:`.Session` objects.

    See :ref:`unitofwork_contextual` for a tutorial.

    .. note::

       When using :ref:`asyncio_toplevel`, the async-compatible
       :class:`_asyncio.async_scoped_session` class should be
       used in place of :class:`.scoped_session`.

    class="st">"""

    _support_async: bool = False

    session_factory: sessionmaker[_S]
    class="st">"""The `session_factory` provided to `__init__` is stored in this
    attribute and may be accessed at a later time.  This can be useful when
    a new non-scoped :class:`.Session` is needed.class="st">"""

    registry: ScopedRegistry[_S]

    def __init__(
        self,
        session_factory: sessionmaker[_S],
        scopefunc: Optional[Callable[[], Any]] = None,
    ):
        class="st">"""Construct a new :class:`.scoped_session`.

        :param session_factory: a factory to create new :class:`.Session`
         instances. This is usually, but not necessarily, an instance
         of :class:`.sessionmaker`.
        :param scopefunc: optional function which defines
         the current scope.   If not passed, the :class:`.scoped_session`
         object assumes class="st">"thread-local" scope, and will use
         a Python ``threading.local()`` in order to maintain the current
         :class:`.Session`.  If passed, the function should return
         a hashable token; this token will be used as the key in a
         dictionary in order to store and retrieve the current
         :class:`.Session`.

        class="st">"""
        self.session_factory = session_factory

        if scopefunc:
            self.registry = ScopedRegistry(session_factory, scopefunc)
        else:
            self.registry = ThreadLocalRegistry(session_factory)

    @property
    def _proxied(self) -> _S:
        return self.registry()

    def __call__(self, **kw: Any) -> _S:
        rclass="st">"""Return the current :class:`.Session`, creating it
        using the :attr:`.scoped_session.session_factory` if not present.

        :param \**kw: Keyword arguments will be passed to the