TopolearnMost APIs are a thin layer over a database; getting sessions and models right avoids connection leaks and data exposure.
FastAPI has no built-in database layer; you bring one. SQLModel, by FastAPI's author, fits most directly: a class can be both a Pydantic model and a SQLAlchemy table, so one definition gives you validation and persistence. Plain SQLAlchemy 2.0 with separate Pydantic schemas is equally common, and the session pattern is the same.
The moving parts:
create_engine(url)once per process. The engine owns a connection pool. For SQLite with sync endpoints addconnect_args={"check_same_thread": False}, because requests run on different threads.SQLModel.metadata.create_all(engine)creates the tables in demos; real projects use Alembic migrations.- A session dependency that opens a
Session, yields it to the endpoint, and closes it when the request is done. Theyieldmakes it a dependency with cleanup code; the Advanced level explains exactly when that cleanup runs.
Queries use SQLAlchemy 2.0 style: session.exec(select(Hero).where(Hero.age > 30).limit(10)).all(), session.get(Hero, hero_id) for primary-key lookups, and session.add(obj), session.commit(), session.refresh(obj) to insert and then load database-generated values such as the id.
Keep the separate-models discipline from the response models lesson. table=True marks the table class; HeroCreate and HeroPublic are plain data models for input and output, so clients can't set id and never see columns like secret_name. Hero.model_validate(data) turns validated input into a table row.
SQLModel and SQLAlchemy bind query values safely, so never build SQL strings from request data with f-strings. And when a lookup finds nothing, raise a 404 rather than returning None.
Example
from typing import Annotated
from fastapi import Depends, FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select
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:///heroes.db", connect_args={"check_same_thread": False})
SQLModel.metadata.create_all(engine)
def get_session():
with Session(engine) as session:
yield session
SessionDep = Annotated[Session, Depends(get_session)]
app = FastAPI()
@app.post("/heroes", response_model=HeroPublic, status_code=201)
def create_hero(data: HeroCreate, session: SessionDep):
hero = Hero.model_validate(data)
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes", response_model=list[HeroPublic])
def list_heroes(session: SessionDep, limit: int = 10):
return session.exec(select(Hero).limit(limit)).all()Remember
- One engine per process; one session per request via a yield dependency
- table=True models map to tables; separate Create/Public models shape the API
- add + commit + refresh to insert and read back generated ids
- Use Alembic for schema changes in real projects
Watch out for
- Creating a new engine per request, which throws away connection pooling.
- Returning table objects without a response model and leaking internal columns.