A Production Guide to Putting a DB on FastAPI: CRUD, Multi-Model Boundaries, Relations, and Async Alembic with SQLModel (Pydantic × SQLAlchemy Integration)

When you want to "put a DB on" FastAPI, the first fork is the choice of ORM. Raw SQLAlchemy, Tortoise, or SQLModel? What FastAPI's official tutorial adopts is SQLModel, and there's a clear reason for this—FastAPI, Pydantic, and SQLModel are all made by the same author (Sebastián Ramírez / tiangolo), and their design philosophies line up in a straight line.

This article is a guide to handling relational DBs at production quality with that SQLModel. While following FastAPI's official SQL tutorial and SQLModel's official flow faithfully to the latest spec, it raises the parts the official docs note as "for learning" (SQLite, create_all, a global session) into a form that withstands real operation. As source material, I'll weave in decisions from the AI video-localization / lip-sync foundation I built (designing a resumable, idempotent GPU pipeline with FastAPI + Celery + SQLAlchemy async).

The rules of this article: APIs and recommendations are based on the FastAPI / SQLModel official documentation (as of June 2026). Confirm the latest before production. SQLModel is a library integrating Pydantic and SQLAlchemy, and its spec is revised following both of their updates. DB connection URLs and credentials are all assumed to be in environment variables (never hardcode).

0. First, the decision: when to use SQLModel and when to use raw SQLAlchemy

Before implementing, finish the tool selection. Leaving this ambiguous lands you later in the hybrid swamp of "forcing SQLModel and calling raw SQLAlchemy."

0.1 What SQLModel really is: the integration of Pydantic + SQLAlchemy

As the official SQLModel words put it—SQLModel is a library "powered by Pydantic and SQLAlchemy", made by the same author as FastAPI. This origin produces a decisive advantage.

Normally, handling a DB with FastAPI doubles the models. You write separately and convert between "SQLAlchemy's table model" and "Pydantic's request/response schema"—a textbook DRY violation, creating the chore of fixing two places every time you add a field.

SQLModel folds this duplication by letting one class serve as both a table definition and a validation schema.

from sqlmodel import Field, SQLModel

# このクラス1つが「DBテーブル」かつ「Pydanticモデル」。
# table=True がDBテーブルとしての顔を有効にする（後述：データ専用モデルは table を付けない）。
class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: int | None = None

0.2 When to use which: SQLModel or raw SQLAlchemy 2.0

"Integration is convenient," but it's not omnipotent. Here's the decision table.

Aspect	SQLModel (this article)	Raw SQLAlchemy 2.0
Suited case	FastAPI-native, build fast with a single source of truth, eliminate model duplication	Complex queries (CTEs, window functions, complex JOINs), existing SQLAlchemy assets, fine-grained mapping control
Model	One class serves as table + schema (little duplication)	Write ORM models and Pydantic schemas separately (high freedom)
Learning cost	Low if you know Pydantic/FastAPI	Many ORM-specific concepts, takes time to master
Escape hatch	If needed, you can descend straight to the internal SQLAlchemy (the same `select` works)	Lower-level than SQLModel but you get full control

The rule of thumb: if you're FastAPI-centric and CRUD and straightforward relations are the main thing, SQLModel. If complex queries or heavy existing assets, raw SQLAlchemy. The important thing is that SQLModel is a thin wrapper over SQLAlchemy, so when needed you can descend to the low level with the same engine and the same select statement—it's not a one-way enclosure. For typed design with raw SQLAlchemy, read alongside the SQLAlchemy 2 typed-ORM production guide. This article covers the path of "using SQLModel to the fullest at production quality."

1. Define the model: `table=True` and `Field`

The foundation is the model definition. SQLModel translates Python type hints directly into table definitions.

from sqlmodel import Field, SQLModel

class Hero(SQLModel, table=True):
    # id は DB が自動採番する。だから「未確定（None）でも作れる」が、DB上は主キー。
    id: int | None = Field(default=None, primary_key=True)
    # index=True で検索高速化のインデックスを張る（name で頻繁に絞るなら付ける）。
    name: str = Field(index=True)
    age: int | None = Field(default=None, index=True)
    secret_name: str

There are three points to grasp.

table=True is the declaration "this class is a DB table." A class without this becomes just a data model (Pydantic-equivalent) that doesn't create a table. This difference is the core of the Chapter 4 design "guard boundaries with multiple models."
Field(default=None, primary_key=True) defines the primary key. Making id int | None is because id is undetermined (None) until the DB numbers it. The type honestly expresses the reality that "it sometimes doesn't exist yet."
Field(index=True) places an index. Attaching it to columns you frequently filter on like WHERE name = ... speeds up searches. Conversely, the index-update cost rides on every write, so not attaching it to everything is the production know-how (cost efficiency).

Field serves both Pydantic and SQLAlchemy: SQLModel's Field makes Pydantic's validation (max_length, etc.) and SQLAlchemy's column attributes (primary_key / index / foreign_key) compatible in one declaration. This is the true nature of "single source of truth."

2. The engine and the session dependency (`yield`)

Once the model is done, prepare the connection to the DB (the engine) and the per-request unit of work (the session).

2.1 Create the engine

from sqlmodel import SQLModel, create_engine

sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

# SQLite固有の注意：FastAPIは1リクエストを複数スレッドで触り得るので、
# check_same_thread=False で「別スレッドからの利用」を許可する。
# ※これはSQLite限定の事情。PostgreSQLでは不要（第7章で差し替える）。
connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)

Create only one engine for the whole app. It holds an internal connection pool, so don't recreate it per request (an accident in both cost and performance).

2.2 Create the tables (※ don't use `create_all` in production)

def create_db_and_tables() -> None:
    # 定義済みの全テーブルを「無ければ作る」。学習・プロトタイプには最適。
    SQLModel.metadata.create_all(engine)

[Most important] Don't bring create_all into production: SQLModel.metadata.create_all only "creates tables if they don't exist" and does not change existing tables (add columns, change types). In production where the schema grows, migrations with Alembic are essential (covered in Chapter 7.3). Treat create_all as "first-time / test / local"-only.

The wiring to prepare tables at startup is done in lifespan (recommended). The official tutorial uses @app.on_event("startup"), but that's a deprecated old API, and the current correct way is the lifespan context manager (production-grade lifespan design is in the production-operations guide).

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()   # 起動時に一度（本番ではここはAlembicに置き換える）
    yield                    # ← アプリ稼働中
    # 終了時の後始末が要ればここに書く

app = FastAPI(lifespan=lifespan)

2.3 Confine the session in a `yield` dependency

This is the most important FastAPI × SQLModel pattern. Carve the session (the unit of work with the DB) out into a dependency, and guarantee with yield that "it's reliably closed when done."

from typing import Annotated
from fastapi import Depends
from sqlmodel import Session

def get_session():
    # with でセッションを開き、yield で渡し、ハンドラ終了後に自動でクローズ。
    # 1リクエスト = 1セッションを保証する（接続リークを構造的に防ぐ）。
    with Session(engine) as session:
        yield session

# 公式推奨：型エイリアスに畳んで全エンドポイントで再利用する（DRY）。
SessionDep = Annotated[Session, Depends(get_session)]

The combination of with Session(engine) as session: and yield works because the session is always closed whether the handler succeeds or throws. Connection leaks—the accident that bites most quietly and most fatally in production—are prevented by structure. Each handler can receive a verified session by just writing session: SessionDep.

3. CRUD: create, read, list

Once you have the session, CRUD is straightforward. SQLModel's method names derive from SQLAlchemy and are intuitive once learned.

3.1 Create

from fastapi import FastAPI

@app.post("/heroes/")
def create_hero(hero: Hero, session: SessionDep) -> Hero:
    session.add(hero)       # セッションに追加（まだDBには書かれていない）
    session.commit()        # ここでINSERTが確定する（トランザクションをコミット）
    session.refresh(hero)   # DBが採番した id 等を、オブジェクトに反映して読み戻す
    return hero

Understand the meaning of the three steps and you won't get stuck.

add: places the object in the session's "pending." At this point the DB is unchanged.
commit: commits the transaction and executes the INSERT.
refresh: reads back into the object the id the DB numbered and DB-side default values. Omitting this is the typical bug where the returned hero.id stays None.

3.2 Read one

from fastapi import HTTPException

@app.get("/heroes/{hero_id}")
def read_hero(hero_id: int, session: SessionDep) -> Hero:
    # 主キーでの取得は session.get が最短（select を書く必要が無い）。
    hero = session.get(Hero, hero_id)
    if not hero:
        # 見つからないなら404。境界で必ず存在チェックする。
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

For taking one by primary key, session.get(Hero, hero_id) is shortest. You don't need to write select.

3.3 Read many — always paginate

from fastapi import Query
from sqlmodel import select

@app.get("/heroes/")
def read_heroes(
    session: SessionDep,
    offset: int = 0,
    # 上限を Query(le=100) で固定する。無制限の limit は「全件取得DoS」になり得る。
    limit: Annotated[int, Query(le=100)] = 100,
) -> list[Hero]:
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes

Don't make a list with an unlimited limit (cost, availability): exposing select(Hero).all() unlimited means one request eats the DB and memory the day the table grows to a million rows. Make pagination with offset/limit mandatory from the start, and fix the upper bound on the server with Query(le=100). A state where "the client can request a mass fetch" is a security problem (a resource-exhaustion attack) as well as a performance problem.

session.exec(select(...)) is the basic form of SQLModel querying. select is SQLAlchemy's select itself, so when it gets complex you can descend straight to the low level (Chapter 0's "escape hatch").

4. Guard "boundaries" with multiple models (the dividing line of production quality)

So far I've used a single Hero. But in production you must not reuse one model. There are two reasons, both directly tied to security.

Don't let input include id—id is for the DB to number. An API where the client can send id is an entry point for an attack that overwrites someone else's record.
Don't leak secrets in the public response—you must not include secret_name (here a representative example of a secret field) or hashed_password directly in the response.

SQLModel solves this with model splitting. Make one table model, and data models per use, consolidating the common part in HeroBase (DRY).

from sqlmodel import Field, SQLModel

# 共通の土台。table を付けない＝ただのデータモデル（Pydantic相当）。
class HeroBase(SQLModel):
    name: str = Field(index=True)
    age: int | None = Field(default=None, index=True)

# テーブルモデルはこれ「だけ」。table=True を持つのは1つに保つ。
class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    secret_name: str        # 機密。公開モデルには載せない。

# 入力スキーマ：id を持たない（クライアントに id を送らせない＝注入を防ぐ）。
class HeroCreate(HeroBase):
    secret_name: str        # 作成時は必須。だが公開レスポンスには出さない。

# 公開スキーマ：id は必須（DBから来た以上 id は必ずある）。secret_name は無い。
class HeroPublic(HeroBase):
    id: int

Let me confirm the design intent in the official words.

HeroCreate having no id is because, per the official docs, "the DB auto-numbers id, so we don't want the client to send it." This structurally prevents ID injection.
HeroPublic's id being required (int, not None) is because, per the official docs, "a response is always a model that came from the DB, so it always has id." The type guarantees the fact that "when public, it's always numbered."
secret_name is in Hero (table) and HeroCreate (input), but not in HeroPublic (public). This is leak prevention by type. Pass the response through HeroPublic, and secret_name is structurally not output.

Apply this to CRUD. Fix "the public shape" with response_model and "the input shape" with the argument.

@app.post("/heroes/", response_model=HeroPublic)   # 出力は HeroPublic に絞る（機密を漏らさない）
def create_hero(hero: HeroCreate, session: SessionDep):  # 入力は HeroCreate（id を受けない）
    # 入力スキーマ → テーブルモデルへ変換。バリデーション済みの値だけが通る。
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero        # Hero を返しても response_model=HeroPublic で secret_name は除去される

@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()

A two-layer schema boundary is security design itself: convert "input → table" with Hero.model_validate(hero), and "table → public" with response_model=HeroPublic—this double checkpoint of entrance and exit plugs ID injection and secret leakage at once. It's the exact same principle as "don't include hashed_password in the public model" that I touched on in the auth guide, and in SQLModel this is the framework's standard practice (the thinking on boundary validation is in the Pydantic guide).

5. Update and delete: reflect "only the sent values" with `exclude_unset`

Update (PATCH) has a partial-update trap. "I wanted to update only age, but the name I didn't send got overwritten with None"—this is the most common accident in partial updates. SQLModel handles it correctly with exclude_unset.

5.1 The update model is "all fields optional"

# 更新は「一部だけ送る」のが普通。だから全フィールドを任意（Noneデフォルト）にする。
# 公式注記：全フィールドが任意なので HeroBase からは継承できない（HeroBaseは name 必須）。
class HeroUpdate(SQLModel):
    name: str | None = None
    age: int | None = None
    secret_name: str | None = None

5.2 The PATCH path operation

@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):
    hero_db = session.get(Hero, hero_id)
    if not hero_db:
        raise HTTPException(status_code=404, detail="Hero not found")
    # exclude_unset=True が肝：「クライアントが実際に送ったフィールドだけ」を dict 化する。
    # 送らなかったフィールドは dict に入らない＝既存値を維持する（誤った上書きを防ぐ）。
    hero_data = hero.model_dump(exclude_unset=True)
    # sqlmodel_update：dict の値だけをオブジェクトに適用する（SQLModel 専用の安全な更新）。
    hero_db.sqlmodel_update(hero_data)
    session.add(hero_db)
    session.commit()
    session.refresh(hero_db)
    return hero_db

exclude_unset=True is the correct answer for partial updates: model_dump(exclude_unset=True) makes a dict of only the fields the client explicitly sent. Fields not sent don't appear in the dict, so sqlmodel_update keeps the existing values. This makes "send only age and only age changes" hold correctly. Conversely, forgetting exclude_unset crushes all un-sent fields with the default value (None)—a typical production accident, so remember it as mandatory for PATCH.

5.3 Delete

@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}

In production, doubt "physical deletion": from the perspective of auditing, recovery from misoperation, and referential integrity, there are many scenes where logical deletion (a soft delete setting a deleted_at column) is appropriate. Before physical deletion with session.delete, check "is this really data OK to delete" against the requirements. In the video-localization foundation too, I designed job history to be kept via state transitions, not deleted (covered later).

6. Relations: `Relationship` and `foreign_key`

Connections between tables are expressed with the two-piece set of a foreign key (foreign_key) and a relation (Relationship). Let me use Team and Hero (the heroes belonging to it) as an example.

from sqlmodel import Field, Relationship, SQLModel

class TeamBase(SQLModel):
    name: str = Field(index=True)

class Team(TeamBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    # 1チームに複数ヒーロー。back_populates で Hero.team と双方向に結ぶ。
    heroes: list["Hero"] = Relationship(back_populates="team")

class HeroBase(SQLModel):
    name: str = Field(index=True)
    age: int | None = Field(default=None, index=True)
    # 外部キー：DBレベルで team.id を参照する。これが整合性の土台。
    team_id: int | None = Field(default=None, foreign_key="team.id")

class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    secret_name: str
    # 多側から1側へ。back_populates="heroes" で Team.heroes と対応させる。
    team: Team | None = Relationship(back_populates="heroes")

Understand them by separating roles.

Field(foreign_key="team.id") is the DB's foreign-key constraint. The DB guarantees that team_id always points to an actually-existing team.id (referential integrity).
Relationship(back_populates=...) is a convenient reference on the Python object. You can access without writing SQL—the belonging team via hero.team, the list of belonging heroes via team.heroes. Linking the two with back_populates keeps them consistent when you change one (bidirectional).

6.1 Return relations included: nesting `response_model`

To return related data included, nest the public models. There's a circular-reference trap here, so follow the official practice.

class TeamPublic(TeamBase):
    id: int

class HeroPublic(HeroBase):
    id: int

# ヒーロー＋所属チームを返す。team は TeamPublic（…WithHeroes ではない）にして循環を断つ。
class HeroPublicWithTeam(HeroPublic):
    team: TeamPublic | None = None

# チーム＋所属ヒーロー一覧。heroes は HeroPublic（…WithTeam ではない）にして循環を断つ。
class TeamPublicWithHeroes(TeamPublic):
    heroes: list[HeroPublic] = []

@app.get("/heroes/{hero_id}", response_model=HeroPublicWithTeam)
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    # team も含めて HeroPublicWithTeam の形で返る

Break circular references by structure: making HeroPublicWithTeam's team a TeamPublicWithHeroes would expand infinitely—"hero → team → hero → ...". So the side used for nesting must always be a one-level-shallower public model (TeamPublic / HeroPublic)—this is the official practice. Fixing "how far to return relations included" to one level by type also prevents response bloat.

7. Productionize: graduate from SQLite, and face async, Alembic, the pool, and N+1

So far that's a "working DB app." From here is the diff to make it not fall over, not leak, be fast. Fill in what the official tutorial stops at with SQLite.

7.1 SQLite → PostgreSQL

SQLite is ideal for learning and embedding, but not suited to concurrent writes, network access, or production load. In production, PostgreSQL is the standard. The swap centers on the connection URL and pool settings.

import os
from sqlmodel import create_engine

# 接続URLは必ず環境変数から。コードにもgitにも資格情報を置かない。
DATABASE_URL = os.environ["DATABASE_URL"]   # 例: postgresql+psycopg://user:pass@host:5432/db

engine = create_engine(
    DATABASE_URL,
    pool_size=10,           # 常時保持する接続数（CPU/同時実行に合わせる）
    max_overflow=20,        # 一時的に超過できる上限
    pool_pre_ping=True,     # 死んだ接続を使う前に軽く疎通確認（切断後の事故を防ぐ）
    pool_recycle=1800,      # 一定時間で接続を作り直す（DB側のidleタイムアウト対策）
)
# 注意：PostgreSQL では SQLite 用の connect_args={"check_same_thread": False} は不要。

The connection pool affects both "cost" and "availability": establishing a DB connection is heavy. Reusing connections with a pool lowers latency and DB load. pool_pre_ping=True is a near-mandatory production setting that prevents the accident of grabbing a dead connection severed by a load balancer's or DB's idle timeout. The pool's optimal value changes per environment (serverless, via RDS Proxy, etc.), so always match the connection-count upper bound against the DB's max_connections.

7.2 Async: AsyncEngine / AsyncSession (asyncpg)

FastAPI is an async framework. So DB access doesn't stop other requests on I/O waits, in production the standard is to use an async driver (asyncpg for PostgreSQL) and AsyncSession.

from collections.abc import AsyncGenerator
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlmodel import select

# 非同期エンジン。ドライバは asyncpg（URLの +asyncpg に注目）。
async_engine = create_async_engine(
    os.environ["DATABASE_URL"],   # 例: postgresql+asyncpg://user:pass@host/db
    pool_pre_ping=True,
)
AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False)

async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    # 非同期でも yield 依存の形は同じ。async with で確実にクローズする。
    async with AsyncSessionLocal() as session:
        yield session

AsyncSessionDep = Annotated[AsyncSession, Depends(get_async_session)]

@app.get("/heroes/", response_model=list[HeroPublic])
async def read_heroes(session: AsyncSessionDep):   # 非同期ハンドラ
    # 同期版の session.exec(...) は使えない。await session.execute(...) を使う。
    result = await session.execute(select(Hero).limit(100))
    return result.scalars().all()

If you go async, stay async to the end (mixing is an accident): making a synchronous blocking DB call inside an async handler stops the event loop and clogs all requests. In an async def handler use await session.execute(...), and in a def (sync) handler use a sync session—don't mix sync and async is the iron rule. In the video-localization foundation, I unified the API layer with SQLAlchemy async and completely separated heavy GPU processing to the synchronous Celery worker side (the boundary: I/O is async, CPU/GPU load goes to workers).

7.3 Migrations: ditch `create_all` for Alembic

As foreshadowed in Chapter 2.2, don't use create_all in production. The schema grows—add columns, change types, re-place indexes—and what applies that change history safely and reversibly is Alembic.

# 初期化（一度だけ）。以後はモデルの変更を検出してマイグレーションを自動生成できる。
alembic init alembic
alembic revision --autogenerate -m "create hero and team tables"
alembic upgrade head     # 本番への適用はこのコマンド（CI/CDから流す）

Since SQLModel's models internally hold SQLAlchemy's metadata, autogenerate works just by pointing Alembic's target_metadata = SQLModel.metadata. In production, zero-downtime migration procedures (adding columns backward-compatibly, removing in two stages, etc.) matter, and these are detailed in the Alembic zero-downtime migration guide.

7.4 Crush the N+1 problem: `selectinload` / `joinedload`

The biggest pitfall of relations is the N+1 problem. "Take 100 heroes, and when I accessed each hero's team, the team-fetch SQL flew an extra 100 times"—this is N+1. It hits the DB with 1+N queries.

from sqlalchemy.orm import selectinload

@app.get("/heroes/", response_model=list[HeroPublicWithTeam])
async def read_heroes_with_team(session: AsyncSessionDep):
    # selectinload で「関連 team を別の1クエリでまとめて取る」。N+1 が 2クエリに畳まれる。
    result = await session.execute(
        select(Hero).options(selectinload(Hero.team)).limit(100)
    )
    return result.scalars().all()

The rule of thumb: for including a one-to-many, selectinload (one extra query, batch-fetch with IN), and for including a many-to-one, joinedload (one query with a JOIN), basically. On endpoints that return relations via response_model, always attach eager load—make "if you put a relation in response_model, add eager load" a habit as a set.

7.5 Transaction boundaries and idempotent upsert

What makes multiple writes collectively succeed or fail is a transaction. The unit of commit becomes the boundary of "all reflected or all rolled back."

# 複数の書き込みを1トランザクションに束ねる：途中で失敗したら全体がロールバックされる。
def transfer_hero(session: Session, hero_id: int, new_team_id: int) -> None:
    hero = session.get(Hero, hero_id)
    if hero is None:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero.team_id = new_team_id
    # ここで他の書き込みを足しても、commit までは確定しない。
    session.add(hero)
    session.commit()         # ここで初めて全変更が一括確定する（途中例外なら何も残らない）

Idempotency is also a production requirement. So that "the same request arriving twice yields one operation's worth," use an upsert (insert if absent, update if present) with a natural key.

from sqlalchemy.dialects.postgresql import insert as pg_insert

def upsert_hero(session: Session, name: str, secret_name: str) -> None:
    # name を一意キーとして、衝突したら更新する（ON CONFLICT DO UPDATE）。
    # リトライや二重送信で重複行を作らない＝冪等。
    stmt = pg_insert(Hero).values(name=name, secret_name=secret_name)
    stmt = stmt.on_conflict_do_update(
        index_elements=["name"],                       # 一意制約のある列
        set_={"secret_name": secret_name},
    )
    session.execute(stmt)
    session.commit()

Idempotency is the heart of a "resumable pipeline": in the AI video-localization foundation, I made the DB the single source of truth for job state, turned all stages into async tasks, and managed each stage's state and the failed stage in the DB. Long videos are split into segments and resumable from the last completed segment. Further, deriving the cache key from "input + engine/tuning," a re-run under the same conditions idempotently reuses the existing result. These were all field applications of this chapter's principles—"make the DB state the source of trust, and keep writes idempotent."

8. Testing: a test engine and `dependency_overrides`

There's no production launch without a verification path. FastAPI × SQLModel can swap the session dependency wholesale, so DB-backed APIs can also be tested deterministically.

8.1 In-memory SQLite + dependency override

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

from app.main import app, get_session

@pytest.fixture(name="session")
def session_fixture():
    # テスト専用：インメモリSQLite。StaticPool で「同じ接続」を共有し、テーブルを保持する。
    engine = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    SQLModel.metadata.create_all(engine)     # テストではこれでよい（本番はAlembic）
    with Session(engine) as session:
        yield session

@pytest.fixture(name="client")
def client_fixture(session: Session):
    # 本番の get_session を、テスト用セッションを返す関数で上書きする。
    def get_session_override():
        return session
    app.dependency_overrides[get_session] = get_session_override
    yield TestClient(app)
    app.dependency_overrides.clear()         # テスト間の汚染を防ぐため必ずクリア

def test_create_hero(client):
    res = client.post("/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"})
    assert res.status_code == 200
    data = res.json()
    assert data["name"] == "Deadpond"
    assert "id" in data
    assert "secret_name" not in data          # 公開モデルが機密を漏らさないことを検証

There are two points. (1) Stand up an independent in-memory engine for tests, so you don't dirty the production DB. Without sharing connections via StaticPool, the in-memory DB is recreated per query and the table disappears. (2) dependency_overrides[get_session] swaps only the session without changing the handler's contents at all.

8.2 Make tests fast and isolated with transaction rollback

Instead of recreating tables per test, there's also the strategy of wrapping each test in a transaction and rolling back when done. The DB state automatically rewinds between tests, so isolation is fast and reliable.

@pytest.fixture(name="session")
def session_fixture():
    # 外側のトランザクションを開始し、テスト終了時に rollback で全変更を捨てる。
    connection = engine.connect()
    transaction = connection.begin()
    with Session(bind=connection) as session:
        yield session
    transaction.rollback()      # コミットされた分も含めて巻き戻す（テスト間が完全独立）
    connection.close()

dependency_overrides is 'the biggest return of testing': you can inject the session from outside without putting a single line of "test branching" in production code. This directly recovers the Chapter 2.3 act of "confining the session in a dependency." Without cleaning up the override every time with finally/clear(), tests get contaminated and produce false positives/negatives, so cleanup is mandatory.

9. Summary: a production FastAPI × SQLModel cheat sheet

A quick reference for when you're unsure.

What it is: SQLModel = Pydantic (validation) + SQLAlchemy (ORM) integrated under the same author. One class serves as table + schema, FastAPI-native, with little duplication. For complex queries or heavy existing assets, raw SQLAlchemy (you can descend from SQLModel if needed).
Model: class Hero(SQLModel, table=True). Field(default=None, primary_key=True) / Field(index=True). id allows pre-numbering None as int | None.
Engine/session: one engine per app (pool built in). SQLite needs connect_args={"check_same_thread": False}. Confine the session in get_session's yield dependency, and reuse with SessionDep = Annotated[Session, Depends(get_session)].
CRUD: create is add→commit→refresh (read back the numbered id). Single is session.get(Hero, id). List is session.exec(select(Hero).offset().limit()).all(), with limit fixed by Query(le=100).
Boundaries: split HeroBase/Hero(table=True)/HeroPublic/HeroCreate/HeroUpdate. Don't let input include id, don't put secrets in the public. The double checkpoint of Hero.model_validate(input) and response_model=public.
Update/delete: PATCH is hero.sqlmodel_update(data.model_dump(exclude_unset=True)) (reflect only sent values). For delete, also consider logical deletion in production.
Relations: Field(foreign_key="team.id") + Relationship(back_populates=...). When returning, nest like HeroPublicWithTeam, but the inside of the nest is a one-level-shallower public model to break cycles.
Productionize: SQLite→PostgreSQL, async (AsyncEngine/AsyncSession, asyncpg) stays async to the end. Ditch create_all for Alembic. Connection pool (pool_pre_ping). N+1 with selectinload/joinedload. Transaction boundaries and idempotent upsert.
Testing: in-memory SQLite + StaticPool, isolated with dependency_overrides[get_session]. Faster still with a rollback strategy.

SQLModel is a library so fast to get going that "if you know Pydantic, a DB rides on the same day." But production quality, here too, is decided by boundary design. Split models for input and public, confine the session in a dependency, replace create_all with Alembic, eager-load relations, and keep writes idempotent—none of it is flashy, but this accumulation creates a "no-leak, no-fall, easy-to-change DB layer."

In the AI video-localization / lip-sync foundation, I designed a resumable, idempotent GPU pipeline with FastAPI + Celery + SQLAlchemy async at the core, making the DB the single source of truth for job state. I managed every stage's state and the failed stage in the DB, resumed long videos by segment, and idempotently folded re-runs under the same conditions with a cache—a project that ran this article's principles, "consolidate state in the DB and keep writes idempotent," in the field. With generative AI (Claude Code) as my partner, my approach is to build fast and cheaply, solo while guaranteeing quality with tests and verification gates.

"I want to put a DB on FastAPI, but should I use SQLModel or raw SQLAlchemy, and how should I design model boundaries, migrations, async, and idempotency?"—I'll accompany you end-to-end, from that decision through implementation, migration, and operation. Feel free to reach out, even from the requirements-organizing stage.

References (official documentation)

SQL (Relational) Databases (FastAPI) — table=True, the get_session dependency, CRUD, multiple models, with SQLModel
SQLModel (official top) — the philosophy of Pydantic × SQLAlchemy integration and the basic Hero model
SQLModel with FastAPI (tutorial) — engine, session, table creation at startup
Multiple Models with FastAPI (SQLModel) — the boundary design of HeroBase/HeroCreate/HeroPublic/HeroUpdate
Relationships in FastAPI (SQLModel) — Relationship(back_populates=), foreign_key, returning relations with response_model
Update Data with FastAPI (SQLModel) — HeroUpdate, sqlmodel_update, model_dump(exclude_unset=True)
SQLAlchemy 2.0 documentation — engine, connection pool, selectinload/joinedload, async
Alembic documentation — migrations (the production alternative to create_all)

A Production Guide to Putting a DB on FastAPI: CRUD, Multi-Model Boundaries, Relations, and Async Alembic with SQLModel (Pydantic × SQLAlchemy Integration)

0. First, the decision: when to use SQLModel and when to use raw SQLAlchemy

0.1 What SQLModel really is: the integration of Pydantic + SQLAlchemy

0.2 When to use which: SQLModel or raw SQLAlchemy 2.0

1. Define the model: `table=True` and `Field`

2. The engine and the session dependency (`yield`)

2.1 Create the engine

2.2 Create the tables (※ don't use `create_all` in production)

2.3 Confine the session in a `yield` dependency

3. CRUD: create, read, list

3.1 Create

3.2 Read one

3.3 Read many — always paginate

4. Guard "boundaries" with multiple models (the dividing line of production quality)

5. Update and delete: reflect "only the sent values" with `exclude_unset`

5.1 The update model is "all fields optional"

5.2 The PATCH path operation

5.3 Delete

6. Relations: `Relationship` and `foreign_key`

6.1 Return relations included: nesting `response_model`

7. Productionize: graduate from SQLite, and face async, Alembic, the pool, and N+1

7.1 SQLite → PostgreSQL

7.2 Async: AsyncEngine / AsyncSession (asyncpg)

7.3 Migrations: ditch `create_all` for Alembic

7.4 Crush the N+1 problem: `selectinload` / `joinedload`

7.5 Transaction boundaries and idempotent upsert

8. Testing: a test engine and `dependency_overrides`

8.1 In-memory SQLite + dependency override

8.2 Make tests fast and isolated with transaction rollback

9. Summary: a production FastAPI × SQLModel cheat sheet

References (official documentation)

FastAPI Production-Operations Guide: Building APIs That Don't Fall Over with the Right Use of async, Pydantic v2 Boundary Validation, DI, and Observability

Python Data Types Complete Guide: The 'Right Use' of Numbers, Strings, and Collections, and Designs That Don't Break in Production

The Complete Guide to Python Mappings: dict Internals, Choosing Among collections, Designing Custom Mappings, and Production Operation

FastAPI Authentication & Authorization Production Guide: Protecting an API with the OAuth2 Password Flow × JWT (PyJWT) × Security Scopes

Also worth reading

Flask's Data Layer: Designing and Operating a Production DB with Flask-SQLAlchemy 3.x (2.0 Style) and Flask-Migrate

marshmallow Practical Guide: Robustly Designing Python Object Serialization / Validation at the Boundary (v4-Compatible)

Practical Pydantic error-handling guide: turn ValidationError into usable, safe API errors

0. First, the decision: when to use SQLModel and when to use raw SQLAlchemy

0.1 What SQLModel really is: the integration of Pydantic + SQLAlchemy

0.2 When to use which: SQLModel or raw SQLAlchemy 2.0

1. Define the model: table=True and Field

2. The engine and the session dependency (yield)

2.1 Create the engine

2.2 Create the tables (※ don't use create_all in production)

2.3 Confine the session in a yield dependency

3. CRUD: create, read, list

3.1 Create

3.2 Read one

3.3 Read many — always paginate

4. Guard "boundaries" with multiple models (the dividing line of production quality)

5. Update and delete: reflect "only the sent values" with exclude_unset

5.1 The update model is "all fields optional"

5.2 The PATCH path operation

5.3 Delete

6. Relations: Relationship and foreign_key

6.1 Return relations included: nesting response_model

7. Productionize: graduate from SQLite, and face async, Alembic, the pool, and N+1

7.1 SQLite → PostgreSQL

7.2 Async: AsyncEngine / AsyncSession (asyncpg)

7.3 Migrations: ditch create_all for Alembic

7.4 Crush the N+1 problem: selectinload / joinedload

7.5 Transaction boundaries and idempotent upsert

8. Testing: a test engine and dependency_overrides

8.1 In-memory SQLite + dependency override

8.2 Make tests fast and isolated with transaction rollback

9. Summary: a production FastAPI × SQLModel cheat sheet

References (official documentation)

Related articles

FastAPI Production-Operations Guide: Building APIs That Don't Fall Over with the Right Use of async, Pydantic v2 Boundary Validation, DI, and Observability

Python Data Types Complete Guide: The 'Right Use' of Numbers, Strings, and Collections, and Designs That Don't Break in Production

The Complete Guide to Python Mappings: dict Internals, Choosing Among collections, Designing Custom Mappings, and Production Operation

FastAPI Authentication & Authorization Production Guide: Protecting an API with the OAuth2 Password Flow × JWT (PyJWT) × Security Scopes

Also worth reading

Flask's Data Layer: Designing and Operating a Production DB with Flask-SQLAlchemy 3.x (2.0 Style) and Flask-Migrate

marshmallow Practical Guide: Robustly Designing Python Object Serialization / Validation at the Boundary (v4-Compatible)

Practical Pydantic error-handling guide: turn ValidationError into usable, safe API errors

1. Define the model: `table=True` and `Field`

2. The engine and the session dependency (`yield`)

2.2 Create the tables (※ don't use `create_all` in production)

2.3 Confine the session in a `yield` dependency

5. Update and delete: reflect "only the sent values" with `exclude_unset`

6. Relations: `Relationship` and `foreign_key`

6.1 Return relations included: nesting `response_model`

7.3 Migrations: ditch `create_all` for Alembic

7.4 Crush the N+1 problem: `selectinload` / `joinedload`

8. Testing: a test engine and `dependency_overrides`