SQLAlchemy 2.0 Async with FastAPI: Best Practices

SQLAlchemy 2.0's async API with FastAPI is the correct choice for production — but it has subtle differences from the sync API that cause real bugs. This covers the patterns you need and the mistakes to avoid.

Session factory and dependency

# app/db.py
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

engine = create_async_engine(
    settings.DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,  # Check connection health before use
    echo=settings.DEBUG,
)

AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,  # Critical — see below
)

async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

expire_on_commit=False is critical for async. With expire_on_commit=True (the default), accessing attributes after commit() triggers a lazy load — but in async mode, that's a MissingGreenlet error. Set it to False and access attributes before committing if you need them in the response.

Correct model definitions

Use SQLAlchemy 2.0's Mapped annotations throughout:

# app/models/user.py
import uuid
from datetime import datetime
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy import String, Boolean, DateTime
from app.db import Base

class User(Base):
    __tablename__ = "users"
    
    id: Mapped[uuid.UUID] = mapped_column(
        primary_key=True,
        default=uuid.uuid4,
    )
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str]
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
    
    # Relationship — use selectin loading for async
    api_keys: Mapped[list["APIKey"]] = relationship(
        back_populates="owner",
        lazy="selectin",  # NOT "select" or "subquery"
    )

lazy="selectin" is the right loading strategy for async. It issues a second query with SELECT IN rather than lazy loading, which is compatible with async sessions.

Repository pattern

Don't write raw SQLAlchemy in route handlers. Use a repository pattern:

# app/repositories/user.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models import User

class UserRepository:
    def __init__(self, db: AsyncSession):
        self.db = db
    
    async def get_by_email(self, email: str) -> User | None:
        result = await self.db.execute(
            select(User).where(User.email == email)
        )
        return result.scalar_one_or_none()
    
    async def create(self, email: str, hashed_password: str) -> User:
        user = User(email=email, hashed_password=hashed_password)
        self.db.add(user)
        await self.db.flush()  # Get the ID without committing
        return user
    
    async def get_with_keys(self, user_id: uuid.UUID) -> User | None:
        result = await self.db.execute(
            select(User)
            .options(selectinload(User.api_keys))
            .where(User.id == user_id)
        )
        return result.scalar_one_or_none()

# Usage in routes
async def get_user_repo(db: AsyncSession = Depends(get_db)):
    return UserRepository(db)

Common async pitfalls

1. Lazy loading after session close

# Wrong — session closes when context exits
async with AsyncSessionLocal() as db:
    user = await db.get(User, user_id)
# Session closed here

# This triggers lazy load — MissingGreenlet error!
print(user.api_keys)

# Right — load everything you need inside the session
async with AsyncSessionLocal() as db:
    result = await db.execute(
        select(User)
        .options(selectinload(User.api_keys))
        .where(User.id == user_id)
    )
    user = result.scalar_one()
    api_keys = user.api_keys  # Loaded inside session

2. Using sync session methods

# Wrong — sync, blocks event loop
user = db.query(User).filter(User.id == user_id).first()

# Right — async
result = await db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()

3. Missing flush before accessing generated IDs

# Wrong — id may not be populated yet
db.add(user)
print(user.id)  # May be None

# Right — flush populates DB-generated values without committing
db.add(user)
await db.flush()
print(user.id)  # UUID is set

Transactions across multiple operations

Use a single session for operations that must be atomic:

async def create_user_with_api_key(
    email: str,
    db: AsyncSession = Depends(get_db),
) -> tuple[User, APIKey]:
    user_repo = UserRepository(db)
    key_repo = APIKeyRepository(db)
    
    user = await user_repo.create(email, hash_password("temp"))
    # user.id is available after flush (happens in create)
    
    raw_key, key_hash, prefix = generate_api_key()
    api_key = await key_repo.create(
        owner_id=user.id,
        key_hash=key_hash,
        prefix=prefix,
    )
    
    # Both user and api_key commit together — atomic
    # get_db() commits on yield return
    return user, raw_key  # Return raw key before commit

FastAPI AI Kit implements these patterns throughout: get_db() dependency, repository classes for all models, and selectin loading by default. The boilerplate decisions are made for you.