FastAPI with async SQLAlchemy 2.1 and Alembic

# AGENTS.md: FastAPI with async SQLAlchemy

FastAPI service on SQLAlchemy 2.1's asyncio extension, with Alembic for migrations and uv for packages. SQLAlchemy 2.1
needs Python 3.11 or newer and no longer installs greenlet by default: declare `sqlalchemy[asyncio]`, or async
engines fail with an ImportError about greenlet.

## Commands
- `uv sync`; add packages with `uv add` (exact pins: `"sqlalchemy[asyncio]==2.1.5"`, `aiosqlite==0.22.1` or
  `asyncpg==0.32.0`, `alembic==1.20.0`).
- Run: `uv run fastapi dev app/main.py`. Tests: `uv run pytest -q`. Lint: `uv run ruff check . && uv run ruff format --check .`.
- Migrations: `uv run alembic revision --autogenerate -m "..."`, review the file, then `uv run alembic upgrade head`
  and `uv run alembic check`.

## Rules
- Async all the way: `create_async_engine`, `async_sessionmaker(engine, expire_on_commit=False)`, `AsyncSession`
  via a `SessionDep` dependency. No sync `Session`, `requests` or blocking file I/O inside `async def` routes.
- One session per request. Never share an `AsyncSession` across tasks (`asyncio.gather` over one session is a bug).
- Relationships use `lazy="raise"`. Load what you return with `selectinload` or `joinedload`. A `MissingGreenlet`
  error means a lazy load slipped through; fix the query, do not wrap it in `run_sync`.
- Commit explicitly in the route or service; roll back and map `IntegrityError` to 409.
- Return Pydantic response models, never ORM objects. Pydantic v2 only.
- Database URLs and passwords come from the environment (`DATABASE_URL`), never `alembic.ini` or code.
- Dispose of the engine in the app's `lifespan` shutdown.
- Never edit a migration that has run outside your machine. Add a new one.

## Tests
- Async tests through `httpx.AsyncClient` and `ASGITransport` with a throwaway database per test.
- List endpoints assert their query count. New resources: use the `async-sqlalchemy-resource` skill.

# CLAUDE.md

@AGENTS.md

## Claude Code notes

- Skills from this loadout live in `.claude/skills/`: `async-sqlalchemy-resource`. Use them when their description matches the task.
- Follow AGENTS.md above; where this file and AGENTS.md disagree, AGENTS.md wins.

<!-- From the Agent Loadout "FastAPI with async SQLAlchemy 2.1 and Alembic" v1.0.0 — https://loadout.jnw-it.com/loadouts/fastapi-async-sqlalchemy/ -->

# GEMINI.md

@./AGENTS.md

## Gemini CLI notes

- Skills from this loadout live in `.agents/skills/`: `async-sqlalchemy-resource`. Use them when their description matches the task.
- Follow AGENTS.md above; where this file and AGENTS.md disagree, AGENTS.md wins.

<!-- From the Agent Loadout "FastAPI with async SQLAlchemy 2.1 and Alembic" v1.0.0 — https://loadout.jnw-it.com/loadouts/fastapi-async-sqlalchemy/ -->

---
name: async-sqlalchemy-resource
description: Add a database-backed resource to a FastAPI app on async SQLAlchemy 2.1 (model, Alembic migration, router and async tests) without lazy-load errors or N+1 queries. Use when adding or changing tables, relationships or endpoints that read or write the database.
---

# Add an async SQLAlchemy resource

The references are a complete working set: `db.py` (engine, session factory, `SessionDep`), `models.py`,
`customers.py` (router), `0001_customers_orders.py` (migration), `alembic_env.py` (async Alembic env) and
`test_customers.py`.

## 1. Model

- Typed mappings: `Mapped[int]`, `mapped_column(...)`. `Mapped[str | None]` means nullable; plain `Mapped[str]` is
  `NOT NULL`.
- Every relationship gets `lazy="raise"`. Under asyncio an implicit lazy load fails with `MissingGreenlet`, and in
  sync code it is a hidden extra query. `raise` turns both into an immediate, obvious error.
- Index foreign keys you filter on (`index=True`). Use `ondelete=` on the foreign key, not ORM cascade alone.

## 2. Migration

```bash
uv run alembic revision --autogenerate -m "add orders"
```

Read the generated file before keeping it. Autogenerate misses renames (it emits drop plus add, which loses data),
some server defaults and enum changes. Then:

```bash
uv run alembic upgrade head
uv run alembic check          # fails if models and migrations disagree
uv run alembic downgrade -1 && uv run alembic upgrade head   # downgrade must work too
```

The env reads `DATABASE_URL` from the environment. Never write a URL with a password into `alembic.ini`.

## 3. Router

- Take the session as `session: SessionDep`. One session per request. Never store it globally or share it between
  `asyncio` tasks: an `AsyncSession` is not safe for concurrent use.
- Load what the response needs in the query: `selectinload(Customer.orders)` for collections, `joinedload` for
  many-to-one. Use `await session.scalars(select(...))`, `await session.get(...)` and `await session.commit()`.
- `expire_on_commit=False` on the session factory keeps attributes readable after commit.
- Map `IntegrityError` to 409 after `await session.rollback()`. Return Pydantic models
  (`ConfigDict(from_attributes=True)`), never ORM objects.
- Need a relationship on an object you already have? `await session.refresh(obj, ["orders"])`, or
  `await obj.awaitable_attrs.orders` on relationships that are not `lazy="raise"`.

## 4. Tests

- `httpx.AsyncClient(transport=ASGITransport(app=app))` with `pytest.mark.anyio`, so tests run on the same event
  loop as the app.
- A fresh `sqlite+aiosqlite://` engine with `StaticPool` per test, and `app.dependency_overrides[get_session]`.
- Count queries with an event listener on `engine.sync_engine` and assert the number of SELECTs for list endpoints.
  That is the test that catches N+1 regressions.

## Done means

pytest, `ruff check`, `ruff format --check` and `alembic check` all pass, and the migration's downgrade works.

"""Async engine, session factory and the per-request session dependency."""

import os
from collections.abc import AsyncIterator
from typing import Annotated

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncAttrs, AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite+aiosqlite:///./app.db")

engine = create_async_engine(DATABASE_URL, pool_pre_ping=True)
# expire_on_commit=False: attributes stay loaded after commit, so building the
# response does not trigger a lazy load (which raises under asyncio).
SessionFactory = async_sessionmaker(engine, expire_on_commit=False)


class Base(AsyncAttrs, DeclarativeBase):
    pass


async def get_session() -> AsyncIterator[AsyncSession]:
    async with SessionFactory() as session:
        yield session


SessionDep = Annotated[AsyncSession, Depends(get_session)]

"""Customer and Order. Relationships use lazy="raise" so a missed eager load fails loudly."""

from datetime import datetime

from sqlalchemy import ForeignKey, String, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.db import Base


class Customer(Base):
    __tablename__ = "customers"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(254), unique=True)
    orders: Mapped[list["Order"]] = relationship(back_populates="customer", lazy="raise")


class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    customer_id: Mapped[int] = mapped_column(ForeignKey("customers.id", ondelete="CASCADE"), index=True)
    total_pence: Mapped[int]
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    customer: Mapped[Customer] = relationship(back_populates="orders", lazy="raise")

"""Customers router: async session per request, eager loads, explicit commits."""

from fastapi import APIRouter, HTTPException, status
from pydantic import BaseModel, ConfigDict, EmailStr, Field
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import selectinload

from app.db import SessionDep
from app.models import Customer, Order

router = APIRouter(prefix="/customers", tags=["customers"])


class OrderOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    total_pence: int


class CustomerIn(BaseModel):
    email: EmailStr


class CustomerOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    email: str
    orders: list[OrderOut]


class OrderIn(BaseModel):
    total_pence: int = Field(gt=0, le=10_000_000)


@router.post("", status_code=status.HTTP_201_CREATED)
async def create_customer(body: CustomerIn, session: SessionDep) -> CustomerOut:
    customer = Customer(email=body.email, orders=[])
    session.add(customer)
    try:
        await session.commit()
    except IntegrityError:
        await session.rollback()
        raise HTTPException(status.HTTP_409_CONFLICT, "email already registered") from None
    return CustomerOut.model_validate(customer)


@router.get("")
async def list_customers(session: SessionDep, limit: int = 50) -> list[CustomerOut]:
    # One query for customers and one for all their orders, however many rows.
    stmt = select(Customer).options(selectinload(Customer.orders)).order_by(Customer.id).limit(min(limit, 200))
    customers = (await session.scalars(stmt)).all()
    return [CustomerOut.model_validate(c) for c in customers]


@router.post("/{customer_id}/orders", status_code=status.HTTP_201_CREATED)
async def add_order(customer_id: int, body: OrderIn, session: SessionDep) -> OrderOut:
    if await session.get(Customer, customer_id) is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "customer not found")
    order = Order(customer_id=customer_id, total_pence=body.total_pence)
    session.add(order)
    await session.commit()
    return OrderOut.model_validate(order)

"""customers and orders

Revision ID: 0001
Revises:
Create Date: 2026-10-17
"""

import sqlalchemy as sa
from alembic import op

revision = "0001"
down_revision = None
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.create_table(
        "customers",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("email", sa.String(length=254), nullable=False),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint("email"),
    )
    op.create_table(
        "orders",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("customer_id", sa.Integer(), nullable=False),
        sa.Column("total_pence", sa.Integer(), nullable=False),
        sa.Column("created_at", sa.DateTime(), server_default=sa.text("(CURRENT_TIMESTAMP)"), nullable=False),
        sa.ForeignKeyConstraint(["customer_id"], ["customers.id"], ondelete="CASCADE"),
        sa.PrimaryKeyConstraint("id"),
    )
    op.create_index("ix_orders_customer_id", "orders", ["customer_id"])


def downgrade() -> None:
    op.drop_index("ix_orders_customer_id", table_name="orders")
    op.drop_table("orders")
    op.drop_table("customers")

"""Alembic env.py for an async engine (from `alembic init -t async`), wired to the app's models.

The URL comes from DATABASE_URL, never from alembic.ini, so no credentials are committed.
"""

import asyncio
import os

from alembic import context
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import create_async_engine

import app.models  # noqa: F401  (registers every model on Base.metadata)
from app.db import Base

target_metadata = Base.metadata
URL = os.environ.get("DATABASE_URL", "sqlite+aiosqlite:///./app.db")


def run_migrations_offline() -> None:
    context.configure(url=URL, target_metadata=target_metadata, literal_binds=True, render_as_batch=True)
    with context.begin_transaction():
        context.run_migrations()


def do_run_migrations(connection: Connection) -> None:
    # render_as_batch lets ALTER-style changes work on SQLite; harmless elsewhere.
    context.configure(connection=connection, target_metadata=target_metadata, render_as_batch=True)
    with context.begin_transaction():
        context.run_migrations()


async def run_migrations_online() -> None:
    engine = create_async_engine(URL, poolclass=pool.NullPool)
    async with engine.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await engine.dispose()


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

"""Async API tests: httpx.AsyncClient over ASGITransport, a fresh in-memory database
per test, and a query counter that catches N+1 loads."""

from collections.abc import AsyncIterator

import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy import event, select
from sqlalchemy.exc import InvalidRequestError
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import StaticPool

from app.db import Base, get_session
from app.main import app
from app.models import Customer

pytestmark = pytest.mark.anyio


@pytest.fixture
def anyio_backend() -> str:
    return "asyncio"


@pytest.fixture
async def db() -> AsyncIterator[dict]:
    engine = create_async_engine("sqlite+aiosqlite://", poolclass=StaticPool)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    factory = async_sessionmaker(engine, expire_on_commit=False)
    queries: list[str] = []
    event.listen(engine.sync_engine, "before_cursor_execute", lambda *a: queries.append(a[2]))

    async def override() -> AsyncIterator:
        async with factory() as session:
            yield session

    app.dependency_overrides[get_session] = override
    yield {"factory": factory, "queries": queries}
    app.dependency_overrides.clear()
    await engine.dispose()


@pytest.fixture
async def client(db: dict) -> AsyncIterator[AsyncClient]:
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c


async def test_create_and_list_without_n_plus_one(client: AsyncClient, db: dict) -> None:
    for i in range(5):
        r = await client.post("/customers", json={"email": f"c{i}@example.com"})
        assert r.status_code == 201
        await client.post(f"/customers/{r.json()['id']}/orders", json={"total_pence": 1000 + i})
    db["queries"].clear()
    r = await client.get("/customers")
    assert r.status_code == 200
    assert [len(c["orders"]) for c in r.json()] == [1, 1, 1, 1, 1]
    selects = [q for q in db["queries"] if q.lstrip().upper().startswith("SELECT")]
    assert len(selects) == 2, selects


async def test_duplicate_email_is_409(client: AsyncClient) -> None:
    assert (await client.post("/customers", json={"email": "a@example.com"})).status_code == 201
    r = await client.post("/customers", json={"email": "a@example.com"})
    assert r.status_code == 409


async def test_order_for_missing_customer_is_404(client: AsyncClient) -> None:
    r = await client.post("/customers/999/orders", json={"total_pence": 500})
    assert r.status_code == 404


async def test_bad_total_is_422(client: AsyncClient) -> None:
    r = await client.post("/customers", json={"email": "b@example.com"})
    bad = await client.post(f"/customers/{r.json()['id']}/orders", json={"total_pence": 0})
    assert bad.status_code == 422


async def test_lazy_load_raises_instead_of_hiding_a_query(client: AsyncClient, db: dict) -> None:
    await client.post("/customers", json={"email": "lazy@example.com"})
    async with db["factory"]() as session:
        customer = (await session.scalars(select(Customer))).one()
        with pytest.raises(InvalidRequestError):
            _ = customer.orders
        await session.refresh(customer, ["orders"])
        assert customer.orders == []