Databases with SQLAlchemy

This page wires FastAPI to a relational database using SQLAlchemy 2.0 (sync style) and the Depends system: one session per request, clean separation between DB models and Pydantic schemas.

Setup: Engine, Session Factory, Models

pip install sqlalchemy

# db/session.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

DATABASE_URL = "postgresql://user:pass@localhost:5432/appdb"

engine = create_engine(DATABASE_URL, pool_pre_ping=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

# db/models.py
from sqlalchemy import String, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from db.session import Base

class Item(Base):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), index=True)
    price: Mapped[float]
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    owner: Mapped["User"] = relationship(back_populates="items")

The get_db Dependency (One Session per Request)

from db.session import SessionLocal

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

FastAPI opens a session before the endpoint and closes it after — including on exceptions. Endpoints never create sessions themselves; they receive db: Session = Depends(get_db).

CRUD Endpoint Using the Session

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session

app = FastAPI()

@app.post("/items/", response_model=ItemOut, status_code=201)
def create_item(item_in: ItemCreate, db: Session = Depends(get_db)):
    item = Item(**item_in.model_dump())
    db.add(item)
    db.commit()
    db.refresh(item)          # reload DB-generated fields (id)
    return item

@app.get("/items/{item_id}", response_model=ItemOut)
def read_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

Bridging ORM Models and Pydantic (from_attributes)

from pydantic import BaseModel, ConfigDict

class ItemOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)   # Pydantic v2; v1: orm_mode = True

    id: int
    name: str
    price: float

from_attributes=True lets response_model=ItemOut read data from the ORM object (item.name) instead of only dicts. Returning the ORM object directly keeps passwords or internal columns out when the output schema excludes them.

Queries: Filtering, Joins, Relationships

from sqlalchemy import select

@app.get("/items/", response_model=list[ItemOut])
def list_items(min_price: float = 0, db: Session = Depends(get_db)):
    stmt = (
        select(Item)
        .where(Item.price >= min_price)
        .order_by(Item.price.desc())
        .limit(50)
    )
    return db.scalars(stmt).all()

# lazy vs eager loading:
stmt = select(Item).options(selectinload(Item.owner))  # avoid N+1 queries

SQLAlchemy 2.0 uses a unified select() style. Beware the N+1 problem: accessing item.owner in a loop triggers one query per item unless you eager-load with selectinload or joinedload.

Creating Tables and Migrations

Base.metadata.create_all(bind=engine)   # quick start (dev only)

For real projects use Alembic: versioned migrations (alembic revision --autogenerate, alembic upgrade head) so schema changes are reviewable and reversible. create_all never alters existing tables.

Quick Questions and Answers

Q: Why do I get "Object is not attached to a session" or detached-instance errors?

A: The session closed before you read a lazy relationship. Set expire_on_commit=False so objects stay usable after commit(), eager-load relationships, or serialize within the request (the response_model conversion happens while the session is still open).

Q: Where should transactions begin and end?

A: Keep the unit of work inside the request: db.commit() per logical operation, or open a transaction in get_db and commit/rollback in the finally for all-or-nothing endpoints. Do not share one session across requests — sessions are not thread-safe.

Q: Why does the DB pool get exhausted under load?

A: Sessions that are never closed (missing finally: db.close()) or leaked connections from forgotten commits. Sync SQLAlchemy blocks a thread per query — with many concurrent sync endpoints, tune pool_size/max_overflow and consider the async driver (advanced level) so the event loop handles waiting.

Conclusion

The pattern to memorize: get_db yield-dependency → endpoint receives a Session → SQLAlchemy models for storage → Pydantic schemas (from_attributes=True) for the API boundary. With Alembic for migrations, this is the backbone of most FastAPI services.

← Back to Intermediate Index