sqlalchemy-patterns

SQLAlchemy Patterns for FastAPI (async delta)

Load python-foundation:sqlalchemy-patterns via the Skill tool FIRST. It contains the shared SQLAlchemy 2.0 core: detection, Mapped/mapped_column model definition, column type guidance, select() querying, relationship structure, lazy-strategy overview, and migration metadata rules. This skill covers only the async/FastAPI delta.

---

Async session setup

# app/db/session.py
from collections.abc import AsyncGenerator

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

from app.core.config import settings

engine = create_async_engine(
    settings.DATABASE_URL,
    echo=False,
    pool_pre_ping=True,
    pool_size=10,
    max_overflow=20,
)

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


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

Use expire_on_commit=False so that model attributes remain accessible after a commit without triggering lazy loads — important in async contexts where implicit IO is not allowed.

Use pool_pre_ping=True to detect stale connections before use.

The get_db() dependency owns the transaction boundary: it commits on successful yield exit and rolls back on exception. Never call session.commit() in a router handler.

---

Async querying

Every execution is awaited; statement construction follows the foundation skill.

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.users.models import User


async def get_user_by_id(db: AsyncSession, user_id: int) -> User | None:
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()


async def get_user_with_orders(db: AsyncSession, user_id: int) -> User | None:
    result = await db.execute(
        select(User)
        .options(selectinload(User.orders))
        .where(User.id == user_id)
    )
    return result.scalar_one_or_none()


async def create_user(db: AsyncSession, email: str, hashed_password: str, display_name: str) -> User:
    user = User(email=email, hashed_password=hashed_password, display_name=display_name)
    db.add(user)
    await db.flush()  # assigns id without committing; session.commit() happens in get_db()
    return user