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

---

Async lazy loading rules

The foundation skill defines the lazy-strategy catalog; in async contexts only a subset is safe:

• lazy="selectin" — safe; loads collections with a separate SELECT IN query.
• lazy="raise" — safe; raises MissingGreenlet if accessed without explicit eager loading, forcing selectinload() at query time. Best for large or rarely-needed collections.
• lazy="select" (the SQLAlchemy default) — forbidden: sync lazy load raises in async context. Never leave a relationship() without an explicit lazy=.
• lazy="subquery" — forbidden: not supported by async drivers.

---

Alembic integration (async)

alembic.ini

Point script_location to the alembic/ directory and configure the async database URL:

[alembic]
script_location = alembic
sqlalchemy.url = driver://user:pass@localhost/dbname

The URL in alembic.ini is overridden in env.py — do not put production credentials here.

env.py — async pattern

# alembic/env.py
import asyncio
from logging.config import fileConfig

from alembic import context
from sqlalchemy.ext.asyncio import create_async_engine

from app.core.config import settings
from app.db.base import Base  # import all models so Base.metadata is populated

config = context.config
fileConfig(config.config_file_name)

target_metadata = Base.metadata


def run_migrations_offline() -> None:
    context.configure(
        url=settings.DATABASE_URL,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations() -> None:
    connectable = create_async_engine(settings.DATABASE_URL)
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()


def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    asyncio.run(run_async_migrations())


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

Per the foundation skill's metadata rules, import every model before target_metadata = Base.metadata. The FastAPI convention is an app/db/base.py aggregator:

# app/db/base.py
from app.db.session import Base  # noqa: F401 — must import Base
from app.users.models import User  # noqa: F401
from app.orders.models import Order, OrderLine  # noqa: F401

---

Async anti-patterns

Anti-pattern	Problem	Correct approach
session.execute(select(...)) without await on AsyncSession	Implicit IO in async — raises MissingGreenlet	Always await session.execute(...)
session.commit() in a router handler	Couples transport layer to transaction lifecycle	Let the get_db() dependency commit on yield exit
Bare relationship() with no lazy=	Defaults to lazy="select" (sync lazy load) — raises in async context	Always set lazy="selectin" or lazy="raise"
lazy="subquery" on any relationship	Not supported by async drivers	lazy="selectin" or query-time selectinload()