Data and State

FastAPI is not tied to any particular database or ORM. This chapter connects a database with SQLModel, the library the official docs use in their examples, manages sessions through dependencies, reads configuration with pydantic-settings, and runs work after the response with background tasks.

SQLModel and SQLAlchemy

SQLModel, written by FastAPI's author, layers Pydantic on top of SQLAlchemy so a single class can define both a table and an API model. Using SQLAlchemy directly is also a common choice, and the pattern of injecting the session as a dependency is the same.

pip install sqlmodel

Models and the engine

A class with table=True becomes a database table; other classes are plain data models for input and output. Putting shared fields in a base class and inheriting from it avoids repetition.

# app/db.py
from typing import Annotated

from fastapi import Depends
from sqlmodel import Field, Session, SQLModel, create_engine


class HeroBase(SQLModel):
    name: str = Field(index=True)
    age: int | None = None


class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    secret_name: str


class HeroCreate(HeroBase):
    secret_name: str


class HeroPublic(HeroBase):
    id: int


engine = create_engine(
    "sqlite:///database.db",
    connect_args={"check_same_thread": False},  # only needed for SQLite
)


def get_session():
    with Session(engine) as session:
        yield session


SessionDep = Annotated[Session, Depends(get_session)]

get_session is a dependency that uses yield. Each request gets its own session, passed into the path operation, and the with block closes it once the response is done. You can create tables at startup in lifespan with SQLModel.metadata.create_all(engine), but in production a migration tool such as Alembic is the better choice.

CRUD endpoints

With the session dependency in place, reads, inserts and lookups are short. Because the response model is HeroPublic, secret_name never appears in responses.

from typing import Annotated

from fastapi import FastAPI, HTTPException, Query
from sqlmodel import select

from app.db import Hero, HeroCreate, HeroPublic, SessionDep

app = FastAPI()


@app.post("/heroes", response_model=HeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero


@app.get("/heroes", response_model=list[HeroPublic])
def read_heroes(
    session: SessionDep,
    offset: int = 0,
    limit: Annotated[int, Query(le=100)] = 100,
):
    return session.exec(select(Hero).offset(offset).limit(limit)).all()


@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero