Async SQLAlchemy Session per Request in FastAPI

Key takeaways:

  • Build the async engine and async_sessionmaker once in lifespan; they own the pool.
  • Provide a fresh session per request through a yield dependency.
  • Commit on success and roll back on error inside the dependency.
  • Alias it as SessionDep so handlers stay clean and typed.
  • Override the dependency in tests for full isolation.

This guide is the concrete setup behind Async Database Sessions, and it applies the dependency injection yield pattern to databases.

The Problem This Solves

Sharing a session across requests corrupts transaction state, and opening one without disciplined cleanup leaks connections. A request-scoped session provided by a yield dependency gives each request its own isolated, automatically-closed transaction.

Prerequisites

  • SQLAlchemy async with an async driver such as asyncpg.
  • A FastAPI app using a lifespan and dependency injection.

Step-by-Step Implementation

1. Engine and factory in lifespan

from contextlib import asynccontextmanager

from fastapi import FastAPI
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine


@asynccontextmanager
async def lifespan(app: FastAPI):
    engine = create_async_engine(app.state.settings.database_url,
                                 pool_size=10, max_overflow=5)
    # expire_on_commit=False keeps attributes usable after commit during serialization.
    app.state.session_factory = async_sessionmaker(engine, expire_on_commit=False)
    yield
    await engine.dispose()

2. The session dependency

from collections.abc import AsyncGenerator
from typing import Annotated

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


async def get_session(request: Request) -> AsyncGenerator[AsyncSession, None]:
    async with request.app.state.session_factory() as session:
        try:
            yield session
            await session.commit()      # Commit once on success.
        except Exception:
            await session.rollback()    # Roll back on any error.
            raise


SessionDep = Annotated[AsyncSession, Depends(get_session)]

3. Use it in a handler

from fastapi import APIRouter

router = APIRouter()


@router.get("/users/{user_id}")
async def get_user(user_id: int, session: SessionDep) -> dict:
    user = await session.get(User, user_id)
    return {"id": user.id, "email": user.email}

4. Override it in tests

@pytest.fixture
def client(app, test_session_factory):
    async def _test_session():
        async with test_session_factory() as s:
            yield s
    app.dependency_overrides[get_session] = _test_session
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

Edge Cases and Gotchas

  • expire_on_commit. Leave it False so model attributes remain accessible during response serialization after commit.
  • Nested transactions. For multi-step units of work, use explicit session.begin() blocks rather than relying solely on the dependency's commit.
  • Background jobs. Workers create their own session factory; do not reuse the request session outside the request.

Verification

async def test_rolls_back_on_error(session):
    with pytest.raises(ValueError):
        async with session.begin():
            session.add(User(email="x@y.z"))
            raise ValueError("boom")
    assert await session.scalar(select(func.count()).select_from(User)) == 0