prompts/docs/skills/async-fastapi-sqlmodel/references/session.md

Async SQLAlchemy Session Management

!!! info "Primary sources" - Python functools.cache - Python asynccontextmanager - SQLAlchemy asyncio extension - SQLAlchemy session basics - FastAPI dependencies with yield

---

Purpose

Define one canonical session model for FastAPI + SQLAlchemy asyncio:

• configure one shared session factory,
• create one AsyncSession per request or per unit-of-work,
• never share one AsyncSession across concurrent tasks.

---

Scope and Non-Goals

• In scope: session factory creation, FastAPI dependency wiring, request/task scoping, transaction demarcation.
• Out of scope: ORM model design, query optimization strategy, schema migration tooling.

---

Rules

• Create one cached async_sessionmaker per app-owned AsyncEngine.
• Let repositories resolve the cached maker by database URL, with an injectable factory override for tests.
• Use a fresh AsyncSession for each request or explicit unit-of-work.
• Pass an AsyncSession directly to data-access functions.
• Borrow a caller-provided session without closing or committing it.
• 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.

---

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.

Cache it by the application-owned engine so repeated composition calls return the same maker:

from functools import cache

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from .engine import dispose_engine
from .engine import get_engine


@cache
def get_session_factory(database_url: str) -> async_sessionmaker[AsyncSession]:
    return async_sessionmaker(
        bind=get_engine(database_url),
        class_=AsyncSession,
        expire_on_commit=False,
    )


async def dispose_session_factory(database_url: str) -> None:
    get_session_factory.cache_clear()
    await dispose_engine(database_url)

functools.cache caches by argument equality and requires hashable arguments. The database URL is an explicit string key shared with the cached engine factory. The cache retains the returned maker until get_session_factory.cache_clear() runs. Cache the synchronous maker function, never an async function and never a produced AsyncSession.

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:

async with session_factory() as session:
    ...

The factory can be shared across requests and tasks. Sessions produced by it cannot be shared across concurrent tasks.

An async_sessionmaker has no connection pool or async dispose() method of its own. dispose_session_factory() means "invalidate the cached maker, then dispose its engine." Clearing the maker first ensures no subsequent composition call can retrieve a maker bound to the engine being shut down.

Use the helper when shutting down or replacing the database resources:

await dispose_session_factory(database_url)

Otherwise, a later call can return a maker that still references the old engine object. This matters in lifespan tests, application restarts within one process, and test suites that replace engines.

---

Optional Session Ownership

A small asynccontextmanager can make repository methods composable. It borrows an existing session when supplied; otherwise it creates and closes one from a supplied factory:

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager


@asynccontextmanager
async def session_scope(
    *,
    database_url: str,
    session: AsyncSession | None = None,
    session_factory: async_sessionmaker[AsyncSession] | None = None,
) -> AsyncIterator[AsyncSession]:
    if session is not None:
        yield session
        return

    active_factory = session_factory or get_session_factory(database_url)

    async with active_factory() as owned_session:
        yield owned_session

The branch is intentionally explicit. Python's nullcontext can express the same borrow-or-own idea, but the branch keeps ownership and typing obvious.

This helper manages session lifetime only:

• It does not close, commit, or roll back a supplied session; the caller owns it.
• It closes a session that it creates. Closing releases resources and rolls back an unfinished transaction; it does not commit.
• It does not start a transaction. Put session.begin() at the use-case boundary.
• A supplied session wins; neither the factory override nor cached factory is used.
• A supplied factory overrides cached resolution, which keeps tests and specialized wiring explicit.
• Otherwise, database_url selects the cached factory returned by get_session_factory().

Do not turn this into an implicit unit-of-work helper that sometimes commits. Whether work joins an existing transaction or creates a new one must remain visible to the caller.

---

Repository and Function Boundaries

Pass the database URL to repository constructors. The repository stores repeatable database configuration, not mutable session state, and session_scope() resolves the cached factory when a standalone operation needs a session. An optional factory override keeps tests independent:

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker


async def find_item(session: AsyncSession, item_id: int) -> Item | None:
    statement = select(Item).where(Item.id == item_id)
    return await session.scalar(statement)


class ItemRepository:
    def __init__(
        self,
        database_url: str,
        session_factory: async_sessionmaker[AsyncSession] | None = None,
    ) -> None:
        self.database_url = database_url
        self.session_factory = session_factory

    async def find(
        self,
        item_id: int,
        *,
        session: AsyncSession | None = None,
    ) -> Item | None:
        async with session_scope(
            database_url=self.database_url,
            session=session,
            session_factory=self.session_factory,
        ) as active_session:
            return await find_item(active_session, item_id)

This split gives each layer one job:

• The repository object identifies its database configuration and creates a session only for a standalone call.
• Production calls reuse the cached factory; tests can inject a factory override.
• A caller can pass a session to join an existing unit of work; the repository borrows it.
• The access function owns only the query and requires an existing AsyncSession.
• Application wiring supplies the production factory.
• Tests can supply a factory bound to a test engine or call find_item() with a transaction-scoped test session.

When several repository operations must share one transaction, pass the same session through each call. Put the transaction at the use-case boundary:

async with session_factory() as session:
    async with session.begin():
        item = await repository.find(item_id, session=session)
        await update_item(session, item, changes)

This preserves atomicity without making repository objects hold mutable AsyncSession instances across calls.

---

Canonical FastAPI Dependency Pattern

from collections.abc import AsyncIterator

from fastapi import Depends
from fastapi import Request
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import async_sessionmaker


type SessionFactory = async_sessionmaker[AsyncSession]


def resolve_session_factory(request: Request) -> SessionFactory:
    return get_session_factory(request.app.state.settings.database_url)


async def get_db_session(
    session_factory: SessionFactory = Depends(resolve_session_factory),
) -> AsyncIterator[AsyncSession]:
    async with session_factory() as session:
        yield session

Route usage:

from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession

from .session import get_db_session

router = APIRouter()


@router.post("/items")
async def create_item(session: AsyncSession = Depends(get_db_session)) -> dict:
    async with session.begin():
        # write operations here
        ...
    return {"status": "ok"}

---

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() 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 request/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 async dependency.

---

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 requests.
• Sharing one AsyncSession across parallel tasks.
• Passing an application-global AsyncSession to a repository constructor.
• Caching an AsyncSession instead of caching async_sessionmaker.
• Leaving a cached maker pointing at a disposed or replaced engine.
• Calling the session factory inside low-level access functions such as find_item().
• Hidden session creation in lower access functions with no caller control.
• Closing or committing a session supplied by the caller.
• Starting a new transaction inside a helper that may receive a session already in a transaction.
• Mixing commit/rollback ownership across layers without a declared boundary.

---

Operational Checks

• Exactly one cached async_sessionmaker exists per application engine.
• Session factory caches are cleared before their engines are disposed or replaced.
• Request handlers receive sessions from one canonical dependency.
• No code path creates AsyncSession in module import side effects.
• Background jobs and API handlers each create task-local sessions.

---

Testing Checks

• Repository constructors accept a test session factory without FastAPI startup.
• Session-taking access functions accept a transaction-scoped test session directly.
• Optional-session tests verify that borrowed sessions remain open and created sessions close.
• Optional-session tests verify that neither path commits implicitly.
• Cache tests clear get_session_factory before and after replacing engines.
• Dependency override exists for the FastAPI session factory.
• Rollback behavior is verified for failed write units.
• Parallel-task tests verify no shared AsyncSession instances.
• Lifespan tests confirm session factory is initialized and teardown-safe.