FastAPI + SQLAlchemy Async: The Session Management Patterns That Actually Survive Production

Your app works perfectly in local dev. Then you deploy, traffic goes up, and you start seeing MissingGreenlet errors in Sentry, connections pile up in pg_stat_activity, and every third request seems to hang. If you're running SQLAlchemy 2.0's async engine under FastAPI, this is a rite of passage — and it's almost always caused by one of four mistakes in how sessions and connections are scoped.

This is not a "hello world" async setup guide. It's the wiring, the pool tuning, and the failure modes you need to understand before they show up as an incident.

The direct answer: what a correct setup looks like

The production-safe pattern for FastAPI + SQLAlchemy 2.0 async has four pieces, and they need to work together:

1. One AsyncEngine created once at app startup, backed by an async driver like asyncpg.
2. One async_sessionmaker bound to that engine, configured with expire_on_commit=False.
3. A yield-based FastAPI dependency that opens exactly one AsyncSession per request and guarantees it closes.
4. Explicit transaction boundaries — you decide when to commit, not the ORM.

Get these four right and most of the notorious async SQLAlchemy pitfalls simply don't happen. Get any one wrong and you'll eventually see connection leaks, MissingGreenlet exceptions, or stale data served from an expired session. Let's build it properly, then break down why each piece matters.

Installation

You need three things: SQLAlchemy 2.0+, an async Postgres driver, and asyncpg rather than psycopg2.

pip install "sqlalchemy[asyncio]>=2.0" asyncpg fastapi uvicorn

If you're managing schema migrations — and you should be — add Alembic to the mix:

pip install alembic

Alembic runs synchronously by default, which trips people up: your application uses an async engine, but Alembic's migration runner typically doesn't need to. You can either let Alembic use a sync driver (psycopg2) purely for migrations while your app uses asyncpg at runtime, or configure Alembic's env.py to run migrations through an async connection using run_sync. Either works. The important thing is not to assume your app's async engine configuration transfers automatically to Alembic — it's a separate engine object with its own URL and driver.

Models and Schemas

SQLAlchemy 2.0's declarative style uses typed Mapped columns, which pairs cleanly with FastAPI's Pydantic schemas — but don't conflate the two. Your ORM models describe database structure; your Pydantic schemas describe API shape. Keep them as separate classes.

from datetime import datetime
from sqlalchemy import String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Item(Base):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(255))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

This is SQLAlchemy 2.0's native typed ORM style — there's no reason to reach for the legacy 1.x Query API here. The async engine and Mapped columns are designed to work together.

App Settings and the engine

Keep your database URL and pool parameters in one settings object (Pydantic Settings, or plain environment variables — the mechanism doesn't matter, the discipline does). The key detail: the URL must specify the async driver explicitly.

DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/appdb"

Note the +asyncpg. Forgetting this is a common first mistake — SQLAlchemy will happily accept a bare postgresql:// URL and then fail because the default driver isn't async-capable.

Now create the engine and session factory once, at module load, not per-request:

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

engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,
    pool_recycle=1800,
    echo=False,
)

AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autoflush=False,
)

A few of these deserve explanation because getting them wrong is exactly how you end up debugging production at 2 a.m.

Connection pool configuration, explained

• pool_size — the number of persistent connections the pool keeps open. Size this against your actual concurrency and your database's max connection limit, not against guesswork. Remember: pool_size is per engine instance, and each worker process gets its own pool. If you run four workers with pool_size=10, your database sees 40 baseline connections from the app alone.
• max_overflow — how many additional connections the pool can open beyond pool_size under burst load, before requests start queuing. This is your shock absorber for traffic spikes; it's not a substitute for right-sizing pool_size.
• pool_pre_ping — issues a lightweight liveness check before handing a connection out. This matters enormously with managed Postgres, load balancers, and any infrastructure that can silently drop idle connections. Without it, you get an error on first use of a stale connection instead of a clean transparent reconnect.
• pool_recycle — the maximum age (in seconds) a connection can reach before the pool discards and replaces it, rather than reusing it. This defends against server-side or middlebox connection timeouts that would otherwise kill a connection sitting idle in your pool.

The tradeoff to internalize: a bigger pool doesn't mean better throughput. Postgres has its own connection ceiling, and every idle connection has memory and scheduling overhead on the database side. Before deploying, multiply pool_size + max_overflow by the number of worker processes you run and compare that against what your database can actually sustain — this is the single most common capacity-planning mistake teams make when they scale past a single worker.

Database Session: the dependency that actually works

This is the part generic tutorials get subtly wrong. The dependency must yield the session, and cleanup must happen in a finally block so it runs even when the request raises an exception.

from typing import AsyncGenerator
from fastapi import Depends

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

DbSession = Depends(get_db)

Wire it into an endpoint like this:

from fastapi import APIRouter

router = APIRouter()

@router.get("/items/{item_id}")
async def read_item(item_id: int, db: AsyncSession = DbSession):
    result = await db.execute(select(Item).where(Item.id == item_id))
    item = result.scalar_one_or_none()
    return item

Why session-per-request, and nothing else

The scoping rule is simple and non-negotiable: one AsyncSession per request, created by the dependency, closed by the dependency, never shared across requests or background tasks.

Reasons this matters:

• AsyncSession isn't safe for concurrent use across coroutines. If two requests share a session, you get race conditions on the underlying connection state — this is where a chunk of MissingGreenlet and "connection already in use" errors come from.
• A module-level or app-level session that outlives a request accumulates identity-map state and stale objects, which is how you end up serving one user's cached row to another.
• Background tasks and websocket handlers need their own session, opened the same way — don't reach into a request-scoped session from a BackgroundTasks callback that might run after the response has already been sent and the dependency's finally block has already closed it.

Transaction handling: who owns commit and rollback

FastAPI's dependency injection gives you a clean hook, but it doesn't decide your transaction boundaries for you — you do. Two workable patterns:

Pattern A — explicit commit in the endpoint or service layer. The dependency only manages session lifecycle (open/close); your business logic calls commit() when the unit of work is done.

@router.post("/items")
async def create_item(payload: ItemCreate, db: AsyncSession = DbSession):
    item = Item(name=payload.name)
    db.add(item)
    await db.commit()
    await db.refresh(item)
    return item

Pattern B — begin() as a context manager, which commits on clean exit and rolls back on exception automatically:

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        async with session.begin():
            yield session

Pattern B is tighter and harder to misuse — you can't forget to commit or roll back. The tradeoff is less granular control if a single request needs multiple independent transactions (rare, but it happens with multi-step import jobs). Pick one pattern per codebase and apply it consistently; mixing the two across endpoints is a reliable way to confuse whoever inherits the code, including future you.

expire_on_commit=False in the sessionmaker matters here: with the default True, every attribute on your ORM objects becomes stale after commit and triggers a fresh load on next access — which brings us to the lazy-load trap below.

The four pitfalls that generic tutorials skip

1. MissingGreenlet

This error means synchronous, blocking database code got triggered inside an async context without SQLAlchemy's async greenlet bridge active — almost always a lazy-loaded relationship accessed outside the session's async context, for example after the response has started serializing, or inside a Pydantic validator that touches a relationship attribute. Fix: eager-load what you need with selectinload() or joinedload() inside the same await db.execute(...) call, before the session closes. According to SQLAlchemy async patterns, selectinload() is best for one-to-many relationships (e.g., User → Posts) and runs two separate queries, while joinedload() is best for many-to-one relationships (e.g., Post → Author) and runs a single query with a SQL JOIN — choose based on your relationship cardinality rather than relying on implicit lazy loading.

2. Lazy-load and expire-on-commit traps

Even without hitting MissingGreenlet, expire_on_commit=True (the SQLAlchemy default) means every object attribute is marked stale after commit(). The next access issues an implicit query — which, in an async context, is exactly the kind of implicit I/O that causes the errors above. Setting expire_on_commit=False on your async_sessionmaker and explicitly calling await db.refresh(obj) when you actually need fresh data is the safer default for async work.

3. Connection leaks

If a session is opened but an exception path skips the close() — because someone wrote a dependency without a finally, or a background task grabbed a session and never returned it — connections accumulate until the pool exhausts and every subsequent request blocks waiting for one. The finally: await session.close() shown above is not optional boilerplate; it's the leak fix.

4. Mixing sync and async database calls

Reaching for a synchronous SQLAlchemy engine or a blocking driver call (like psycopg2) inside an async def endpoint blocks the event loop for every concurrent request being served by that worker — not just the one making the call. Keep the entire data-access path async, end to end: async engine, async session, await on every execute/commit/refresh. If you must call sync code (a legacy library, a CPU-bound task), push it to a thread pool explicitly rather than letting it silently block the loop.

Testing async sessions with pytest

Async database tests need an event loop and a session scoped correctly for the test lifecycle — usually a fresh transaction per test that rolls back at the end, so tests don't pollute each other.

import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

@pytest_asyncio.fixture
async def db_session():
    engine = create_async_engine(TEST_DATABASE_URL)
    TestSessionLocal = async_sessionmaker(bind=engine, expire_on_commit=False)
    async with TestSessionLocal() as session:
        async with session.begin():
            yield session
            await session.rollback()
    await engine.dispose()

Override the get_db dependency in your FastAPI app with this fixture via app.dependency_overrides, and every test gets an isolated transaction against a real Postgres instance rather than a mocked-out ORM. Requires pytest-asyncio installed and configured (either asyncio_mode = "auto" in your pytest config, or explicit @pytest.mark.asyncio markers).

Quick reference: pool parameters at a glance

The end

The async engine and AsyncSession in SQLAlchemy 2.0 are solid, production-ready tools — the failures people hit almost never come from the library itself, they come from scoping sessions wrong, trusting implicit lazy loads in an async context, or letting sync code sneak into an async path. Get the four pieces in the direct-answer section right — engine, sessionmaker, yield-based dependency, explicit transaction ownership — and the pitfalls above stop being production incidents and start being things you recognize on sight.

FAQ

Do I need asyncpg specifically, or will any driver work? You need a driver that supports Python's async I/O model. For Postgres, asyncpg is the common choice paired with SQLAlchemy's async engine; a plain synchronous driver won't work with create_async_engine.

Should expire_on_commit be True or False for FastAPI apps? False is the safer default for async APIs. It avoids triggering implicit lazy-load queries after commit, which is a frequent source of MissingGreenlet errors. Call await session.refresh(obj) explicitly on the rare occasions you need guaranteed-fresh attributes.

Can I share one AsyncSession across multiple requests for performance? No. Session-per-request is the scoping rule for a reason — AsyncSession isn't designed for concurrent cross-request use, and sharing one leads to race conditions and hard-to-reproduce state bugs.

How do I size pool_size and max_overflow for real traffic? Start from your database's connection ceiling, divide by the number of application worker processes you run, and leave headroom for other services hitting the same database. There's no universal number — it's a function of your specific deployment topology, not a constant you can copy from a tutorial.

Does Alembic need to use the async engine too? Not necessarily. Many setups run Alembic migrations with a synchronous driver purely for the migration process, while the application itself uses the async engine at runtime. Both are legitimate; just don't assume one engine configuration serves both purposes automatically.