~
Async SQLAlchemy 2.0 in production
all writing
·
  • #python
  • #sqlalchemy
  • #fastapi
  • #asyncio
  • #postgresql

Async SQLAlchemy 2.0 in Production: What the Docs Don't Tell You

Hard-won lessons from running async SQLAlchemy 2.0 in production, from the MissingGreenlet panic to pool sizing math and the lazy loading trap.

There’s a rite of passage for every Python developer who decides to go async with SQLAlchemy. It goes like this: you set up your async engine, write a few queries, watch them work beautifully in development, deploy to staging, and then at 2 AM your error tracker lights up with a wall of MissingGreenlet exceptions. You stare at the traceback, confused, because the code was working. You grep the docs. You grep Stack Overflow. You pour another coffee.

I’ve been through this cycle more than once, running FastAPI with async SQLAlchemy 2.0 and asyncpg against Postgres under real production load. The async ORM works well when you understand its mental model. The problem is that the mental model has sharp edges that the documentation glosses over, and they all show up in production under load.

This post is the guide I wish I’d had before my first deployment.

The Mental Model: One Session, One Task

The most important thing to internalize about async SQLAlchemy is that an AsyncSession is not thread-safe and is not safe to share between concurrent tasks. Each asyncio task (or coroutine that runs concurrently) needs its own session instance. If two coroutines share a session and issue queries concurrently, you’ll get MissingGreenlet errors or, worse, silently corrupted query results.

This is different from how many developers think about database sessions. In synchronous SQLAlchemy, a session is bound to a thread, and most web frameworks give you one thread per request. The mapping is implicit and hard to mess up. In async code, you can have hundreds of concurrent coroutines on a single thread, all potentially reaching for the same session if you’re not careful.

The FastAPI Session-Per-Request Pattern

The cleanest way to handle this in FastAPI is a dependency that yields a fresh session per request:

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

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost:5432/mydb",
    pool_size=10,
    max_overflow=5,
    pool_pre_ping=True,
)

# async_sessionmaker replaces the old sessionmaker(class_=AsyncSession)
SessionFactory = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncSession:
    async with SessionFactory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
from fastapi import Depends

@app.get("/tenants/{tenant_id}/documents")
async def list_documents(
    tenant_id: str,
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(Document)
        .where(Document.tenant_id == tenant_id)
        .options(selectinload(Document.tags))
        .limit(50)
    )
    return result.scalars().all()

A few things to note here. The async with SessionFactory() context manager ensures the session is closed when the request finishes, even if an exception is raised. The try/except commits on success and rolls back on failure. And expire_on_commit=False is set on the factory, which I’ll explain shortly.

This pattern gives you one session per HTTP request, properly scoped and automatically cleaned up. Don’t try to be clever by sharing sessions across requests or by creating a global session. That road leads to MissingGreenlet at 2 AM.

Lazy Loading Is a Trap

In synchronous SQLAlchemy, lazy loading is the default and it mostly just works. You access document.tags and SQLAlchemy transparently fires a query to load the related objects. It’s convenient, if occasionally wasteful.

In async SQLAlchemy, lazy loading doesn’t work. At all. Accessing an unloaded relationship raises MissingGreenlet because SQLAlchemy can’t issue a synchronous database call from within an async context. This is the single most common source of bugs I’ve seen in async SQLAlchemy codebases.

Eager Loading Strategies

The fix is to explicitly load everything you need before you leave the session context. SQLAlchemy gives you several options:

from sqlalchemy.orm import selectinload, joinedload, subqueryload

# selectinload: issues a second query with an IN clause
# Good for one-to-many and many-to-many relationships
stmt = (
    select(Document)
    .where(Document.tenant_id == tenant_id)
    .options(selectinload(Document.tags))
)

# joinedload: adds a JOIN to the original query
# Good for many-to-one and one-to-one relationships
stmt = (
    select(Document)
    .where(Document.tenant_id == tenant_id)
    .options(joinedload(Document.author))
)

# Nested eager loading for deep relationships
stmt = (
    select(Document)
    .where(Document.tenant_id == tenant_id)
    .options(
        selectinload(Document.tags),
        joinedload(Document.author).selectinload(User.roles),
    )
)

My rule of thumb: use joinedload for single-object relationships (many-to-one), use selectinload for collections (one-to-many, many-to-many). Avoid subqueryload in async code unless you’ve benchmarked it, as it can produce surprisingly complex queries under certain join conditions.

Catching N+1 Queries

The insidious thing about the lazy loading trap is that it doesn’t always blow up immediately. If you access a relationship inside an async with session.begin() block where the session is still active, SQLAlchemy might be able to issue the lazy load using the async greenlet bridge. Your code works in dev, where you have one concurrent user, and fails in production, where you have a hundred.

I added a simple guard to catch this during development:

import warnings
from sqlalchemy import event

@event.listens_for(engine.sync_engine, "before_cursor_execute")
def warn_on_lazy_load(conn, cursor, statement, parameters, context, executemany):
    # Flag queries that look like lazy loads (single-row SELECTs by FK)
    if "lazy load" in str(context) if context else False:
        warnings.warn(
            f"Lazy load detected: {statement[:100]}",
            stacklevel=2,
        )

This isn’t bulletproof, but it caught several lazy load paths during development that would have been MissingGreenlet exceptions in production. The real solution is code review discipline: every query should explicitly declare its loading strategy.

The Awaitable Attributes Escape Hatch

SQLAlchemy 2.0 introduced AsyncAttrs, a mixin that makes relationship access awaitable:

from sqlalchemy.ext.asyncio import AsyncAttrs
from sqlalchemy.orm import DeclarativeBase

class Base(AsyncAttrs, DeclarativeBase):
    pass

class Document(Base):
    __tablename__ = "documents"
    # ... columns ...
    tags = relationship("Tag", back_populates="document")

# Now you can do this:
doc = await db.get(Document, doc_id)
tags = await doc.awaitable_attrs.tags  # Issues the query asynchronously

This works, but I’d use it sparingly. It’s essentially lazy loading with an await, which means you’re still issuing extra queries per access. Fine for a one-off in a management script. In a hot path serving API requests, you want explicit eager loading in the original query, not scattered awaits throughout your handler.

Pool Sizing Math

Getting your connection pool right matters more in async code than in synchronous code, because async code’s concurrency model is fundamentally different. A synchronous server with 4 worker processes and 1 thread each needs at most 4 database connections. An async server with 4 worker processes and hundreds of concurrent coroutines per worker might try to use hundreds of connections if you’re not careful.

Here’s how the numbers work:

engine = create_async_engine(
    "postgresql+asyncpg://...",
    pool_size=10,        # Persistent connections per worker
    max_overflow=5,      # Temporary connections when pool is exhausted
    pool_timeout=30,     # Seconds to wait for a connection before error
    pool_recycle=3600,   # Recycle connections after 1 hour
    pool_pre_ping=True,  # Verify connections are alive before using them
)

With 4 Uvicorn workers, this configuration creates up to 4 × (10 + 5) = 60 connections to Postgres. That’s 60 connections from one application server. If you have three replicas behind a load balancer, you’re looking at 180 potential connections.

Postgres’s default max_connections is 100.

You see the problem. I’ve seen this exact miscalculation bring down a production database. The app worked fine with one replica, was sketchy with two, and fell over completely when the third came up and Postgres started rejecting connections with “too many connections” errors.

The Formula

The maximum connections your app can open:

max_connections = num_workers × (pool_size + max_overflow) × num_replicas

This number must be less than Postgres’s max_connections minus whatever connections your other services need (monitoring, migrations, admin tools, PgBouncer itself).

Working backwards from that constraint, I landed on pool_size=5, max_overflow=3 per worker, with 4 workers and 3 replicas: 4 × 8 × 3 = 96 max connections. With Postgres set to max_connections=200, that left plenty of room for other consumers.

PgBouncer Caveats

If you run PgBouncer in front of Postgres (and you probably should for a multi-tenant SaaS), there’s an important interaction with asyncpg. PgBouncer’s transaction pooling mode doesn’t support prepared statements, and asyncpg uses prepared statements by default.

The fix:

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@pgbouncer-host:6432/mydb",
    pool_size=5,
    max_overflow=3,
    connect_args={
        "prepared_statement_cache_size": 0,  # Disable prepared statements
    },
)

Without this, you’ll get cryptic errors about “prepared statement does not exist” that only appear under load when PgBouncer reassigns the underlying Postgres connection to a different client. I lost half a day to this one.

Transactions and the begin() Pattern

Async SQLAlchemy gives you two transaction patterns. The implicit one (which I showed in the get_db dependency) commits at the end of the context. The explicit one uses session.begin():

async def transfer_document(db: AsyncSession, doc_id: str, new_tenant_id: str):
    async with db.begin():
        doc = await db.get(Document, doc_id, with_for_update=True)
        if doc is None:
            raise NotFoundError(f"Document {doc_id} not found")

        old_tenant_id = doc.tenant_id
        doc.tenant_id = new_tenant_id

        # Update the audit log in the same transaction
        audit = AuditLog(
            entity_id=doc_id,
            action="transfer",
            old_value=old_tenant_id,
            new_value=new_tenant_id,
        )
        db.add(audit)
    # Transaction is committed (or rolled back) when the block exits

The with_for_update=True parameter adds SELECT ... FOR UPDATE to prevent concurrent modifications. This is essential for any read-modify-write cycle.

Why expire_on_commit=False Matters

By default, SQLAlchemy expires all attributes on model instances after a commit. In synchronous code, accessing an expired attribute triggers a lazy reload. In async code, that lazy reload raises MissingGreenlet.

# With expire_on_commit=True (the default):
async with SessionFactory() as session:
    doc = await session.get(Document, doc_id)
    await session.commit()
    print(doc.title)  # MissingGreenlet! The attribute was expired by commit.

# With expire_on_commit=False:
async with SessionFactory() as session:
    doc = await session.get(Document, doc_id)
    await session.commit()
    print(doc.title)  # Works fine. The attribute is still in memory.

Set expire_on_commit=False on your session factory. It’s one line of configuration that prevents an entire class of bugs. The trade-off is that your in-memory objects might be stale after commit, but in a request-scoped session that’s about to be disposed anyway, staleness isn’t a concern.

Testing Async DB Code Without Losing Your Mind

Testing async database code is where many teams throw up their hands and just write integration tests that hit a real database. That’s actually not a terrible approach, but you can make it much less painful with a few patterns.

A Reusable Test Session Fixture

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

@pytest_asyncio.fixture
async def db_engine():
    engine = create_async_engine(
        "postgresql+asyncpg://test:test@localhost:5432/test_db",
        echo=False,
    )
    # Create all tables before the test session
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    # Drop all tables after the test session
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await engine.dispose()

@pytest_asyncio.fixture
async def db_session(db_engine):
    session_factory = async_sessionmaker(db_engine, expire_on_commit=False)
    async with session_factory() as session:
        async with session.begin():
            yield session
            # Rollback after each test for isolation
            await session.rollback()

The key trick: wrap each test in a transaction and roll it back when the test finishes. This gives you real database queries with full isolation between tests, without the overhead of creating and dropping tables for every test case.

Overriding the Dependency in FastAPI Tests

from httpx import AsyncClient, ASGITransport

@pytest_asyncio.fixture
async def client(db_session):
    # Override the get_db dependency to use our test session
    app.dependency_overrides[get_db] = lambda: db_session

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        yield client

    app.dependency_overrides.clear()

@pytest.mark.asyncio
async def test_list_documents(client, db_session):
    # Seed test data
    doc = Document(tenant_id="test-tenant", title="Test Doc", body="Content")
    db_session.add(doc)
    await db_session.flush()

    response = await client.get("/tenants/test-tenant/documents")
    assert response.status_code == 200
    assert len(response.json()) == 1
    assert response.json()[0]["title"] == "Test Doc"

One thing I learned the hard way: use db_session.flush() instead of db_session.commit() inside tests. flush() sends the SQL to the database (so your queries can see the data) without committing the transaction (so the rollback at the end still cleans everything up).

The Checklist

After running async SQLAlchemy in production for over a year, here’s what I’d tell anyone starting out:

  1. One session per request, always. Use FastAPI’s dependency injection to scope sessions to request lifetime. Never share sessions between concurrent tasks.
  2. Set expire_on_commit=False on your session factory. This prevents the most common class of MissingGreenlet errors.
  3. Eagerly load every relationship you’ll need. Use selectinload for collections, joinedload for single objects. Treat any relationship access outside a session context as a bug.
  4. Do the pool sizing math before you deploy. Calculate workers × (pool_size + max_overflow) × replicas and make sure it fits within Postgres’s max_connections.
  5. Disable prepared statements if you’re behind PgBouncer. Set prepared_statement_cache_size=0 in connect_args.
  6. Test against a real database with per-test transaction rollback. It’s faster than you think and catches real bugs that mocking misses.
  7. Use session.flush() in tests, not session.commit(). Keeps your test data visible to queries while preserving the rollback cleanup.
  8. Add a lazy load detector in development. Even a simple warning-based listener catches problems before they reach production.

The async ORM is genuinely good. It integrates well with FastAPI and asyncpg, the query API is powerful, and the 2.0 style annotations are a significant improvement over the 1.x API. But it demands more discipline than the synchronous version. You need to think about session scope, loading strategy, and connection math at every layer. Get those right, and it’s a pleasure to work with. Get them wrong, and you’ll be the one staring at MissingGreenlet tracebacks at 2 AM.

This post is based on running async SQLAlchemy 2.0 with FastAPI and asyncpg on a multi-tenant SaaS platform. Code examples are simplified from real production patterns.