Chapter 13 · Database Integration (SQLModel and Async SQLAlchemy)

Chapter goals: Use the officially recommended SQLModel to create tables, inject Session dependencies, and perform full CRUD; understand the "table model vs data model" separation strategy, and master the async engine variant.

13.1 SQLModel: Pydantic Meets SQLAlchemy

FastAPI does not bind to any database, but author tiangolo specifically wrote SQLModel — built on SQLAlchemy and Pydantic, one class serves as both a Pydantic model (for request/response validation) and an ORM table model. SQLModel supports all databases that SQLAlchemy supports (PostgreSQL, MySQL, SQLite, etc.). This chapter uses SQLite for a zero-configuration demo.

uv add sqlmodel   # or pip install sqlmodel

13.2 Single-Model Version: From Table Creation to CRUD

The simplest runnable full-code example:

# main.py
from sqlmodel import Field, Session, SQLModel, create_engine, select
from fastapi import Depends, FastAPI, HTTPException
from contextlib import asynccontextmanager

class Hero(SQLModel, table=True):
    # table=True declares this as a "table model"; primary key as int | None,
    # meaning the Python object can be created without an id — the database generates it
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)      # index=True creates a DB index, speeding up name queries
    age: int | None = None
    secret_name: str

engine = create_engine(
    "sqlite:///database.db",
    connect_args={"check_same_thread": False},  # SQLite-specific: allows cross-thread connection use
)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)   # Create tables from all table models (skips if already exists)

@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()   # In production, use a migration tool like Alembic instead
    yield

app = FastAPI(lifespan=lifespan)

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

# Annotated dependency alias: reusable across the project
SessionDep = Annotated[Session, Depends(get_session)]

@app.post("/heroes/")
def create_hero(hero: Hero, session: SessionDep):
    session.add(hero)
    session.commit()
    session.refresh(hero)    # Refresh to pick up the database-generated id
    return hero

@app.get("/heroes/")
def read_heroes(session: SessionDep, offset: int = 0, limit: int = 100):
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes

@app.get("/heroes/{hero_id}")
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

@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"ok": True}

Key points:

• check_same_thread=False is SQLite-specific. FastAPI may use multiple threads in a single request (e.g., sync dependencies); relaxing this limit is safe because "one Session per request" guarantees correctness;
• get_session uses yield: create the session when the request starts, close it automatically when the response completes — this is the standard application of yield dependencies from Chapter 11;
• session.exec() is SQLModel's wrapper (returns typed rows), equivalent to SQLAlchemy's session.execute(select(...)).

How to do updates

SQLModel/SQLAlchemy has no built-in update endpoint pattern: first hero = session.get(Hero, id) to fetch the object, then use Pydantic's hero_data.model_dump(exclude_unset=True) to filter out fields the client did not send, then setattr each field and commit(). exclude_unset=True is the key — it makes "unsent fields do not overwrite existing values" possible.

13.3 Multi-Model Separation: Never Expose Table Models Directly to the API

The single-model version has two security issues: the client can specify id at creation time (potentially overwriting existing data); secret_name is returned裸 to everyone. The correct approach is to split into multiple models, using inheritance to eliminate duplication:

from sqlmodel import SQLModel, Field

class HeroBase(SQLModel):
    """Fields shared by all models"""
    name: str
    secret_name: str
    age: int | None = None


class Hero(HeroBase, table=True):
    """Table model: has id, stored in the database"""
    id: int | None = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    """Input model: client sends this to create a hero"""
    pass


class HeroRead(HeroBase):
    """Output model: never includes secret_name"""
    id: int
    secret_name: str = ""  # Hidden from client

13.4 Async Engine Version

For production (PostgreSQL/MySQL), use the async engine:

from sqlmodel import create_engine
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

async_engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")

async def get_async_session():
    async with AsyncSession(async_engine) as session:
        yield session

Replace session.exec(...) with await session.exec(...) and session.get(...) with await session.get(...) throughout.

13.5 Chapter Summary

• SQLModel unifies Pydantic (validation) + SQLAlchemy (ORM) in one class;
• table=True makes a class a database table model; Field(primary_key=True) declares the PK;
• yield in get_session() ensures the DB session is closed after each request;
• Separate table models (DB) from input/output models (API) for security;
• Async engine + AsyncSession for production databases.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. What does SQLModel combine?

2. Why does get_session use yield instead of return?

3. Why separate Hero (table model) from HeroRead (output model)?

4. What is check_same_thread=False used for in SQLite?

🛠️ Hands-On Exercises

1. Write a full CRUD app with SQLModel + SQLite: create, read all, read by ID, update, delete — each as a separate endpoint.
2. Split the single Hero model into HeroBase / Hero (table) / HeroCreate / HeroRead; verify secret_name never appears in GET responses.
3. Convert the SQLite app to async PostgreSQL; replace session.exec() with await session.exec().

Proceed to Chapter 14: Security Basics — OAuth2 and JWT.