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

> A guide to handling relational DBs at production quality with SQLModel—the choice in FastAPI's official SQL tutorial. From table=True model definitions of SQLModel (which integrates Pydantic and SQLAlchemy under one author), yield session dependencies, CRUD, a design that guards input/public/update boundaries with multiple models, relations, to SQLite→PostgreSQL async, Alembic, connection pools, N+1 avoidance, transactions, and testing—all in real code.

- Published: 2026-06-26
- Author: 友田 陽大
- Tags: Python, FastAPI, SQLModel, データベース, SQLAlchemy, 型安全
- URL: https://tomodahinata.com/en/blog/fastapi-sqlmodel-database-crud-relationships-production-guide
- Category: Python backend
- Pillar guide: https://tomodahinata.com/en/blog/fastapi-production-async-pydantic-observability-guide

## Key points

- SQLModel is a library that integrates Pydantic (validation) and SQLAlchemy (ORM) under the same author, tiangolo. One model can serve as both a 'table definition' and a 'validation schema,' FastAPI-native and with little duplication
- The table model is class Hero(SQLModel, table=True). Field(default=None, primary_key=True) / Field(index=True). Confine the session in a yield dependency in get_session, and reuse it across all endpoints with SessionDep = Annotated[Session, Depends(get_session)]
- The key to production is 'guarding boundaries with multiple models.' Split HeroBase/Hero(table=True)/HeroPublic/HeroCreate/HeroUpdate, don't let input include id, and don't leak secret etc. to the public. Updates use hero.sqlmodel_update(data.model_dump(exclude_unset=True))
- Relations are Relationship(back_populates=...) and Field(foreign_key=...). Specify HeroPublicWithTeam etc. in response_model to return relations included. Break cycles by not using '...WithTeam' nested
- Productionizing: SQLite→PostgreSQL + async (AsyncEngine/AsyncSession, asyncpg), create_all is discouraged in production so migrate with Alembic, connection pools, N+1 avoidance with selectinload, transaction boundaries, idempotent upsert, test isolation with dependency_overrides

---

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](/case-studies/ai-video-localization-lipsync) 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.**

```python
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](/blog/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.**

```python
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

```python
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)

```python
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](/blog/fastapi-production-async-pydantic-observability-guide)).

```python
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."**

```python
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

```python
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

```python
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

```python
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.

1. **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.
2. **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).

```python
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.

```python
@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](/blog/pydantic-v2-production-validation-type-safety)).

---

## 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"

```python
# 更新は「一部だけ送る」のが普通。だから全フィールドを任意（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

```python
@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

```python
@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.

```python
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.

```python
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] = []
```

```python
@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.

```python
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`.**

```python
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.

```bash
# 初期化（一度だけ）。以後はモデルの変更を検出してマイグレーションを自動生成できる。
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](/blog/alembic-zero-downtime-migrations-sqlalchemy).

### 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.

```python
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."

```python
# 複数の書き込みを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.

```python
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](/case-studies/ai-video-localization-lipsync), 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

```python
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.

```python
@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](/case-studies/ai-video-localization-lipsync), 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)](https://fastapi.tiangolo.com/tutorial/sql-databases/) — `table=True`, the `get_session` dependency, CRUD, multiple models, with SQLModel
- [SQLModel (official top)](https://sqlmodel.tiangolo.com/) — the philosophy of Pydantic × SQLAlchemy integration and the basic `Hero` model
- [SQLModel with FastAPI (tutorial)](https://sqlmodel.tiangolo.com/tutorial/fastapi/) — engine, session, table creation at startup
- [Multiple Models with FastAPI (SQLModel)](https://sqlmodel.tiangolo.com/tutorial/fastapi/multiple-models/) — the boundary design of `HeroBase`/`HeroCreate`/`HeroPublic`/`HeroUpdate`
- [Relationships in FastAPI (SQLModel)](https://sqlmodel.tiangolo.com/tutorial/fastapi/relationships/) — `Relationship(back_populates=)`, `foreign_key`, returning relations with `response_model`
- [Update Data with FastAPI (SQLModel)](https://sqlmodel.tiangolo.com/tutorial/fastapi/update/) — `HeroUpdate`, `sqlmodel_update`, `model_dump(exclude_unset=True)`
- [SQLAlchemy 2.0 documentation](https://docs.sqlalchemy.org/) — engine, connection pool, `selectinload`/`joinedload`, async
- [Alembic documentation](https://alembic.sqlalchemy.org/) — migrations (the production alternative to `create_all`)