SQL Databases with SQLModel

This content was generated with the assistance of AI and should be verified against the official documentation before being relied on in production.

The reference material consulted while preparing these pages is listed in the section bibliography.

This page is written against FastAPI 0.142.2 (released 2026-09-30), Pydantic 2.13, Starlette 1.7 and SQLModel 0.0.47 (released 2026-09-23), checked on 2026-10-02. SQLModel 0.0.47 requires SQLAlchemy >=2.0.14,<2.1, so every example runs on SQLAlchemy 2.0 underneath SQLModel (the venv had 2.0.54), even though a newer 2.1 line exists on PyPI. FastAPI supports Python 3.10 through 3.14; the code on this page was executed on Python 3.13 with TestClient against SQLite (a file database, an in-memory one for the tests, and aiosqlite for the async variant). Nothing needed a database server: the PostgreSQL pieces (the asyncpg driver, the psycopg URL, the Compose file, the PostgreSQL-specific Alembic run) were only checked as far as they can be without a server, and each section says so. FastAPI does not require any database; this page shows the officially documented route, SQLModel, and then the plain-SQLAlchemy and raw DB-API routes for when you want less magic. The running scenario is the Bookshelf API (books, authors and reviews).

SQLModel: SQLAlchemy 2.0 and Pydantic in One Class

SQLModel is a library by the author of FastAPI, built to be the natural fit for FastAPI applications that need a relational database. It adds almost no machinery of its own; it fuses two libraries you already know:

Layer Provides Where you meet it

SQLAlchemy 2.0

Engine, connection pool, Core SQL expressions, the ORM (mapped classes, relationships, the unit of work), dialects for PostgreSQL, MySQL, SQLite, Oracle, SQL Server

create_engine, Session, select, Relationship, every database error (sqlalchemy.exc)

Pydantic v2

Validation, parsing, serialization, JSON Schema

Field constraints, model_validate, model_dump, the OpenAPI schemas FastAPI generates

SQLModel

One class that is both: a SQLModel subclass with table=True is an ORM-mapped table and a Pydantic model; without table=True it is a plain Pydantic data model

Field(primary_key=True, foreign_key=…​, index=True), Relationship, session.exec(), sqlmodel_update()

sqlmodel provides its own Field, Relationship, Session, AsyncSession (in sqlmodel.ext.asyncio.session), a typed select and col, and re-exports SQLAlchemy’s create_engine and func unchanged (verified: they are the same objects). A typical import line is from sqlmodel import Field, Session, SQLModel, create_engine, select. Everything else on the SQLAlchemy side (exceptions, selectinload, event, pool classes, the async extension) is imported from sqlalchemy directly. That is intentional: SQLModel is a thin layer, and the SQLAlchemy documentation remains the reference for anything relational.

Install it with the project tool of your choice (the official tutorial uses uv add sqlmodel; pip install sqlmodel works the same). A driver is needed for anything other than SQLite, which ships with Python:

uv add sqlmodel                    # SQLModel pulls in SQLAlchemy 2.0 and Pydantic v2
uv add "psycopg[binary]"           # PostgreSQL, synchronous (psycopg 3)
uv add asyncpg                     # PostgreSQL, asynchronous
uv add aiosqlite                   # SQLite, asynchronous (handy for tests)
uv add alembic                     # migrations

Source: FastAPI: SQL (Relational) Databases

SQLModel is still 0.0.x, like FastAPI is still 0.x: pin both. When should you reach for something else?

Choice Pick it when Cost

SQLModel

You want one set of classes for table, validation and OpenAPI, and the official FastAPI tutorial’s approach

Table models and data models are two different jobs (below); the SQLModel release line is 0.0.x and lags SQLAlchemy 2.1

Plain SQLAlchemy 2.0 ORM with Pydantic schemas

You want full SQLAlchemy typing (Mapped[…​]), no SQLModel version coupling, or an existing SQLAlchemy codebase

You write the Pydantic schemas yourself and convert with from_attributes

SQLAlchemy Core or raw DB-API

Reporting queries, bulk operations, tiny services, or a team that prefers SQL text

You manage mapping, transactions and thread-safety yourself

A non-relational store

The data is documents, a cache, search or vectors rather than related rows

See the NoSQL page of this section and Databases

The relational background (keys, joins, normalization, transactions) is on this site: SQL, relations, normalization and transactions. For the same ORM ideas on the JVM, compare Hibernate Reference. For choosing a store at all, see Choosing the Right Database.

The Bookshelf Schema

The page builds towards four tables: authors, books, the link table that makes authors and books many-to-many (a book can have several authors, an author several books), and reviews, which belong to exactly one book.

erDiagram AUTHOR ||--o{ BOOKAUTHORLINK : "writes" BOOK ||--o{ BOOKAUTHORLINK : "is written by" BOOK ||--o{ REVIEW : "receives" AUTHOR { int id PK string name "indexed" int born "nullable" } BOOK { int id PK string title "indexed" string isbn "unique" int year "nullable" } BOOKAUTHORLINK { int book_id PK, FK "ON DELETE CASCADE" int author_id PK, FK "ON DELETE CASCADE" } REVIEW { int id PK int book_id FK "indexed, ON DELETE CASCADE" int rating "1 to 5" string text }
Figure 1. The Bookshelf schema

The table names (author, book, bookauthorlink, review) are what SQLModel derives from the class names: the lower-cased class name. Set tablename on the class when you want something else.

Engine, Session Dependency and Table Creation

Three objects do all the plumbing, and each has a clear lifetime:

Object Lifetime Job

engine

One per process, created at import time

Holds the connection pool and the dialect. Creating it does not connect.

Session

One per request

The unit of work: tracks the objects you loaded or added and flushes them in a transaction. Not thread-safe, never shared.

Tables

Created once

SQLModel.metadata.create_all(engine) for a prototype, Alembic for anything that must evolve (below)

The engine is built from a URL. The official tutorial uses a SQLite file so that the code runs as it is. The session is a dependency with yield: it opens a Session, hands it to the path operation, and closes it afterwards, which is the pattern described on Dependencies with yield. The SessionDep alias keeps every route signature short. Table creation moves into the lifespan, the modern replacement for the deprecated startup and shutdown events (FastAPI 0.93.0, 2023-03-07), whose full treatment is on Settings, Lifespan & Background Tasks.

import os
import secrets
from collections.abc import AsyncIterator, Iterator
from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI, Header, HTTPException, Query, Response
from pydantic import field_validator
from sqlalchemy.exc import IntegrityError
from sqlmodel import Field, Session, SQLModel, create_engine, func, select


DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///bookshelf.db")
connect_args = {"check_same_thread": False} if DATABASE_URL.startswith("sqlite") else {}
engine = create_engine(DATABASE_URL, connect_args=connect_args)


def get_session() -> Iterator[Session]:
    with Session(engine) as session:
        yield session


SessionDep = Annotated[Session, Depends(get_session)]


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    SQLModel.metadata.create_all(engine)
    yield
    engine.dispose()


app = FastAPI(title="Bookshelf API", lifespan=lifespan)


def require_admin(x_admin_key: Annotated[str | None, Header()] = None) -> None:
    expected = os.environ.get("BOOKSHELF_ADMIN_KEY", "")
    if not expected or not x_admin_key or not secrets.compare_digest(x_admin_key, expected):
        raise HTTPException(status_code=401, detail="Admin key required")

Source: FastAPI: SQL (Relational) Databases

This is the top of bookshelf_db.py; the models and the routes follow in the next sections. Points to take from it:

  • connect_args={"check_same_thread": False} is for SQLite only and is explained in the pooling section: FastAPI may use one connection from different threads. It is not passed for other databases.

  • The URL comes from the environment (DATABASE_URL), with a SQLite file as the local default. Never hard-code a server URL with a password; a settings class that reads a secret file is shown in the PostgreSQL section.

  • get_session is a generator function declared with def, so FastAPI runs it in the threadpool, exactly like a def path operation. A with Session(engine) as session block guarantees close() runs, which returns the connection to the pool.

  • create_all creates only tables that do not exist. It never alters an existing table, so changing a model later needs a migration. It is also blocking, which is acceptable once at startup. In the lifespan it runs before the first request is served; the engine.dispose() after yield closes the pooled connections on shutdown.

  • The require_admin dependency protects the destructive route below: the key comes from the environment, is compared with secrets.compare_digest (constant-time), and an unset key rejects everybody. Real applications use the OAuth2 machinery from Security Fundamentals instead.

  • The official tutorial page still shows @app.on_event("startup") for table creation. Prefer the lifespan form.

Table Models and Data Models

A single class that is the table, the request body and the response is tempting and wrong: the client could set the id, the response would expose columns that must stay private, and a PATCH would need every field to be optional. The documented answer is a small family of classes. Only one of them has table=True:

Class table=True Role

BookBase

no

The fields shared by all the others (title, ISBN, year) with their validation rules, written once

Book

yes

The table: BookBase plus the database-only columns (id, internal_notes). Never used as a request body, and never returned as is.

BookCreate

no

The POST body: the shared fields, no id

BookPublic

no

The response: shared fields plus the id, which is always present once the row exists

BookUpdate

no

The PATCH body: every field optional, so a client sends only what changes

 
class BookBase(SQLModel):
    title: str = Field(index=True, min_length=1, max_length=200)
    isbn: str = Field(unique=True, min_length=10, max_length=17)
    year: int | None = Field(default=None, ge=1400, le=2100)


class Book(BookBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    internal_notes: str | None = None  # never leaves the server


class BookCreate(BookBase):
    pass


class BookPublic(BookBase):
    id: int


class BookUpdate(SQLModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    isbn: str | None = Field(default=None, min_length=10, max_length=17)
    year: int | None = Field(default=None, ge=1400, le=2100)

    @field_validator("title", "isbn")
    @classmethod
    def not_null(cls, value: str | None) -> str:
        if value is None:  # sent explicitly as null, not merely omitted
            raise ValueError("must not be null")
        return value


class BooksPage(SQLModel):
    items: list[BookPublic]
    total: int
    offset: int
    limit: int

Source: FastAPI: SQL Databases, multiple models

What each detail does:

  • Field(…​) carries both worlds. min_length, max_length, ge and le are Pydantic constraints (validated in the request, and published in the OpenAPI schema); index=True, unique=True and primary_key=True are database settings. One declaration, both effects. int | None makes a column nullable.

  • id: int | None = Field(default=None, primary_key=True). The None default is for the moment before the row is inserted. In BookPublic the id is a plain int, so the OpenAPI response schema marks it required.

  • BookUpdate and explicit nulls. All fields default to None, which makes a missing field and "title": null indistinguishable by value. exclude_unset=True (below) tells them apart, but an explicit null for a NOT NULL column would reach the database. The field_validator rejects it with a 422. Validators run only for values that were sent, so omitted fields are untouched.

  • Table models skip validation on construction. Book(title="", isbn="1") is created without complaint (verified): that is how SQLAlchemy mapped classes behave. Validation comes from Book.model_validate(book_create), which does validate (and also copies the data into a table object). That is why the POST route validates the body model first and then converts, rather than building Book(**data) by hand.

  • Never return the table model. Declaring → Book alone returns every column, including the private one: a verified run returned internal_notes to the client, and the OpenAPI schema grew a Book component. Setting response_model=BookPublic filters the output and documents the right schema. The return annotation alone is also enough since FastAPI 0.89.0 (2023-01-07) when it is itself a data model such as → BookPublic; the routes here keep response_model= because they return a different class (the table object) than the one they document. See Response Models & Return Types.

  • Pydantic v2 vocabulary. SQLModel 0.0.47 needs Pydantic 2.11 or newer: model_validate and model_dump, not the v1 parse_obj and .dict(), and from_attributes (set by SQLModel for you) instead of orm_mode. Pydantic basics are on Pydantic v2 Essentials and body models on Request Body & Models.

  • session.exec(), not session.execute() or session.query(). exec is SQLModel’s own method: for a select(Book) it returns the Book objects directly (a scalar result), where SQLAlchemy’s execute returns rows you must unpack. session.query() still works but emits a DeprecationWarning that points back to exec() (verified).

Full CRUD with PATCH

The routes below are the rest of bookshelf_db.py, in file order. Create first:

@app.post("/books", response_model=BookPublic, status_code=201)
def create_book(book: BookCreate, session: SessionDep) -> Book:
    db_book = Book.model_validate(book)
    session.add(db_book)
    try:
        session.commit()
    except IntegrityError:
        session.rollback()
        raise HTTPException(status_code=409, detail="ISBN already exists")
    session.refresh(db_book)
    return db_book

Source: FastAPI: SQL Databases, create a hero

Book.model_validate(book) validates and converts; session.add stages the INSERT; session.commit writes it. The unique ISBN constraint is enforced by the database, so a duplicate surfaces as IntegrityError at commit time. The handler rolls back (the session would otherwise be unusable until rollback()) and answers 409 Conflict. After the commit, session.refresh(db_book) reloads the row so the generated id is available: SQLAlchemy expires every attribute on commit, and refresh is the explicit way to repopulate them. (The status-code and error-handling side of this is on Status Codes & Error Handling.)

Read one, update and delete (listing is covered under pagination below):

@app.get("/books/{book_id}", response_model=BookPublic)
def read_book(book_id: int, session: SessionDep) -> Book:
    book = session.get(Book, book_id)
    if book is None:
        raise HTTPException(status_code=404, detail="Book not found")
    return book


@app.patch("/books/{book_id}", response_model=BookPublic)
def update_book(book_id: int, changes: BookUpdate, session: SessionDep) -> Book:
    book = session.get(Book, book_id)
    if book is None:
        raise HTTPException(status_code=404, detail="Book not found")
    book.sqlmodel_update(changes.model_dump(exclude_unset=True))
    session.add(book)
    try:
        session.commit()
    except IntegrityError:
        session.rollback()
        raise HTTPException(status_code=409, detail="ISBN already exists")
    session.refresh(book)
    return book

Source: FastAPI: SQL Databases, read one hero

@app.delete("/books/{book_id}", status_code=204, dependencies=[Depends(require_admin)])
def delete_book(book_id: int, session: SessionDep) -> Response:
    book = session.get(Book, book_id)
    if book is None:
        raise HTTPException(status_code=404, detail="Book not found")
    session.delete(book)
    session.commit()
    return Response(status_code=204)

Source: FastAPI: SQL Databases, update with HeroUpdate

The PATCH route is the heart of this page:

  1. changes.model_dump(exclude_unset=True) returns a dictionary containing only the fields the client actually sent. Fields that merely hold their default None are excluded. Without exclude_unset, a request that sends only {"year": 1966} would overwrite the title and ISBN with None.

  2. book.sqlmodel_update(data) copies those keys onto the table object. It accepts a dictionary or a Pydantic model and an optional update= dictionary for extra values.

  3. commit() writes only what changed (SQLAlchemy’s unit of work diffs the attributes), and refresh() returns the final state.

PUT and PATCH differ in what an omitted field means. PUT replaces the resource, so omitted fields are reset; PATCH leaves them alone. This route is a PATCH, so {"year": null} does clear the year (sent, and nullable), while {"title": null} is rejected. The generic partial-update discussion is in the official Body - Updates page.

Delete answers 204 No Content; the empty body is why the handler returns a bare Response. Deleting is destructive, so the route requires the admin key (without it: 401). A real service would check a user’s role instead.

A real session against the first version (TestClient; the JSON is exactly what the server returned)
curl -X POST localhost:8000/books -H 'Content-Type: application/json' \
  -d '{"title":"Dune","isbn":"978-0441172719","year":1965}'
curl -X POST localhost:8000/books -H 'Content-Type: application/json' \
  -d '{"title":"Dune again","isbn":"978-0441172719"}'
curl -X PATCH localhost:8000/books/1 -H 'Content-Type: application/json' -d '{"year":1966}'
curl -X PATCH localhost:8000/books/1 -H 'Content-Type: application/json' -d '{"title":null}'
curl -X DELETE localhost:8000/books/1                                   # no key
curl -X DELETE localhost:8000/books/1 -H "X-Admin-Key: $BOOKSHELF_ADMIN_KEY"
201 {"title":"Dune","isbn":"978-0441172719","year":1965,"id":1}
409 {"detail":"ISBN already exists"}
200 {"title":"Dune","isbn":"978-0441172719","year":1966,"id":1}
422 {"detail":[{"type":"value_error","loc":["body","title"],"msg":"Value error, must not be null","input":null,"ctx":{"error":{}}}]}
401 {"detail":"Admin key required"}
204

The leading number is the HTTP status. The same checks are in the test section below as real pytest tests.

Table 1. What the schemas look like in OpenAPI
Component Effect

BookCreate

The request body in Swagger UI: title, ISBN and year with minLength, maxLength, minimum, maximum. No id.

BookPublic

The response schema: the same fields plus a required integer id. internal_notes does not appear.

BookUpdate

The PATCH body: every property optional (anyOf string or null), none required.

BooksPage

The list envelope with items, total, offset, limit.

Book (table model)

Appears in /openapi.json only if some route returns or accepts it without a response_model. Seeing it there is a warning sign.

The component list of the first version was exactly BookCreate, BookPublic, BookUpdate, BooksPage and the two validation-error schemas FastAPI always adds (verified from /openapi.json); see OpenAPI & Interactive Docs.

Pagination with offset and limit

An unbounded GET /books is a denial-of-service waiting to happen. The list route takes offset and limit query parameters, bounds both, and orders the result so pages are stable:

@app.get("/books", response_model=BooksPage)
def list_books(
    session: SessionDep,
    offset: Annotated[int, Query(ge=0)] = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
) -> BooksPage:
    total = session.exec(select(func.count()).select_from(Book)).one()
    books = session.exec(select(Book).order_by(Book.id).offset(offset).limit(limit)).all()
    items = [BookPublic.model_validate(book) for book in books]
    return BooksPage(items=items, total=total, offset=offset, limit=limit)

Source: FastAPI: SQL Databases, read heroes

  • Query(ge=0) and Query(ge=1, le=100) validate in the request: asking for limit=1000 is a 422 (Input should be less than or equal to 100, verified), not a surprise query. This is the same constraint style as the official tutorial’s Query(le=100); see Query Parameters.

  • order_by(Book.id) is not optional. Without an ORDER BY, the database may return rows in any order and the same row can appear on two pages (or on none). Order by a unique, indexed column.

  • select(func.count()).select_from(Book) runs a second query for the total. Counting is a full scan on large tables; drop the total when the client does not need it.

  • Offset pagination gets slower as offset grows, because the database still walks past the skipped rows, and it can skip or repeat rows when data changes between requests. Keyset (cursor) pagination avoids both; the trade-offs are on Pagination: Offset vs. Keyset.

  • Build the envelope with BookPublic.model_validate(book) for every item. Passing table objects straight into an envelope model happened to drop the authors of a nested response in a verified run on the relationships version below.

curl 'localhost:8000/books?offset=1&limit=1'
{"items":[{"title":"Neuromancer","isbn":"978-0441569595","year":1984,"id":2}],"total":2,"offset":1,"limit":1}

Relationships: Authors and Books

A book has several authors and an author several books: a many-to-many relationship, which needs the link table from the ER diagram. SQLModel expresses it with a link model (a table with two foreign keys that form its primary key) and two Relationship attributes that point at each other with back_populates. Reviews are one-to-many: each review has one book, and cascade_delete=True on the parent side removes the reviews when the book is deleted.

The full file below replaces bookshelf_db.py. Models first:

import os
import secrets
from collections.abc import AsyncIterator, Iterator
from contextlib import asynccontextmanager
from typing import Annotated, Generic, TypeVar

from fastapi import Depends, FastAPI, Header, HTTPException, Query, Response
from pydantic import field_validator
from sqlalchemy import event
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import selectinload
from sqlmodel import Field, Relationship, Session, SQLModel, col, create_engine, func, select


class BookAuthorLink(SQLModel, table=True):
    book_id: int | None = Field(default=None, foreign_key="book.id", primary_key=True, ondelete="CASCADE")
    author_id: int | None = Field(default=None, foreign_key="author.id", primary_key=True, ondelete="CASCADE")


class AuthorBase(SQLModel):
    name: str = Field(index=True, min_length=1, max_length=120)
    born: int | None = None


class Author(AuthorBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    books: list["Book"] = Relationship(back_populates="authors", link_model=BookAuthorLink)


class BookBase(SQLModel):
    title: str = Field(index=True, min_length=1, max_length=200)
    isbn: str = Field(unique=True, min_length=10, max_length=17)
    year: int | None = Field(default=None, ge=1400, le=2100)


class Book(BookBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    authors: list[Author] = Relationship(back_populates="books", link_model=BookAuthorLink)
    reviews: list["Review"] = Relationship(back_populates="book", cascade_delete=True)


class ReviewBase(SQLModel):
    rating: int = Field(ge=1, le=5)
    text: str = Field(default="", max_length=2000)


class Review(ReviewBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    book_id: int = Field(foreign_key="book.id", index=True, ondelete="CASCADE")
    book: Book | None = Relationship(back_populates="reviews")


# Data models (no table=True): what the API accepts and returns.
class AuthorCreate(AuthorBase):
    pass


class AuthorPublic(AuthorBase):
    id: int


class BookCreate(BookBase):
    author_ids: list[int] = []


class BookUpdate(SQLModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    isbn: str | None = Field(default=None, min_length=10, max_length=17)
    year: int | None = Field(default=None, ge=1400, le=2100)
    author_ids: list[int] | None = None

    @field_validator("title", "isbn", "author_ids")
    @classmethod
    def not_null(cls, value):
        if value is None:
            raise ValueError("must not be null")
        return value


class BookPublic(BookBase):
    id: int


class BookWithAuthors(BookPublic):
    authors: list[AuthorPublic] = []


class AuthorWithBooks(AuthorPublic):
    books: list[BookPublic] = []


class ReviewCreate(ReviewBase):
    pass


class ReviewPublic(ReviewBase):
    id: int
    book_id: int


T = TypeVar("T")


class Page(SQLModel, Generic[T]):
    items: list[T]
    total: int
    offset: int
    limit: int

Source: SQLModel: Create models with a link table

Details worth knowing:

  • Relationship attributes are not columns and not Pydantic fields on the table model’s database side. Book.authors is a Python list the ORM loads from the database. back_populates keeps both sides in sync when you modify one.

  • link_model=BookAuthorLink tells SQLAlchemy to use that table to join Author and Book. The link table has no id; its two foreign keys are the composite primary key, which also prevents duplicate pairs.

  • ondelete="CASCADE" versus cascade_delete=True. The first is the database foreign-key action (ON DELETE CASCADE in the DDL, visible in the migration below); the second makes the ORM delete the children it has loaded. They solve the same problem at two levels, and this schema declares both. The database action only fires if the database enforces foreign keys, which brings up the SQLite pragma below.

  • Data models stay flat. AuthorPublic has no books and BookPublic has no authors; the nested variants (BookWithAuthors, AuthorWithBooks) each go one level deep. If both pointed at each other, serialization would recurse forever. Decide the direction per endpoint.

  • Page is a generic Pydantic model (Generic[T]), so one envelope serves books and reviews, and OpenAPI gets the specific names Page_BookWithAuthors_ and Page_ReviewPublic_.

The second part is the database setup, which gains one event listener, then the lifespan, the app and the routes:

DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///bookshelf.db")
IS_SQLITE = DATABASE_URL.startswith("sqlite")
engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False} if IS_SQLITE else {},
)

if IS_SQLITE:

    @event.listens_for(engine, "connect")
    def enable_sqlite_foreign_keys(dbapi_connection, connection_record):
        cursor = dbapi_connection.cursor()
        cursor.execute("PRAGMA foreign_keys=ON")  # SQLite ignores FKs unless asked
        cursor.close()


def get_session() -> Iterator[Session]:
    with Session(engine) as session:
        yield session


SessionDep = Annotated[Session, Depends(get_session)]
LimitParam = Annotated[int, Query(ge=1, le=100)]
OffsetParam = Annotated[int, Query(ge=0)]


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    SQLModel.metadata.create_all(engine)
    yield
    engine.dispose()

Source: SQLAlchemy: SQLite dialect, foreign key support

SQLite ignores foreign keys unless every connection runs PRAGMA foreign_keys=ON. The connect event fires for each new DBAPI connection, so the pragma is applied for all of them. PostgreSQL enforces foreign keys always; without the listener a SQLite test database would silently accept orphan rows and skip ON DELETE CASCADE.

app = FastAPI(title="Bookshelf API", lifespan=lifespan)


def require_admin(x_admin_key: Annotated[str | None, Header()] = None) -> None:
    expected = os.environ.get("BOOKSHELF_ADMIN_KEY", "")
    if not expected or not x_admin_key or not secrets.compare_digest(x_admin_key, expected):
        raise HTTPException(status_code=401, detail="Admin key required")


def load_authors(session: Session, author_ids: list[int]) -> list[Author]:
    if not author_ids:
        return []
    authors = session.exec(select(Author).where(col(Author.id).in_(author_ids))).all()
    missing = set(author_ids) - {a.id for a in authors}
    if missing:
        raise HTTPException(status_code=404, detail=f"Unknown author id(s): {sorted(missing)}")
    return list(authors)


@app.post("/authors", response_model=AuthorPublic, status_code=201)
def create_author(author: AuthorCreate, session: SessionDep) -> Author:
    db_author = Author.model_validate(author)
    session.add(db_author)
    session.commit()
    session.refresh(db_author)
    return db_author


@app.get("/authors/{author_id}", response_model=AuthorWithBooks)
def read_author(author_id: int, session: SessionDep) -> Author:
    author = session.get(Author, author_id)
    if author is None:
        raise HTTPException(status_code=404, detail="Author not found")
    return author  # author.books is lazy-loaded while the session is still open

Source: SQLAlchemy: SQLite dialect, foreign key support

@app.post("/books", response_model=BookWithAuthors, status_code=201)
def create_book(book: BookCreate, session: SessionDep) -> Book:
    authors = load_authors(session, book.author_ids)
    db_book = Book.model_validate(book, update={"authors": authors})
    session.add(db_book)
    try:
        session.commit()
    except IntegrityError:
        session.rollback()
        raise HTTPException(status_code=409, detail="ISBN already exists")
    session.refresh(db_book)
    return db_book


@app.get("/books", response_model=Page[BookWithAuthors])
def list_books(
    session: SessionDep,
    offset: OffsetParam = 0,
    limit: LimitParam = 20,
    author_id: int | None = None,
) -> Page[BookWithAuthors]:
    query = select(Book)
    count = select(func.count()).select_from(Book)
    if author_id is not None:
        query = query.join(BookAuthorLink).where(BookAuthorLink.author_id == author_id)
        count = count.join(BookAuthorLink).where(BookAuthorLink.author_id == author_id)
    total = session.exec(count).one()
    books = session.exec(
        query.options(selectinload(Book.authors)).order_by(Book.id).offset(offset).limit(limit)
    ).all()
    items = [BookWithAuthors.model_validate(book) for book in books]
    return Page(items=items, total=total, offset=offset, limit=limit)

Source: SQLModel: Relationships with FastAPI

@app.get("/books/{book_id}", response_model=BookWithAuthors)
def read_book(book_id: int, session: SessionDep) -> Book:
    book = session.get(Book, book_id)
    if book is None:
        raise HTTPException(status_code=404, detail="Book not found")
    return book


@app.patch("/books/{book_id}", response_model=BookWithAuthors)
def update_book(book_id: int, changes: BookUpdate, session: SessionDep) -> Book:
    book = session.get(Book, book_id)
    if book is None:
        raise HTTPException(status_code=404, detail="Book not found")
    data = changes.model_dump(exclude_unset=True)
    author_ids = data.pop("author_ids", None)
    book.sqlmodel_update(data)
    if author_ids is not None:
        book.authors = load_authors(session, author_ids)
    session.add(book)
    try:
        session.commit()
    except IntegrityError:
        session.rollback()
        raise HTTPException(status_code=409, detail="ISBN already exists")
    session.refresh(book)
    return book


@app.delete("/books/{book_id}", status_code=204, dependencies=[Depends(require_admin)])
def delete_book(book_id: int, session: SessionDep) -> Response:
    book = session.get(Book, book_id)
    if book is None:
        raise HTTPException(status_code=404, detail="Book not found")
    session.delete(book)  # reviews and link rows go with it
    session.commit()
    return Response(status_code=204)


@app.post("/books/{book_id}/reviews", response_model=ReviewPublic, status_code=201)
def create_review(book_id: int, review: ReviewCreate, session: SessionDep) -> Review:
    if session.get(Book, book_id) is None:
        raise HTTPException(status_code=404, detail="Book not found")
    db_review = Review.model_validate(review, update={"book_id": book_id})
    session.add(db_review)
    session.commit()
    session.refresh(db_review)
    return db_review


@app.get("/books/{book_id}/reviews", response_model=Page[ReviewPublic])
def list_reviews(
    book_id: int, session: SessionDep, offset: OffsetParam = 0, limit: LimitParam = 20
) -> Page[ReviewPublic]:
    where = col(Review.book_id) == book_id
    total = session.exec(select(func.count()).select_from(Review).where(where)).one()
    reviews = session.exec(select(Review).where(where).order_by(Review.id).offset(offset).limit(limit)).all()
    items = [ReviewPublic.model_validate(review) for review in reviews]
    return Page(items=items, total=total, offset=offset, limit=limit)

Source: SQLAlchemy: Relationship Loading Techniques

How the routes use the relationship API:

  • Attaching authors at creation. load_authors fetches the requested authors with col(Author.id).in_(…​) (col() tells type checkers that this attribute is a column) and answers 404 for unknown ids. Book.model_validate(book, update={"authors": authors}) validates the body and injects the loaded Author objects into the relationship; the link rows are written at commit.

  • Replacing authors with PATCH. author_ids is popped from the update dictionary before sqlmodel_update (it is not a Book column) and assigned as book.authors = …​; SQLAlchemy computes which link rows to insert and delete.

  • Filtering through the link. ?author_id=2 joins BookAuthorLink; the join condition is inferred from the foreign keys.

  • Eager loading. selectinload(Book.authors) loads the authors of all returned books with one extra query. Without it, building each BookWithAuthors touches book.authors and triggers one lazy query per book: the N+1 problem (explained for the JVM on Fetching and the N+1 Problem; the mechanics are the same). A counted run of GET /books with five books issued 3 statements with selectinload (count, books, authors) and 7 with lazy loading (count, books, then one per book).

  • Lazy loading works in read_book only because the session is still open and the driver is synchronous. The lazy access runs in the threadpool inside the open session. In the async variant it fails, as shown next.

  • Deleting. session.delete(book) removes the book, its link rows and its reviews. In the verified run the review list of a deleted book was empty afterwards and the author survived with an empty books list.

The relationships version (TestClient run)
201 {"name":"Frank Herbert","born":1920,"id":1}
201 {"title":"Dune","isbn":"978-0441172719","year":1965,"id":1,"authors":[{"name":"Frank Herbert","born":1920,"id":1}]}
404 {"detail":"Unknown author id(s): [42]"}
200 {"items":[{"title":"Hunters of Dune","isbn":"978-0765347152","year":2006,"id":2,"authors":[{"name":"Frank Herbert","born":1920,"id":1},{"name":"Brian Herbert","born":1947,"id":2}]}],"total":1,"offset":0,"limit":20}
200 {"name":"Frank Herbert","born":1920,"id":1,"books":[{"title":"Dune","isbn":"978-0441172719","year":1965,"id":1},{"title":"Hunters of Dune","isbn":"978-0765347152","year":2006,"id":2}]}
201 {"rating":5,"text":"Spice!","id":1,"book_id":1}
422 {"detail":[{"type":"less_than_equal","loc":["body","rating"],"msg":"Input should be less than or equal to 5","input":9,"ctx":{"le":5}}]}
204
200 {"items":[],"total":0,"offset":0,"limit":20}

The lines are, in order: create author, create book with author 1, create book with unknown author 42, filter by author 2, read author 1 with books, create a review, an invalid rating, delete (with admin key), and the review list after deleting.

Session Lifecycle, Transactions and def versus async def

A session is a transaction scope. The rules that avoid most database bugs in FastAPI:

  • One session per request, created by the dependency. Never a module-level session, and never one shared between threads or tasks.

  • Commit in the path operation. Since FastAPI 0.118.0 (2025-09-29) the teardown of a default-scope yield dependency runs after the response has been sent, so a commit placed after yield cannot change the status code and the client may see 201 for a write that failed. Commit before returning, and let the dependency only close the session (see Dependencies with yield for scope="function").

  • Roll back on failure. After an exception inside a transaction (for example the IntegrityError above) call session.rollback() before the session is reused. Closing the session also discards uncommitted work.

  • expire_on_commit and refresh. By default a commit expires all loaded attributes, so the next access reloads the row (a lazy query). That is why the routes call refresh() before returning, and why the async session below sets expire_on_commit=False. The relational model of transactions is on Transactions.

  • def routes with a synchronous driver. FastAPI runs def dependencies and routes in a threadpool (40 worker threads by default in AnyIO, verified), so a blocking driver such as psycopg or sqlite3 is fine and does not block the event loop. Do not call a synchronous session from an async def route: the call blocks the loop. See Async & Concurrency, in particular its sections on blocking calls and tuning.

  • The pool must be sized with the threadpool in mind. Up to 40 threads can ask for a connection at once, while the default SQLAlchemy pool offers 5 plus 10 overflow (below).

Testing with a SQLite Override

The session dependency is the single seam through which every route reaches the database, which makes it the thing to override in tests: app.dependency_overrides[get_session] swaps in a session bound to a throwaway in-memory database. The in-memory URL (sqlite://) with StaticPool and check_same_thread=False gives every thread the same single connection, and therefore the same tables. This keeps tests fast and independent of a server; run a small set of tests against real PostgreSQL as well, because SQLite is more lenient (types, foreign keys, ILIKE, concurrency).

from collections.abc import Iterator

import pytest
from fastapi.testclient import TestClient
from sqlalchemy.pool import StaticPool
from sqlmodel import Session, SQLModel, create_engine

from bookshelf_db import app, get_session


@pytest.fixture(name="session")
def session_fixture() -> Iterator[Session]:
    engine = create_engine(
        "sqlite://",  # a private in-memory database per test
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,  # one shared connection, or each connection would see its own empty DB
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session


@pytest.fixture(name="client")
def client_fixture(session: Session) -> Iterator[TestClient]:
    app.dependency_overrides[get_session] = lambda: session
    # No "with": the lifespan (and its create_all on the real engine) is not run.
    yield TestClient(app)
    app.dependency_overrides.clear()


def test_create_and_read_book(client: TestClient):
    created = client.post("/books", json={"title": "Dune", "isbn": "978-0441172719"})
    assert created.status_code == 201
    book_id = created.json()["id"]
    assert client.get(f"/books/{book_id}").json()["title"] == "Dune"


def test_duplicate_isbn_is_a_conflict(client: TestClient):
    body = {"title": "Dune", "isbn": "978-0441172719"}
    assert client.post("/books", json=body).status_code == 201
    assert client.post("/books", json=body).status_code == 409


def test_patch_changes_only_sent_fields(client: TestClient):
    book_id = client.post("/books", json={"title": "Dune", "isbn": "978-0441172719", "year": 1965}).json()["id"]
    changed = client.patch(f"/books/{book_id}", json={"year": 1966}).json()
    assert changed["title"] == "Dune" and changed["year"] == 1966

Source: FastAPI: Testing a Database

Run with pytest -q test_books.py: 3 passed on the verified run. TestClient(app) is created without a with block here on purpose: the lifespan (and its create_all on the real engine) does not run, and the test fixture created the tables itself. Use with TestClient(app) in tests that should run the lifespan. More on fixtures, overrides and async tests is on Testing, and the override mechanism is on Dependency Injection.

Async Sessions: AsyncSession with asyncpg

A synchronous session in a threadpool scales well enough for many services. The alternative is an async stack: create_async_engine, AsyncSession and an async driver, with async def dependencies and routes, so database waits do not occupy a thread. SQLAlchemy 2.0 ships the asyncio extension; SQLModel adds a subclass, sqlmodel.ext.asyncio.session.AsyncSession, whose exec() is awaitable and returns the same scalar results as the synchronous one. (Plain SQLAlchemy’s AsyncSession has no exec; use execute() plus scalars() there.)

The driver changes with the URL. For PostgreSQL: a postgresql+asyncpg:// URL (user, secret password, host, port and database name), built from settings, never typed into source. For SQLite in tests: sqlite+aiosqlite:///file.db. The example runs unchanged on both by reading DATABASE_URL; it was executed with aiosqlite only, because no PostgreSQL server was available. Creating the asyncpg engine (no connection is made at that point) succeeds and uses an AsyncAdaptedQueuePool.

import os
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Query
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import selectinload
from sqlmodel import Field, Relationship, SQLModel, select
from sqlmodel.ext.asyncio.session import AsyncSession

# in production: a postgresql+asyncpg:// URL built from settings, never typed into source
DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite+aiosqlite:///bookshelf_async.db")
engine = create_async_engine(DATABASE_URL, pool_pre_ping=True)


class BookBase(SQLModel):
    title: str = Field(index=True, min_length=1, max_length=200)
    isbn: str = Field(unique=True, min_length=10, max_length=17)


class Book(BookBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    reviews: list["Review"] = Relationship(back_populates="book")


class Review(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    book_id: int = Field(foreign_key="book.id", index=True)
    rating: int = Field(ge=1, le=5)
    book: Book | None = Relationship(back_populates="reviews")


class BookCreate(BookBase):
    pass


class ReviewPublic(SQLModel):
    id: int
    rating: int


class BookPublic(BookBase):
    id: int


class BookWithReviews(BookPublic):
    reviews: list[ReviewPublic] = []


async def get_session() -> AsyncIterator[AsyncSession]:
    async with AsyncSession(engine, expire_on_commit=False) as session:
        yield session


SessionDep = Annotated[AsyncSession, Depends(get_session)]


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    yield
    await engine.dispose()


app = FastAPI(title="Bookshelf API (async)", lifespan=lifespan)

Source: SQLAlchemy: Asynchronous I/O (asyncio)

@app.post("/books", response_model=BookPublic, status_code=201)
async def create_book(book: BookCreate, session: SessionDep) -> Book:
    db_book = Book.model_validate(book)
    session.add(db_book)
    try:
        await session.commit()
    except IntegrityError:
        await session.rollback()
        raise HTTPException(status_code=409, detail="ISBN already exists")
    await session.refresh(db_book)
    return db_book


@app.post("/books/{book_id}/reviews", response_model=ReviewPublic, status_code=201)
async def create_review(book_id: int, rating: Annotated[int, Query(ge=1, le=5)], session: SessionDep) -> Review:
    if await session.get(Book, book_id) is None:
        raise HTTPException(status_code=404, detail="Book not found")
    review = Review(book_id=book_id, rating=rating)
    session.add(review)
    await session.commit()
    return review  # safe to read after commit because of expire_on_commit=False


@app.get("/books", response_model=list[BookPublic])
async def list_books(
    session: SessionDep,
    offset: Annotated[int, Query(ge=0)] = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
):
    result = await session.exec(select(Book).order_by(Book.id).offset(offset).limit(limit))
    return result.all()


@app.get("/books/{book_id}", response_model=BookWithReviews)
async def read_book(book_id: int, session: SessionDep) -> Book:
    result = await session.exec(
        select(Book).where(Book.id == book_id).options(selectinload(Book.reviews))
    )
    book = result.first()
    if book is None:
        raise HTTPException(status_code=404, detail="Book not found")
    return book


@app.get("/lazy-mistake/{book_id}", response_model=BookWithReviews)
async def lazy_mistake(book_id: int, session: SessionDep) -> Book:
    book = await session.get(Book, book_id)  # no selectinload: book.reviews is lazy
    if book is None:
        raise HTTPException(status_code=404, detail="Book not found")
    return book  # serialising book.reviews would need blocking I/O inside the event loop

Source: SQLModel: FastAPI and Pydantic tutorial

What changes compared with the synchronous version:

  • Everything that touches the database is awaited: await session.commit(), await session.refresh(…​), await session.get(…​), await session.exec(…​). Forgetting an await returns a coroutine and a RuntimeWarning: coroutine …​ was never awaited.

  • create_all is synchronous DDL, so it runs through conn.run_sync(SQLModel.metadata.create_all) inside engine.begin(), and the shutdown awaits engine.dispose().

  • expire_on_commit=False on the session. After await session.commit(), reading an expired attribute would require I/O, which an async session cannot do implicitly. With False, the committed object stays readable (as in create_review, which returns it without a refresh). The trade-off: the object can be stale if the database changed it (defaults, triggers); await session.refresh(obj) reloads it explicitly when that matters.

  • No implicit lazy loading. Eager-load every relationship you will serialize (selectinload, joinedload). The deliberately wrong lazy-mistake route fails in a verified run with a ResponseValidationError wrapping MissingGreenlet: greenlet_spawn has not been called; can’t call await_only() here. Was IO attempted in an unexpected place?. That message always means "lazy load in async code": fix the query, not the exception.

  • pool_pre_ping=True. The pool tests each connection with a cheap ping when it is checked out, so connections dropped by the server or a firewall are replaced silently. It costs a round trip per checkout.

  • Do not mix. An async def route with a synchronous Session blocks the event loop; a def route with an AsyncSession cannot await. Pick one style per application, or per router.

Choose async when the application is already async end to end (WebSockets, streaming, many concurrent slow calls). The simpler synchronous stack is a completely valid choice that the official tutorial uses.

Alembic Migrations

create_all cannot change an existing table: add a column to Book and the database keeps the old shape. Alembic is SQLAlchemy’s migration tool: it compares your models with the database, writes a versioned script, and applies scripts in order. SQLModel’s metadata is SQLAlchemy metadata, so Alembic works with it directly. The general question of choosing a migration tool across languages is on Schema Evolution.

alembic init migrations                                   # creates alembic.ini and migrations/
# edit migrations/env.py and script.py.mako as below
export DATABASE_URL=sqlite:///migrated.db                 # or the PostgreSQL URL
alembic revision --autogenerate -m "create bookshelf tables"
alembic upgrade head

Two edits make Alembic understand SQLModel. In env.py, point target_metadata at SQLModel.metadata, import the module that defines the models (importing registers the tables), and read the URL from the environment or the settings instead of alembic.ini:

import os
from logging.config import fileConfig

from sqlalchemy import engine_from_config
from sqlalchemy import pool

from alembic import context
from sqlmodel import SQLModel

from db_settings import get_db_settings
import bookshelf_rel  # noqa: F401  (importing the models registers their tables on SQLModel.metadata)

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
# The URL comes from the environment or from the settings (password from a secret file), never from alembic.ini.
url = os.environ.get("DATABASE_URL") or get_db_settings().url.render_as_string(hide_password=False)
config.set_main_option("sqlalchemy.url", url.replace("%", "%%"))  # ConfigParser treats % specially

Source: Alembic: Tutorial

def run_migrations_online() -> None:
    """Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.

    """
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,
            render_as_batch=connection.dialect.name == "sqlite",  # SQLite cannot ALTER most things in place
        )

        with context.begin_transaction():
            context.run_migrations()

Source: Alembic: Auto Generating Migrations

And in script.py.mako, add import sqlmodel next to import sqlalchemy as sa: SQLModel maps str columns to sqlmodel.sql.sqltypes.AutoString, and generated scripts refer to it. Without the import, alembic upgrade fails with NameError: name 'sqlmodel' is not defined.

Notes on the edits:

  • compare_type=True makes autogenerate notice changed column types, not just added and removed columns and tables.

  • render_as_batch is turned on for SQLite only: SQLite cannot alter most column properties in place, and Alembic’s batch mode recreates the table. PostgreSQL does not need it.

  • url.replace("%", "%%") is required because Alembic stores the URL in a ConfigParser, where % introduces an interpolation. A URL with a percent-encoded password (%40) otherwise fails with invalid interpolation syntax (reproduced).

  • Alembic runs migrations over a normal synchronous engine, so it also works for an application that uses asyncpg at runtime, as long as a synchronous driver (psycopg) is installed for the migration URL. Alembic also ships an async template if you prefer to migrate over the async driver.

The first autogenerated revision, abridged (SQLite run), shows the constraints from the models: indexes, the unique ISBN, and ON DELETE CASCADE on the three foreign keys.

def upgrade() -> None:
    op.create_table('author',
    sa.Column('name', sqlmodel.sql.sqltypes.AutoString(length=120), nullable=False),
    sa.Column('born', sa.Integer(), nullable=True),
    sa.Column('id', sa.Integer(), nullable=False),
    sa.PrimaryKeyConstraint('id')
    )
    ...
    op.create_table('bookauthorlink',
    sa.Column('book_id', sa.Integer(), nullable=False),
    sa.Column('author_id', sa.Integer(), nullable=False),
    sa.ForeignKeyConstraint(['author_id'], ['author.id'], ondelete='CASCADE'),
    sa.ForeignKeyConstraint(['book_id'], ['book.id'], ondelete='CASCADE'),
    sa.PrimaryKeyConstraint('book_id', 'author_id')
    )

Source: Alembic: Auto Generating Migrations

Now evolve the model: add pages: int | None = Field(default=None, ge=1) to BookBase, then autogenerate again. Alembic detects the single added column and writes a small, reviewable script:

INFO  [alembic.autogenerate.compare.tables] Detected added column 'book.pages'
INFO  [alembic.runtime.migration] Running upgrade 1b6b67ca6623 -> c737fda8a48b, add pages to book
def upgrade() -> None:
    with op.batch_alter_table('book', schema=None) as batch_op:
        batch_op.add_column(sa.Column('pages', sa.Integer(), nullable=True))


def downgrade() -> None:
    with op.batch_alter_table('book', schema=None) as batch_op:
        batch_op.drop_column('pages')

Source: Alembic: Auto Generating Migrations

The everyday commands, all of which were run on SQLite:

Command What it does

alembic revision --autogenerate -m "message"

Compares the models with the database and writes a script. Always read it: autogenerate misses renames (it sees a drop and an add), some constraints and server defaults.

alembic upgrade head

Applies all pending revisions.

alembic downgrade -1

Reverts the latest revision using its downgrade().

alembic current and alembic history

Show the database’s revision and the chain (1b6b67ca6623 → c737fda8a48b (head)).

alembic upgrade head --sql

Prints the SQL instead of running it (offline mode), for review or for a DBA.

When Alembic owns the schema, remove create_all from the lifespan (or restrict it to tests), so the application does not race the migrations. Run alembic upgrade head as a deployment step before the new application version starts; the Compose file below does that with a one-shot service. The PostgreSQL-specific run was not executed live, but the migration scripts use only portable operations.

PostgreSQL in Docker Compose

SQLite is for tests and prototypes; production is a server database. Compose runs PostgreSQL next to the API for local development. The credential rule is the same as everywhere: the password lives in a file that is not committed and that Compose mounts as a secret, never in the YAML. Create it with a generated value, for example mkdir -p secrets && openssl rand -hex 24 > secrets/bookshelf_db_password.txt, and git-ignore secrets/.

services:
  db:
    image: postgres:18
    restart: unless-stopped
    environment:
      POSTGRES_USER: bookshelf
      POSTGRES_DB: bookshelf
      POSTGRES_PASSWORD_FILE: /run/secrets/bookshelf_db_password
    secrets:
      - bookshelf_db_password
    volumes:
      - pgdata:/var/lib/postgresql  # PostgreSQL 18+ images keep data in a versioned subdirectory
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U bookshelf -d bookshelf"]
      interval: 5s
      timeout: 3s
      retries: 10
    ports:
      - "127.0.0.1:5432:5432"  # development only: remove to keep the database private to the network

  migrate:
    build: .
    command: ["alembic", "upgrade", "head"]
    environment:
      BOOKSHELF_DB_HOST: db
    secrets:
      - bookshelf_db_password
    depends_on:
      db:
        condition: service_healthy

  api:
    build: .
    command: ["fastapi", "run", "main.py", "--port", "8000"]
    environment:
      BOOKSHELF_DB_HOST: db
    secrets:
      - bookshelf_db_password
    ports:
      - "8000:8000"
    depends_on:
      migrate:
        condition: service_completed_successfully

secrets:
  bookshelf_db_password:
    file: ./secrets/bookshelf_db_password.txt  # git-ignored; create it with a generated value

volumes:
  pgdata:

Source: Docker: Compose secrets

  • The official image reads the password from the file named by POSTGRES_PASSWORD_FILE; Compose mounts the secret at /run/secrets/bookshelf_db_password. POSTGRES_USER and POSTGRES_DB are not secret.

  • The healthcheck (pg_isready) makes service_healthy meaningful, so migrations start only when the server accepts connections. migrate is a one-shot service: the API starts once it ends with service_completed_successfully.

  • PostgreSQL 18 images keep data in a version-named subdirectory and declare the volume at /var/lib/postgresql, so the named volume mounts there (older guides mount /var/lib/postgresql/data). Mounting the wrong path means the data is not kept in the volume you think it is.

  • Publishing 5432 only on 127.0.0.1 is for local tools such as psql. Do not publish it in production.

  • The Dockerfile and fastapi run options are on Deployment; Compose itself is covered on Docker Compose Fundamentals and Compose for Local Development.

The application reads the same secret through pydantic-settings, with a secrets_dir pointing at /run/secrets: the file named bookshelf_db_password fills the password field of a settings class with prefix BOOKSHELF_DB_. The class builds the URL with URL.create, which percent-escapes special characters in the password for you, and the password type is SecretStr, so it never appears in a log or a repr.

from functools import lru_cache

from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict
from sqlalchemy import URL


class DatabaseSettings(BaseSettings):
    # BOOKSHELF_DB_HOST=..., or a file /run/secrets/bookshelf_db_password for the password
    model_config = SettingsConfigDict(env_prefix="BOOKSHELF_DB_", secrets_dir="/run/secrets")

    host: str = "localhost"
    port: int = 5432
    user: str = "bookshelf"
    name: str = "bookshelf"
    password: SecretStr  # required: no default, so a missing secret fails at startup

    @property
    def url(self) -> URL:
        # URL.create() escapes special characters in the password for you
        return URL.create(
            "postgresql+psycopg",
            username=self.user,
            password=self.password.get_secret_value(),
            host=self.host,
            port=self.port,
            database=self.name,
        )


@lru_cache
def get_db_settings() -> DatabaseSettings:
    return DatabaseSettings()

Source: Pydantic: Settings Management

The engine is then built from the settings instead of DATABASE_URL:

engine = create_engine(
    get_db_settings().url,
    pool_size=10,        # connections kept open
    max_overflow=5,      # extra connections allowed under bursts
    pool_timeout=30,     # seconds to wait for a free connection
    pool_recycle=1800,   # replace connections older than 30 minutes
    pool_pre_ping=True,  # test a connection before handing it out
)

Source: Pydantic: Settings Management

A verified run read the password from a secrets directory, printed SecretStr('') for the field and postgresql+psycopg://bookshelf:**@db:5432/bookshelf for the URL with the password hidden, and failed with password: Field required when no secret existed (no default means a missing secret stops the application at startup). Building this engine does not connect, so it ran without a server; docker compose config accepted the Compose file; docker compose up itself was *not run live because no container daemon was running. Settings, .env files and secrets directories in depth are on Settings, Lifespan & Background Tasks. If you later add vector search on the same server, see PostgreSQL pgvector.

For production, a managed PostgreSQL service replaces the db container and the password comes from the platform’s secret store; the migration step and the pool settings stay the same.

Plain SQLAlchemy or Raw DB-API Instead

SQLModel is optional. Everything above (session per request through a yield dependency, explicit commit, Pydantic schemas at the edge) works with the libraries underneath.

SQLAlchemy 2.0 ORM with separate Pydantic schemas. The table is a typed Mapped class; the API models are ordinary Pydantic models. ConfigDict(from_attributes=True) lets a response model read attributes from an ORM object, replacing Pydantic v1’s orm_mode:

import os
from collections.abc import AsyncIterator, Iterator
from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import String, create_engine, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, sessionmaker


class Base(DeclarativeBase):
    pass


class BookRow(Base):
    __tablename__ = "book"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), index=True)
    isbn: Mapped[str] = mapped_column(String(17), unique=True)
    year: Mapped[int | None]


class BookIn(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    isbn: str = Field(min_length=10, max_length=17)
    year: int | None = None


class BookOut(BookIn):
    model_config = ConfigDict(from_attributes=True)  # read attributes from ORM objects
    id: int


engine = create_engine(os.environ.get("DATABASE_URL", "sqlite:///plain.db"))
SessionLocal = sessionmaker(engine, expire_on_commit=False)


def get_session() -> Iterator[Session]:
    with SessionLocal() as session:
        yield session


SessionDep = Annotated[Session, Depends(get_session)]


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    Base.metadata.create_all(engine)
    yield
    engine.dispose()


app = FastAPI(lifespan=lifespan)


@app.post("/books", response_model=BookOut, status_code=201)
def create_book(book: BookIn, session: SessionDep) -> BookRow:
    row = BookRow(**book.model_dump())
    session.add(row)
    try:
        session.commit()
    except IntegrityError:
        session.rollback()
        raise HTTPException(status_code=409, detail="ISBN already exists")
    return row


@app.get("/books", response_model=list[BookOut])
def list_books(session: SessionDep, offset: int = 0, limit: int = 20):
    return session.scalars(select(BookRow).order_by(BookRow.id).offset(offset).limit(limit)).all()

Source: SQLAlchemy: ORM Quick Start

The routes are the same as in the SQLModel version, except session.scalars(select(…​)) replaces session.exec. A run returned 201 for the first book, 409 {"detail":"ISBN already exists"} for the duplicate, and the list with the one book. sessionmaker(engine, expire_on_commit=False) is a factory that keeps committed objects readable, so the POST route needs no refresh.

SQLAlchemy Core. No ORM and no mapped classes: tables, SQL expressions and result rows. Useful for bulk and reporting queries:

from sqlalchemy import Column, Integer, MetaData, String, Table, create_engine, insert, select

metadata = MetaData()  # note the capital D: sqlalchemy.MetaData
books = Table(
    "book",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("title", String(200), nullable=False),
    Column("isbn", String(17), nullable=False, unique=True),
)

engine = create_engine("sqlite://")
metadata.create_all(engine)

with engine.begin() as conn:  # begin() commits on success, rolls back on error
    conn.execute(insert(books), [{"title": "Dune", "isbn": "978-0441172719"},
                                 {"title": "Neuromancer", "isbn": "978-0441569595"}])
    rows = conn.execute(select(books.c.id, books.c.title).order_by(books.c.id)).mappings().all()

print([dict(r) for r in rows])

Source: SQLAlchemy: Unified Tutorial

The output was [{'id': 1, 'title': 'Dune'}, {'id': 2, 'title': 'Neuromancer'}]. engine.begin() opens a transaction and commits on success or rolls back on error. The class is MetaData with a capital D: a widely copied older example imports a non-existent Metadata from SQLAlchemy and fails with ImportError (reproduced; the error message even suggests MetaData).

Raw DB-API (sqlite3, psycopg). You write the SQL yourself (the statements are covered on DML: Queries and DML: Modifications, the schema on DDL). The standard is PEP 249: connections, cursors, execute, commit, parameter placeholders. In FastAPI the same rules apply: a connection per request from a yield dependency, parameters passed separately, no global connection. Older guides commonly open one module-level sqlite3 connection and share it across all requests; a shared connection serializes (or breaks) concurrent requests and cannot be used from the threadpool without disabling SQLite’s thread check. The corrected version:

import os
import sqlite3
from collections.abc import Iterator
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel, Field

DB_PATH = os.environ.get("BOOKSHELF_DB", "raw.db")

SCHEMA = """
CREATE TABLE IF NOT EXISTS book (
    id INTEGER PRIMARY KEY,
    title TEXT NOT NULL,
    isbn TEXT NOT NULL UNIQUE
)
"""


def get_connection() -> Iterator[sqlite3.Connection]:
    # One connection per request, never a module-level global.
    # check_same_thread=False: a def dependency and a def path operation can run in different worker threads.
    conn = sqlite3.connect(DB_PATH, check_same_thread=False)
    conn.row_factory = sqlite3.Row
    try:
        yield conn  # path operations commit explicitly, before the response is sent
    finally:
        conn.close()


ConnDep = Annotated[sqlite3.Connection, Depends(get_connection)]
app = FastAPI()

with sqlite3.connect(DB_PATH) as _setup:  # simple bootstrap; use migrations in real projects
    _setup.execute(SCHEMA)


class BookIn(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    isbn: str = Field(min_length=10, max_length=17)


class BookOut(BookIn):
    id: int


@app.post("/books", status_code=201)
def create_book(book: BookIn, conn: ConnDep) -> BookOut:
    try:
        cur = conn.execute("INSERT INTO book (title, isbn) VALUES (?, ?)", (book.title, book.isbn))
    except sqlite3.IntegrityError:
        raise HTTPException(status_code=409, detail="ISBN already exists")
    conn.commit()
    return BookOut(id=cur.lastrowid, **book.model_dump())


@app.get("/books/search")
def search_books(title: str, conn: ConnDep) -> list[BookOut]:
    # Parameters travel separately from the SQL text: never build SQL with f-strings.
    rows = conn.execute("SELECT id, title, isbn FROM book WHERE title LIKE ?", (f"%{title}%",)).fetchall()
    return [BookOut(**dict(row)) for row in rows]

Source: Python: sqlite3

  • Parameters, never string building. WHERE title LIKE ? with a separate tuple lets the driver escape values. The verified request title=' OR 1=1 -- returned an empty list, not the table. f"…​ {title}" is SQL injection.

  • check_same_thread=False is deliberate here (next section): a def dependency and a def path operation can run in different worker threads, and sqlite3 refuses to use a connection in a thread other than the one that created it. Because each request owns its own connection, nothing is shared concurrently, so turning the check off is safe.

  • Commit in the route, for the same teardown-timing reason as with sessions.

  • PostgreSQL’s DB-API drivers use %s placeholders (psycopg) instead of ? and offer their own pools; the shape of the dependency is identical.

Connection Pooling and check_same_thread

Opening a database connection is slow (a network round trip, authentication, TLS). SQLAlchemy’s engine keeps connections in a pool and lends one to each session until it closes. The pool class depends on the database, as observed:

URL Pool class Notes

postgresql+psycopg://…​

QueuePool

Defaults: pool_size=5, max_overflow=10, pool_timeout=30, pool_recycle=-1 (never), pool_pre_ping=False

postgresql+asyncpg://…​

AsyncAdaptedQueuePool

The same options, used with create_async_engine

sqlite:///file.db

QueuePool

A file database pools like any other (SQLAlchemy 2.0)

sqlite:// (in memory)

SingletonThreadPool

One connection per thread, so each thread sees a different empty database

Tune with create_engine arguments: pool_size (connections kept), max_overflow (temporary extra connections), pool_timeout (seconds a request waits), pool_recycle (retire old connections, useful behind firewalls or proxies that drop idle ones), pool_pre_ping (detect dead connections at checkout). The total across all processes must stay below the server’s max_connections: with 4 worker processes and pool_size=10, max_overflow=5 the API can open up to 60 connections. When the pool is exhausted, new requests wait pool_timeout seconds and then fail:

from sqlalchemy import create_engine
from sqlalchemy.exc import TimeoutError as PoolTimeout

# Rarely what you want in production; tiny numbers make the limit visible.
engine = create_engine("sqlite:///pool_demo.db", pool_size=1, max_overflow=0, pool_timeout=1)

first = engine.connect()  # takes the only pooled connection
try:
    engine.connect()  # a second request waits pool_timeout seconds, then gives up
except PoolTimeout as exc:
    print(str(exc).split(" (Background")[0])
finally:
    first.close()

# Defaults for QueuePool: pool_size=5, max_overflow=10, pool_timeout=30, pool_recycle=-1, pool_pre_ping=False
default = create_engine("sqlite:///pool_demo.db")
print(type(default.pool).__name__, default.pool.size())
print(type(create_engine("sqlite://").pool).__name__)  # in-memory SQLite

Source: SQLAlchemy: Connection Pooling

The output was QueuePool limit of size 1 overflow 0 reached, connection timed out, timeout 1.00, then QueuePool 5 and SingletonThreadPool. In FastAPI this appears as 500 errors under load when sessions are held too long (a slow call inside the session, a streaming response, or a leaked session). Close sessions promptly, keep the pool at least as large as the number of concurrent requests that need the database, and size it against the server limit. Use NullPool (no pooling) for migration commands and short-lived scripts.

check_same_thread explained. The sqlite3 module raises ProgrammingError: SQLite objects created in a thread can only be used in that same thread when a connection crosses threads. FastAPI crosses threads routinely: it runs def dependencies and def routes in a worker threadpool, so the thread that opened the connection may not be the one that uses it. A verified reproduction: an async def dependency (event-loop thread) opens a raw connection and a def route (worker thread) uses it, which returned 500 with exactly that error. The official tutorial passes connect_args={"check_same_thread": False} for this reason.

Two facts refine that advice, both observed with SQLAlchemy 2.0.54:

  • For a SQLite file database, SQLAlchemy 2.0’s QueuePool already hands connections across threads without the argument (the cross-thread query in a test returned a normal result). The tutorial’s argument is harmless and keeps older versions and in-memory setups working.

  • For the in-memory test database (StaticPool), the argument is required: without it, a second thread failed with ProgrammingError; with it, the query returned 1. That is why the test fixture above sets it.

Only SQLite takes this argument; it is passed conditionally in the examples so that the same code works against PostgreSQL. Disabling the check is safe when no two threads use the same connection at the same time, which the pool and the per-request session guarantee.

Where Older Guides Are Outdated, and What Changed Recently

  • The official SQL tutorial now uses SQLModel (0.115.1, 2024-10-12). Older tutorials and books show SQLAlchemy models, Pydantic schemas and a global connection or scoped_session. They still work, but the documented approach is a session dependency with yield and SQLModel with SQLAlchemy 2.0 underneath.

  • SQLAlchemy 1.x idioms. declarative_base(), Column(…​) annotations, session.query(…​) and engine.execute(…​) are the 1.x style. The 2.0 style uses DeclarativeBase, Mapped/mapped_column, select() with session.scalars() or session.exec(), and Connection.execute() inside engine.begin().

  • Pydantic v1 names. orm_mode is from_attributes, .dict() is model_dump(), parse_obj is model_validate. FastAPI dropped Pydantic v1 in 0.126.0 (2025-12-20) and pydantic.v1 in 0.128.0 (2025-12-27).

  • Startup and shutdown events. Table creation in @app.on_event("startup") is the older form (even the official tutorial page still shows it). Lifespan context managers have been the supported way since 0.93.0 (2023-03-07).

  • Yield dependency timing. Guides that commit in the teardown, or assume the session stays open for background tasks, predate 0.106.0 (2023-12-25) and 0.118.0 (2025-09-29), which changed when the teardown runs; 0.121.0 (2025-11-03) added Depends(scope="function"). Commit in the route.

  • A global sqlite3 connection and environment-variable switching of storage modules. Replaced by a per-request connection or session from a dependency, with dependency_overrides for tests.

  • Incorrect imports. sqlalchemy.Metadata does not exist (it is MetaData); copy-paste from sources that were not executed is where such errors come from.

  • Python 3.14 and SQLModel. SQLModel 0.0.47 declares Python 3.10 and later; the examples here ran on 3.13 only.

  • SQLAlchemy 2.1. It is released, but SQLModel 0.0.47 pins <2.1; stay on 2.0 until SQLModel lifts the pin. The release history of FastAPI itself is collected on What’s Changed & Migration.

Pitfalls

  • Returning the table model. It exposes every column and publishes a Book schema. Always use a data model as the response_model.

  • Trusting validation on a table model. Book(…​) does not validate. Validate a data model, then convert with model_validate.

  • exclude_unset forgotten in PATCH. Every omitted field becomes None and wipes data. Also reject explicit null for NOT NULL columns.

  • Catching every IntegrityError as "ISBN exists". NOT NULL and foreign-key violations raise the same exception. Inspect the constraint name, or check the condition explicitly (the author lookup above returns 404 before the commit for that reason).

  • Forgetting rollback(). After a failed flush the session is in a failed state and the next use raises PendingRollbackError.

  • Committing in the dependency teardown. After 0.118.0 it runs after the response is sent.

  • Accessing a lazy relationship or an expired attribute after the session closes. DetachedInstanceError: eager-load what you serialize, and serialize inside the route.

  • Lazy loading in a loop (N+1). Three statements became seven for five books. selectinload for collections, joinedload for single-valued relationships.

  • Lazy loading in async code. MissingGreenlet; eager-load.

  • A blocking driver inside async def. It freezes the event loop; use def routes or an async driver.

  • An unbounded list endpoint, or OFFSET on huge tables. Bound limit, order by a unique column, consider keyset.

  • SQLite foreign keys are off by default. Add the PRAGMA foreign_keys=ON listener, or tests will accept data PostgreSQL rejects.

  • create_all as a migration tool. It never alters tables; use Alembic and review autogenerated scripts.

  • % in the Alembic URL. Double it before set_main_option.

  • Passwords in code, YAML or the image. Use secret files or the platform’s secret store, and SecretStr in settings.

  • Exhausting the pool or the server’s max_connections. Size pool_size, max_overflow and worker count together.

  • Mixing SQLModel with an unpinned SQLAlchemy. Let SQLModel’s requirement choose the SQLAlchemy version.

References