# Async SQLAlchemy Session Management !!! info "Primary sources" - [Python `asynccontextmanager`](https://docs.python.org/3/library/contextlib.html#contextlib.asynccontextmanager) - [Python `functools.cache`](https://docs.python.org/3/library/functools.html#functools.cache) - [Python `inspect.signature`](https://docs.python.org/3/library/inspect.html#inspect.signature) - [SQLAlchemy asyncio extension](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html) - [SQLAlchemy session basics](https://docs.sqlalchemy.org/en/21/orm/session_basics.html) - [`nicegui-db` session implementation](https://forgejo.john-stream.com/john/nicegui-db/src/commit/126bc26ad8635a86bacf684d7bda409230347597/src/nicegui_db/db/session.py) --- ## Purpose Define one canonical session model for SQLAlchemy asyncio: - configure a lifespan-owned factory or resolve a URL-keyed cached factory, - create one AsyncSession per task or unit of work, - let callers supply a session when they already own the scope, - never share one AsyncSession across concurrent tasks. --- ## Scope and Non-Goals - In scope: session factory creation, task scoping, and transaction demarcation. - Out of scope: framework dependency wiring, ORM model design, query optimization strategy, and schema migration tooling. --- ## Rules - Create the application `async_sessionmaker` inside `database_scope()` and store it in application state for request dependencies. - Use `get_session_factory(db_url)` and `resolve_session_factory()` for standalone decorated operations that do not receive the application factory. - Use a fresh AsyncSession for each task or explicit unit of work. - Let reusable service functions accept `AsyncSession | None` and apply `@with_session` when standalone invocation is useful. - Pass an `AsyncSession` directly when composing several calls under one caller-owned scope. - Borrow a caller-provided session without beginning, closing, committing, or rolling it back. - Do not share AsyncSession across `asyncio.gather()` or parallel tasks. - Prefer direct dependency injection over global scoped-session patterns in new code. - Use explicit transaction boundaries (`async with session.begin():`) for writes. - Use `db_transaction_scope()` when a standalone operation must own engine, factory, session, and transaction lifetimes together. - Use `begin_nested()` directly and only when partial rollback through a database SAVEPOINT is required. --- ## Sessions and Transactions A session and a transaction solve related but different problems: | Concept | Responsibility | Typical lifetime | | --- | --- | --- | | `AsyncSession` | Provides the ORM workspace: executes queries, tracks loaded and changed objects in its identity map, and flushes pending changes. It also coordinates access to a database connection. | One task or explicit unit of work. | | Transaction | Defines the atomic database boundary: all work inside it commits together on success or rolls back together on failure. | One complete operation that must have a single outcome. | A transaction belongs to a session; it is not an alternative to one. The session is the interface used by application and data-access code, while the transaction determines when that work becomes permanent. A session may coordinate sequential transactions during its lifetime, although short-lived application scopes commonly use one session for one transaction. Use a session without a helper-owned commit boundary for independent reads or lower-level functions that must participate in whatever transaction their caller controls: ```python async with session_factory() as session: item = await find_item(session, item_id) ``` Use an explicit transaction for writes, read-modify-write operations, or several statements that must succeed or fail as one unit: ```python async with session_factory.begin() as session: order = await create_order(session, order_data) await reserve_inventory(session, order) ``` SQLAlchemy sessions use [autobegin](https://docs.sqlalchemy.org/en/21/orm/session_basics.html#auto-begin), so the first database operation normally starts a transaction even for a read. Therefore, “session-only” means that the surrounding helper owns only session lifetime and does not promise to commit; it does not mean that no database transaction exists. Closing such a session releases its resources and rolls back any unfinished transaction. An explicit `begin()` is valuable when application code must make the atomic boundary and commit ownership visible. For most read-only operations, a session context is sufficient. Use an explicit transaction for reads when they need a defined consistency boundary, participate in a larger atomic operation, or use locking such as `SELECT ... FOR UPDATE`. --- ## Session Factory Mechanics An `async_sessionmaker[AsyncSession]` is a reusable configuration object and callable session producer. It stores how sessions should be created, including the engine binding and options such as `expire_on_commit=False`. It is not itself a session, connection, or transaction, and calling it does not make a shared global `AsyncSession`. The template exposes two construction paths with the same session options. The application-owned path creates a factory inside the engine lifecycle: ```python from collections.abc import AsyncGenerator from contextlib import asynccontextmanager from sqlalchemy.ext.asyncio import async_sessionmaker from sqlmodel.ext.asyncio.session import AsyncSession from .engine import engine_scope type SessionFactory = async_sessionmaker[AsyncSession] @asynccontextmanager async def database_scope( db_url: str, *, auto_flush: bool = True, ) -> AsyncGenerator[SessionFactory]: async with engine_scope(db_url) as engine: yield async_sessionmaker( bind=engine, class_=AsyncSession, expire_on_commit=False, autoflush=auto_flush, ) ``` FastAPI lifespan enters this path once and stores the yielded factory on application state. The factory must not outlive the scope because its bound engine is disposed on exit. The standalone path caches a factory by URL and `auto_flush` policy: ```python from functools import cache from .engine import get_engine @cache def get_session_factory( db_url: str, *, auto_flush: bool = True, ) -> SessionFactory: return async_sessionmaker( bind=get_engine(db_url), class_=AsyncSession, expire_on_commit=False, autoflush=auto_flush, ) def resolve_session_factory(settings: Settings | None = None) -> SessionFactory: settings = settings or get_settings() db_url = get_database_url(settings) return get_session_factory(db_url) ``` This path lets framework-independent helpers resolve one stable factory without receiving it through every call. The tradeoff is hidden configuration resolution and a second lifecycle mechanism. `dispose_engine()` clears `get_engine`'s cache but does not clear `get_session_factory`'s cache in the template. A cached factory remains bound to the disposed engine object; SQLAlchemy can create a new pool when that engine is used again, but a later `database_scope()` for the same URL can own a different engine. Treat cached standalone resolution as process-lifetime convenience, avoid repeated application lifecycles in one process, and clear both caches together if the template evolves to support them. Each call to `session_factory()` creates a distinct `AsyncSession`. The caller that invokes the factory owns that session lifetime and must close it, normally with `async with`: ```python async with session_factory() as session: ... ``` The factory can be shared across operations and tasks. Sessions produced by it cannot be shared across concurrent tasks. Passing the application factory directly has three useful consequences: - Lower layers do not resolve settings or global resources. - Tests can inject a test factory directly through `session_scope(session_factory=...)` or FastAPI state. - Transaction ownership remains independent of engine construction. --- ## Database and Convenience Scopes The template provides three framework-independent context managers: ```python from collections.abc import AsyncGenerator from contextlib import asynccontextmanager @asynccontextmanager async def db_session_scope( db_url: str | None = None, ) -> AsyncGenerator[AsyncSession]: db_url = db_url or resolve_database_url() async with database_scope(db_url) as session_factory, session_factory() as session: yield session @asynccontextmanager async def db_transaction_scope( db_url: str | None = None, ) -> AsyncGenerator[AsyncSession]: db_url = db_url or resolve_database_url() async with database_scope(db_url) as session_factory, session_factory.begin() as session: yield session @asynccontextmanager async def session_scope( *, settings: Settings | None = None, session_factory: SessionFactory | None = None, session: AsyncSession | None = None, ) -> AsyncGenerator[AsyncSession]: if session is not None: yield session return session_factory = session_factory or resolve_session_factory(settings=settings) async with session_factory() as owned_session: yield owned_session ``` `db_session_scope()` owns a complete temporary database lifecycle and a session but does not commit. `db_transaction_scope()` owns the same resources plus a root transaction that commits on successful exit and rolls back on exception. Both initialize the schema by default because `database_scope()` enters `engine_scope()` with its default `initialize=True`. They are appropriate for scripts, commands, and isolated operations, not per-request use inside an already-running application. `session_scope()` is the borrow-or-create helper. Its precedence is supplied session, supplied factory, then settings-based cached factory resolution. A supplied session remains entirely caller-owned; the helper does not require an active transaction and does not begin, commit, roll back, or close it. An owned session is closed on exit, and unfinished autobegun work rolls back. Passing `session=None` is the same as omitting the session for `session_scope()` and therefore creates a session. This differs from `with_session`, which tests whether the argument name was bound rather than whether its value is non-null. ## Signature-Aware Session Injection `with_session` allows one async function to support standalone calls and explicit composition: ```python from collections.abc import Awaitable from collections.abc import Callable from functools import wraps from inspect import signature def with_session[**P, R]( func: Callable[P, Awaitable[R]], ) -> Callable[P, Awaitable[R]]: sig = signature(func) @wraps(func) async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: bound = sig.bind_partial(*args, **kwargs) if "session" in bound.arguments: return await func(*args, **kwargs) async with resolve_session_factory()() as session: bound.arguments["session"] = session return await func(*bound.args, **bound.kwargs) return wrapper ``` The function must be async and expose a parameter named exactly `session`. The decorator preserves metadata with `wraps()`, binds positional and keyword arguments through the original signature, and injects a fresh session only when the caller omitted that argument. The distinction between omitted and explicit `None` is deliberate in the implementation: - `await operation()` injects and owns a session. - `await operation(session=existing_session)` borrows the caller's session. - `await operation(None)` or `await operation(session=None)` forwards `None` without injection. The decorated function therefore types the parameter as `AsyncSession | None = None` but should assert or guard after decoration. Explicit `None` is not a request for injection. This preserves ordinary Python call binding, but it means wrappers or callers must omit the argument instead of forwarding a nullable value. `with_session` owns session lifetime only. It does not begin or commit a transaction, so it is naturally suited to reads. Decorated writes must either manage a visible transaction or be called with a session from `db_transaction_scope()` or another caller-owned transaction. Prefer explicit factory or session injection when lifecycle transparency and test substitution matter more than call-site convenience. --- ## Function and Service Boundaries Template service functions support both standalone and composed use by combining `@with_session` with an optional parameter: ```python from sqlmodel import func from sqlmodel import select @with_session async def count_items(session: AsyncSession | None = None) -> int: assert session is not None, "Session must be provided by with_session decorator" result = await session.exec(select(func.count()).select_from(Item)) return result.one() ``` The standalone call injects and closes a session: ```python count = await count_items() ``` A larger use case passes one caller-owned session through several decorated functions: ```python async with session_factory.begin() as session: count = await count_items(session) await create_item(payload, session=session) ``` The decorator sees the bound `session` argument and leaves all ownership with the caller. It never creates a SAVEPOINT or nested transaction. For low-level helpers that should never resolve settings, require a non-optional session and leave them undecorated. Application service objects may store the immutable session factory, but they must not store a mutable session: ```python class ItemService: def __init__(self, session_factory: SessionFactory) -> None: self.session_factory = session_factory async def find(self, item_id: int) -> Item | None: async with self.session_factory() as session: return await find_item(session, item_id) ``` Code that already owns a transaction should call the session-required function directly. Repositories should normally remain in that session-required layer; the service or use-case boundary owns standalone session creation. This avoids optional-session APIs spreading into every data-access function. --- ## SAVEPOINTs and Partial Failure Use [`begin_nested()`](https://docs.sqlalchemy.org/en/21/orm/session_transaction.html#using-savepoint) only when failure inside one portion of an operation should roll back that portion while preserving the outer transaction: ```python async with db_transaction_scope() as session: order = await insert_order(session, payload) try: async with session.begin_nested(): await apply_optional_discount(session, order) except DiscountError: pass await reserve_inventory(session, order) ``` Important SAVEPOINT semantics: - `begin_nested()` starts a root transaction if one is not already active, so call it inside a visible outer transaction when that ownership matters. - Entering `begin_nested()` unconditionally flushes pending session state, regardless of the `autoflush` setting. - Successful exit releases the SAVEPOINT; it does not commit the outer transaction. - Exceptional exit rolls back to the SAVEPOINT and leaves the outer transaction active. - In SQLAlchemy 2.x, `session.commit()` commits the outermost transaction. Never call it to release a SAVEPOINT; let the nested context manager manage its transaction handle. Do not create a SAVEPOINT merely because one service calls another. SAVEPOINTs add database work and alter flush and error-recovery behavior. Use them only for explicit partial-failure requirements such as skipping one conflicting row while retaining the rest of a batch. --- ## Framework Integration Keep framework adapters outside these session primitives. See [FastAPI database integration](fastapi.md) for lifespan ownership, `Annotated` dependency aliases, and read-versus-write request sessions. --- ## Configuration Guidance - `expire_on_commit=False` is commonly preferred in asyncio applications to reduce accidental post-commit reload behavior. - `AsyncSession.refresh()` is preferred over broad expiration patterns when state refresh is needed. - [`async_sessionmaker.begin()`](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html#sqlalchemy.ext.asyncio.async_sessionmaker.begin) is a concise option when one scope must create a session, begin a transaction, commit on success, roll back on failure, and close. Do not use it when borrowing a caller's session. ## SQLModel Alignment - Use SQLModel as the default model and statement layer while keeping the same session ownership model: one `async_sessionmaker`, one `AsyncSession` per task or unit of work. - SQLModel does not replace SQLAlchemy async lifecycle primitives; it provides model declaration, validation, and typing ergonomics on top of them. - Do not mix ad hoc session construction with the canonical session factory. --- ## Concurrency Rules - One session per concurrent task. - If work fans out into parallel tasks, each task receives its own AsyncSession. - Pass sessions explicitly to service functions; avoid mutable global session state. --- ## Anti-Patterns - A singleton/global AsyncSession reused across tasks or operations. - Sharing one AsyncSession across parallel tasks. - Passing an application-global AsyncSession to a repository constructor. - Creating a new `async_sessionmaker` in each operation. - Retaining a session factory after its bound engine scope exits. - Using cached standalone factory resolution when the application factory is already available. - Assuming `with_session` starts or commits a transaction. - Forwarding `session=None` to a decorated function when injection was intended. - Closing or committing a session supplied by the caller. - Silently starting or committing a transaction on a supplied session. - Creating a SAVEPOINT for ordinary nested service calls. - Hiding root transaction, joined transaction, and SAVEPOINT behavior behind one mode-driven `atomic_scope()` helper. - Calling `session.commit()` inside a SAVEPOINT scope. - Mixing commit/rollback ownership across layers without a declared boundary. --- ## Operational Checks - The FastAPI application factory is created inside `database_scope()` and does not outlive its bound engine. - Cached standalone factories are used only where application-state injection is unavailable. - `session_scope()` precedence is supplied session, supplied factory, then settings-based resolution. - Decorated functions receive injection only when the `session` argument is omitted. - No code path creates AsyncSession in module import side effects. - Concurrent jobs and operations each create task-local sessions. --- ## Testing Checks - Service constructors accept a test session factory without framework startup. - Session-taking access functions accept a transaction-scoped test session directly. - `session_scope()` tests cover supplied-session, supplied-factory, and settings-resolution precedence. - `db_transaction_scope()` tests verify commit on success, rollback on failure, session closure, engine disposal, and cache cleanup. - `with_session` tests cover omitted, positional, keyword, and explicit-`None` session arguments. - Composition tests verify decorated service calls borrow one caller-owned session without committing it. - SAVEPOINT tests verify local rollback preserves the outer transaction and successful exit does not commit it. - Tests that depend on SAVEPOINT timing account for `begin_nested()` flushing pending state on entry. - Rollback behavior is verified for failed write units. - Parallel-task tests verify no shared AsyncSession instances. - Lifecycle tests confirm schema initialization, factory availability, deterministic teardown, and expected cache behavior.