How do you use SQLAlchemy 2.0 async with FastAPI?

Summary

Create an async engine (create_async_engine with asyncpg), an async_sessionmaker, provide a session per request via a yield dependency, and use select() with await session.execute() or session.scalars().

Longer Answer

Set expire_on_commit=False so returned objects stay usable after commit. Lazy loading doesn't work implicitly in async (it raises MissingGreenlet), so eager-load relationships with selectinload()/joinedload(). Configure pool size based on worker count and the DB's max connections. Use Alembic for migrations.

Code Example

from sqlalchemy import select
from sqlalchemy.orm import selectinload
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine(settings.database_url, pool_size=10, max_overflow=5)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

@router.get("/{user_id}/orders")
async def user_orders(user_id: int, db: DbSession):
    stmt = select(User).options(selectinload(User.orders)).where(User.id == user_id)
    user = (await db.scalars(stmt)).one_or_none()
    if user is None:
        raise HTTPException(404)
    return user.orders

Key Points

• create_async_engine('postgresql+asyncpg://...')
• Session per request via yield dependency
• expire_on_commit=False
• Eager loading (selectinload) instead of lazy loads

Common mistakes

• Accessing lazy relationships in async code (MissingGreenlet)
• Sharing one session across requests