Databases with SQLAlchemy and session dependencies

Model tables with SQLAlchemy 2.0, give every request its own session, commit explicitly, and stop lazy loading from breaking async endpoints.

Tables as classes

# app/models.py
from datetime import datetime
from sqlalchemy import ForeignKey, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    items: Mapped[list["Item"]] = relationship(back_populates="owner")

class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(120), index=True)
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))
    owner: Mapped[User] = relationship(back_populates="items")

Driver	URL prefix	Notes
psycopg (v3)	`postgresql+psycopg://`	Current choice; has a sync and an async mode
asyncpg	`postgresql+asyncpg://`	Fast async driver; strict about types
SQLite	`sqlite:///./app.db`	Great for tests; needs `check_same_thread=False` with threads
aiosqlite	`sqlite+aiosqlite:///./app.db`	Async SQLite for local development

One session per request

# app/db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("postgresql+psycopg://app:secret@localhost/app", pool_pre_ping=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()          # always released, on success or exception

# app/routers/items.py
from fastapi import Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.orm import Session
from ..db import get_db
from ..models import Item

@router.get("/{item_id}")
def get_item(item_id: int, db: Session = Depends(get_db)):
    item = db.get(Item, item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@router.post("")
def create_item(name: str, db: Session = Depends(get_db)):
    item = Item(name=name)
    db.add(item)
    db.commit()
    db.refresh(item)
    return item

The yield dependency guarantees the session closes even when the endpoint raises. Without it, a failing request leaks a connection from the pool until the pool is exhausted.

Eager loading and transactions

from sqlalchemy.orm import selectinload

# N+1 avoided: one query for users, one for their items
users = db.scalars(
    select(User).options(selectinload(User.items)).order_by(User.id).limit(50)
).all()

# a transaction block: roll back everything on error
with db.begin():
    db.add(User(email="a@example.com"))
    db.add(Item(name="widget", owner_id=1))

⚠️

In an async def endpoint, touching an unloaded relationship triggers a lazy load on a sync connection and raises MissingGreenlet. Either load relationships explicitly with selectinload or use an AsyncSession with await session.execute(...) throughout.

FAQ

Where should I create tables?

Not in application startup for a real service. Use a migration tool (Alembic) where alembic revision --autogenerate compares your models to the database. Creating tables at boot works only until the first schema change.

Why did my object become detached after commit?

By default committing expires the instance, so the next attribute access needs a live session. Setting expire_on_commit=False keeps values readable, and response_model then serialises them without another query.

Pydantic request and response models Testing FastAPI apps with TestClient and pytest

Last refreshed 2026-09-18.

Databases with SQLAlchemy and session dependencies

Tables as classes

One session per request

Eager loading and transactions

FAQ

Related