FastAPI: Databases with async SQLAlchemy 2.0

Most FastAPI services are a thin layer over a database, and most of their production incidents come from how that database is used: sessions shared between requests, connections leaked, N+1 query storms, pools too small for the worker count, commits in the wrong place. SQLAlchemy 2.0 with its async API is the standard toolkit. This part wires it into FastAPI the way experienced teams do. Every example runs on SQLite through aiosqlite; in production you would change only the URL to postgresql+asyncpg://….

1. The pieces and their lifetimes

2. A complete, correct setup

models, lifespan, a session per request, a repository, and endpoints

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from decimal import Decimal
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.testclient import TestClient
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import ForeignKey, Numeric, String, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, selectinload

# ---------------------------------------------------------------- models (SQLAlchemy 2.0 typed style)
class Base(DeclarativeBase):
    pass

class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    customer_email: Mapped[str] = mapped_column(String(254), index=True)
    status: Mapped[str] = mapped_column(String(20), default="pending")
    items: Mapped[list["OrderItem"]] = relationship(back_populates="order", cascade="all, delete-orphan")

class OrderItem(Base):
    __tablename__ = "order_items"
    id: Mapped[int] = mapped_column(primary_key=True)
    order_id: Mapped[int] = mapped_column(ForeignKey("orders.id"))
    sku: Mapped[str] = mapped_column(String(20))
    qty: Mapped[int]
    unit_price: Mapped[Decimal] = mapped_column(Numeric(10, 2))
    order: Mapped[Order] = relationship(back_populates="items")

# ---------------------------------------------------------------- schemas
class ItemIn(BaseModel):
    sku: str
    qty: int = Field(gt=0)
    unit_price: Decimal = Field(gt=0, decimal_places=2)

class OrderIn(BaseModel):
    customer_email: str
    items: list[ItemIn] = Field(min_length=1)

class ItemOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)       # read attributes from ORM objects
    sku: str
    qty: int
    unit_price: Decimal

class OrderOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    customer_email: str
    status: str
    items: list[ItemOut]

# ---------------------------------------------------------------- engine + sessions
@asynccontextmanager
async def lifespan(app: FastAPI):
    engine = create_async_engine("sqlite+aiosqlite:///./shop.db", echo=False)   # postgresql+asyncpg://… in prod
    async with engine.begin() as conn:                     # demo only: real apps use migrations (lesson 16)
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)
    yield {"sessionmaker": async_sessionmaker(engine, expire_on_commit=False)}
    await engine.dispose()                                 # close every pooled connection

async def get_session(request: Request) -> AsyncIterator[AsyncSession]:
    async with request.state.sessionmaker() as session:   # one session per request, always closed
        yield session

SessionDep = Annotated[AsyncSession, Depends(get_session)]

# ---------------------------------------------------------------- repository
class OrderRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def add(self, data: OrderIn) -> Order:
        order = Order(customer_email=data.customer_email.lower(),
                      items=[OrderItem(**item.model_dump()) for item in data.items])
        self.session.add(order)
        await self.session.flush()                          # assigns the id, inside the transaction
        return order

    async def get(self, order_id: int) -> Order | None:
        stmt = select(Order).where(Order.id == order_id).options(selectinload(Order.items))
        return (await self.session.execute(stmt)).scalar_one_or_none()

def get_repo(session: SessionDep) -> OrderRepository:
    return OrderRepository(session)

Repo = Annotated[OrderRepository, Depends(get_repo)]

# ---------------------------------------------------------------- endpoints
app = FastAPI(lifespan=lifespan)

@app.post("/orders", status_code=201)
async def create_order(body: OrderIn, repo: Repo, session: SessionDep) -> OrderOut:
    order = await repo.add(body)
    await session.commit()                                  # commit BEFORE returning (lesson 06)
    return await repo.get(order.id)

@app.get("/orders/{order_id}")
async def read_order(order_id: int, repo: Repo) -> OrderOut:
    order = await repo.get(order_id)
    if order is None:
        raise HTTPException(404, "order not found")
    return order

with TestClient(app) as client:
    body = {"customer_email": "[email protected]",
            "items": [{"sku": "TEA-01", "qty": 2, "unit_price": "120.50"}, {"sku": "MLK-02", "qty": 1, "unit_price": "32.00"}]}
    created = client.post("/orders", json=body)
    print(created.status_code, created.json())
    print(client.get(f"/orders/{created.json()['id']}").json()["items"][1])
    print(client.get("/orders/999").status_code)

Output

201 {'id': 1, 'customer_email': '[email protected]', 'status': 'pending', 'items': [{'sku': 'TEA-01', 'qty': 2, 'unit_price': '120.50'}, {'sku': 'MLK-02', 'qty': 1, 'unit_price': '32.00'}]}
{'sku': 'MLK-02', 'qty': 1, 'unit_price': '32.00'}
404

3. The N+1 problem, counted

Listing 50 orders and touching each one’s items can issue 1 query for the orders plus 1 per order for its items: 51 round trips. In async SQLAlchemy, lazy loading on attribute access is not even allowed. It raises an error instead of silently issuing queries, which is a blessing. Eager loading fixes it:

lazy access fails loudly; selectinload does it in 2 queries

import asyncio

from sqlalchemy import ForeignKey, String, event, select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, selectinload

class Base(DeclarativeBase):
    pass

class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    items: Mapped[list["Item"]] = relationship()

class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(primary_key=True)
    order_id: Mapped[int] = mapped_column(ForeignKey("orders.id"))
    sku: Mapped[str] = mapped_column(String(20))

async def main():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    queries = []
    event.listen(engine.sync_engine, "before_cursor_execute",
                 lambda conn, cur, stmt, *a: queries.append(stmt.split()[0]))
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    Session = async_sessionmaker(engine, expire_on_commit=False)
    async with Session() as s:
        s.add_all([Order(items=[Item(sku=f"SKU-{o}-{i}") for i in range(3)]) for o in range(50)])
        await s.commit()

    async with Session() as s:
        orders = (await s.execute(select(Order))).scalars().all()
        try:
            print(orders[0].items)                                   # implicit lazy load
        except Exception as e:
            print("lazy load in async:", type(e).__name__)

    for label, stmt in [("selectinload", select(Order).options(selectinload(Order.items)))]:
        queries.clear()
        async with Session() as s:
            orders = (await s.execute(stmt)).scalars().all()
            total_items = sum(len(o.items) for o in orders)
        print(f"{label}: {len(orders)} orders, {total_items} items in {len(queries)} SQL statements")
    await engine.dispose()

asyncio.run(main())

Output

lazy load in async: MissingGreenlet
selectinload: 50 orders, 150 items in 2 SQL statements

4. Transactions and constraint errors

a unique constraint becomes a clean 409

import asyncio

from sqlalchemy import String, UniqueConstraint
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Customer(Base):
    __tablename__ = "customers"
    __table_args__ = (UniqueConstraint("email"),)
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(254))

class EmailTaken(Exception):
    pass

async def register(Session, email):
    async with Session() as s:
        try:
            async with s.begin():                        # commits on success, rolls back on any exception
                s.add(Customer(email=email.lower()))
        except IntegrityError:
            raise EmailTaken(email) from None             # mapped to 409 by an exception handler (lesson 09)

async def main():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    Session = async_sessionmaker(engine)
    await register(Session, "[email protected]")
    try:
        await register(Session, "[email protected]")      # same address, different case
    except EmailTaken as e:
        print("409 Conflict: email already registered:", e)
    await engine.dispose()

asyncio.run(main())

Output

409 Conflict: email already registered: [email protected]

Let the database enforce uniqueness. A “check if the email exists, then insert” in Python has a race: two requests both check, both see “free”, both insert. The constraint is the only thing that is correct under concurrency. Catch the error and translate it.

5. Connection pools: the arithmetic that prevents outages

does the fleet fit in the database?

pods, workers_per_pod, pool_size, max_overflow = 6, 4, 10, 5
db_max_connections, reserved_for_admin_and_jobs = 200, 20

per_process = pool_size + max_overflow
fleet_peak = pods * workers_per_pod * per_process
budget = db_max_connections - reserved_for_admin_and_jobs
print(f"peak connections: {pods} pods × {workers_per_pod} workers × {per_process} = {fleet_peak}")
print(f"database budget: {budget} → {'OK' if fleet_peak <= budget else 'OVERLOAD: shrink pools or add a pooler (PgBouncer)'}")
print(f"autoscaling to 10 pods would need {10 * workers_per_pod * per_process} connections")

Output

peak connections: 6 pods × 4 workers × 15 = 360
database budget: 180 → OVERLOAD: shrink pools or add a pooler (PgBouncer)
autoscaling to 10 pods would need 600 connections

Recap