# FastAPIにDBを載せる本番ガイド：SQLModel（Pydantic×SQLAlchemy統合）でCRUD・複数モデル境界・リレーション・非同期Alembic

> FastAPI公式のSQLチュートリアルが採用するSQLModelで、リレーショナルDBを本番品質で扱うガイド。PydanticとSQLAlchemyを同一作者が統合したSQLModelのtable=Trueモデル定義、yieldセッション依存、CRUD、複数モデルで入力/公開/更新の境界を守る設計、リレーション、SQLite→PostgreSQL非同期・Alembic・接続プール・N+1回避・トランザクション・テストまで実コードで解説します。

- 公開日: 2026-06-26
- 著者: 友田 陽大
- タグ: Python, FastAPI, SQLModel, データベース, SQLAlchemy, 型安全
- URL: https://tomodahinata.com/blog/fastapi-sqlmodel-database-crud-relationships-production-guide

## 要点

- SQLModelはPydantic（バリデーション）とSQLAlchemy（ORM）を同一作者tiangoloが統合したライブラリ。1つのモデルが『テーブル定義』と『バリデーションスキーマ』を兼ねられ、FastAPIネイティブで重複が少ない
- テーブルモデルは class Hero(SQLModel, table=True)。Field(default=None, primary_key=True)・Field(index=True)。セッションは get_session の yield 依存に閉じ、SessionDep = Annotated[Session, Depends(get_session)] で全エンドポイントが再利用する
- 本番の肝は『複数モデルで境界を守る』こと。HeroBase/Hero(table=True)/HeroPublic/HeroCreate/HeroUpdateを分け、入力にidを含めさせない・公開にsecret等を漏らさない。更新は hero.sqlmodel_update(data.model_dump(exclude_unset=True))
- リレーションは Relationship(back_populates=...) と Field(foreign_key=...)。response_modelにHeroPublicWithTeam等を指定して関連を含めて返す。循環は『…WithTeam』を入れ子に使わないことで断つ
- 本番化：SQLite→PostgreSQL＋非同期（AsyncEngine/AsyncSession・asyncpg）、create_allは本番非推奨でAlembic移行、接続プール、selectinloadでN+1回避、トランザクション境界、冪等upsert、dependency_overridesでテスト隔離

---

FastAPI で「DB を載せたい」となったとき、最初の分岐は **ORM の選択**です。生の SQLAlchemy か、Tortoise か、はたまた SQLModel か。FastAPI 公式チュートリアルが採用しているのは **SQLModel** で、これには明確な理由があります——**FastAPI・Pydantic・SQLModel はすべて同じ作者（Sebastián Ramírez / tiangolo）が作っており、設計思想が一直線に揃っている**からです。

この記事は、その SQLModel で**リレーショナルDBを本番品質で扱う**ためのガイドです。FastAPI 公式の SQL チュートリアルと SQLModel 公式の流れを**最新仕様に忠実**に追いながら、公式が「学習用」と断っている部分（SQLite・`create_all`・グローバルなセッション）を、**実運用に耐える形**へ引き上げます。題材として、私が構築した [AI動画ローカライズ・リップシンク基盤](/case-studies/ai-video-localization-lipsync)（FastAPI + Celery + **SQLAlchemy async** で、再開可能・冪等な GPU パイプラインを設計）での判断も交えます。

> **この記事のルール**：API・推奨は **FastAPI / SQLModel 公式ドキュメント（2026年6月時点）** に基づきます。本番前に最新確認を。SQLModel は Pydantic と SQLAlchemy を統合したライブラリで、仕様は両者の更新に追随して改定されます。**DB の接続URL・認証情報はすべて環境変数前提**（ハードコード厳禁）で扱います。

---

## 0. まず判断：いつ SQLModel を使い、いつ生の SQLAlchemy にするか

実装の前に、ツール選択を済ませます。ここを曖昧にすると、後で「SQLModel に無理をさせて生 SQLAlchemy を呼び出す」ハイブリッドの泥沼にはまります。

### 0.1 SQLModel の正体：Pydantic + SQLAlchemy の統合

SQLModel 公式の言葉どおり——**SQLModel は Pydantic と SQLAlchemy を土台にした（"powered by Pydantic and SQLAlchemy"）ライブラリ**で、**FastAPI と同じ作者**が作っています。この出自が、決定的な利点を生みます。

通常、FastAPI で DB を扱うと**モデルが二重化**します。「SQLAlchemy のテーブルモデル」と「Pydantic のリクエスト/レスポンススキーマ」を**別々に書き、相互に変換**する——これは典型的な DRY 違反であり、フィールドを足すたびに2箇所を直す手間が生まれます。

SQLModel は、**1つのクラスがテーブル定義とバリデーションスキーマを兼ねられる**ことで、この重複を畳みます。

```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 使い分け：SQLModel か、生 SQLAlchemy 2.0 か

「統合は便利」ですが、万能ではありません。意思決定の表を置きます。

| 観点 | SQLModel（本記事） | 生の SQLAlchemy 2.0 |
| --- | --- | --- |
| 向くケース | FastAPIネイティブ・単一の真実源で素早く作りたい・モデルの重複を消したい | 複雑なクエリ（CTE・ウィンドウ関数・複雑JOIN）・既存のSQLAlchemy資産・細かいマッピング制御 |
| モデル | 1クラスがテーブル＋スキーマを兼ねる（重複が少ない） | ORMモデルとPydanticスキーマを分けて書く（自由度が高い） |
| 学習コスト | Pydantic/FastAPIを知っていれば低い | ORM固有の概念が多く、習得に時間がかかる |
| 逃げ道 | 必要なら**内部のSQLAlchemyにそのまま降りられる**（同じ`select`が使える） | SQLModelより低レベルだが完全な制御が手に入る |

判断の目安：**FastAPI 中心で、CRUD と素直なリレーションが主体なら SQLModel**。**複雑なクエリや既存資産が重いなら生 SQLAlchemy**。重要なのは、SQLModel は **SQLAlchemy の薄いラッパ**なので、必要になれば**同じエンジン・同じ `select` 文で低レベルへ降りられる**点です——一方通行の囲い込みではありません。生 SQLAlchemy の型付き設計は[SQLAlchemy 2 の型付きORM本番ガイド](/blog/sqlalchemy-2-typed-orm-production-guide)を併読してください。本記事は「SQLModel を本番品質で使い切る」道を扱います。

---

## 1. モデルを定義する：`table=True` と `Field`

土台はモデル定義です。SQLModel は**Python の型ヒントをそのままテーブル定義に翻訳**します。

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

押さえるべき点は3つです。

- **`table=True`** が「このクラスは DB テーブルである」という宣言です。これが**無い**クラスは、テーブルを作らない**ただのデータモデル（Pydantic 相当）**になります。この差が、第4章の「複数モデルで境界を守る」設計の核心になります。
- **`Field(default=None, primary_key=True)`** で主キーを定義します。`id` を `int | None` にするのは、**DB が採番するまで `id` は未確定（`None`）だから**です。型が「まだ無いこともある」という現実を正直に表しています。
- **`Field(index=True)`** はインデックスを張ります。`WHERE name = ...` のように**頻繁に絞り込む列**に付けると検索が速くなります。逆に、書き込みのたびにインデックス更新コストが乗るので、**何にでも付けない**のが本番の勘所です（コスト効率）。

> **`Field` は Pydantic と SQLAlchemy の両方を兼ねる**：SQLModel の `Field` は、Pydantic のバリデーション（`max_length` など）と SQLAlchemy のカラム属性（`primary_key` / `index` / `foreign_key`）を**1つの宣言で両立**させます。これが「単一の真実源」の正体です。

---

## 2. エンジンとセッション依存（`yield`）

モデルができたら、DB への接続（エンジン）と、リクエストごとの作業単位（セッション）を用意します。

### 2.1 エンジンを作る

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

エンジンは**アプリ全体で1つ**だけ作ります。内部に**接続プール**を持つので、リクエストごとに作り直してはいけません（コスト・性能の両面で事故）。

### 2.2 テーブルを作る（※本番では `create_all` を使わない）

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

> **【最重要】`create_all` は本番に持ち込まない**：`SQLModel.metadata.create_all` は「テーブルが無ければ作る」だけで、**既存テーブルの変更（列追加・型変更）はしません**。スキーマが育つ本番では、**Alembic によるマイグレーション**が必須です（第7.3章で扱います）。`create_all` は「初回・テスト・ローカル」専用と割り切ってください。

起動時にテーブルを用意する配線は、`lifespan`（推奨）でやります。公式チュートリアルは `@app.on_event("startup")` を使いますが、これは非推奨化された旧APIで、**現行は `lifespan` コンテキストマネージャ**が正道です（[lifespan の本番設計は本番運用ガイド](/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 セッションを `yield` 依存に閉じ込める

ここが FastAPI × SQLModel の最重要パターンです。**セッション（DB との作業単位）を依存に切り出し、`yield` で「使い終わったら確実に閉じる」**を保証します。

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

`with Session(engine) as session:` と `yield` の組み合わせが効くのは、**ハンドラが成功しても例外を投げても、必ずセッションがクローズされる**からです。接続リーク——本番で最も静かに、最も致命的に効いてくる事故——を、**構造で防ぎます**。各ハンドラは `session: SessionDep` と書くだけで、検証済みのセッションを受け取れます。

---

## 3. CRUD：作成・取得・一覧

セッションが手に入れば、CRUD は素直です。SQLModel のメソッド名は SQLAlchemy 由来で、一度覚えれば直感的です。

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

3手順の意味を理解すると、ハマりません。

- **`add`**：オブジェクトをセッションの「保留中」に載せる。この時点では DB は変わりません。
- **`commit`**：トランザクションを確定し、INSERT を実行します。
- **`refresh`**：**DB が採番した `id`** や、DB 側のデフォルト値を、オブジェクトに**読み戻します**。これを省くと、返した `hero.id` が `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
```

主キーで1件取るなら **`session.get(Hero, hero_id)`** が最短です。`select` を書く必要はありません。

### 3.3 一覧取得（Read many）—— 必ずページネーションする

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

> **一覧に上限の無い `limit` を作らない（コスト・可用性）**：`select(Hero).all()` を**無制限**で公開すると、テーブルが100万行に育った日に、1リクエストが DB とメモリを食い潰します。**`offset`/`limit` でのページネーションを最初から必須**にし、`Query(le=100)` で**上限をサーバーが固定**してください。「クライアントが大量取得を要求できる」状態は、性能の問題であると同時にセキュリティの問題（リソース枯渇攻撃）です。

`session.exec(select(...))` が SQLModel の検索の基本形です。`select` は SQLAlchemy の `select` そのものなので、複雑化したらそのまま低レベルへ降りられます（第0章の「逃げ道」）。

---

## 4. 複数モデルで「境界」を守る（本番品質の分かれ目）

ここまでは `Hero` 一つで通しました。が、**本番では1つのモデルを使い回してはいけません**。理由は2つ、どちらもセキュリティに直結します。

1. **入力に `id` を含めさせてはいけない**——`id` は DB が採番するもの。クライアントが `id` を送れる API は、他人のレコードを上書きする攻撃の入口になります。
2. **公開レスポンスに機密を漏らしてはいけない**——`secret_name`（ここでは秘匿フィールドの代表例）や `hashed_password` を、レスポンスにそのまま含めてはいけません。

SQLModel はこれを**モデルの分割**で解決します。**テーブルモデルは1つ、データモデルは用途ごと**に分け、共通部分を `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
```

設計の意図を、公式の言葉で確認します。

- **`HeroCreate` に `id` が無い**のは、公式いわく「**`id` は DB が自動採番するので、クライアントに送らせたくない**」から。これが**ID 注入を構造的に防ぎます**。
- **`HeroPublic` の `id` が必須（`int`、`None` 不可）**なのは、公式いわく「**レスポンスは常に DB から来たモデルなので、必ず `id` を持つ**」から。型が「公開時には必ず採番済み」という事実を保証します。
- **`secret_name` は `Hero`（テーブル）と `HeroCreate`（入力）にはあるが、`HeroPublic`（公開）には無い**。これが**型による漏洩防止**です。レスポンスを `HeroPublic` に通せば、`secret_name` は**構造的に出力されません**。

これを CRUD に適用します。`response_model` で「公開の形」を、引数で「入力の形」を固定します。

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

> **二層スキーマ境界はセキュリティ設計そのもの**：`Hero.model_validate(hero)` で「入力 → テーブル」、`response_model=HeroPublic` で「テーブル → 公開」を変換する——この**入口と出口の二重の関所**が、ID 注入と機密漏洩を同時に塞ぎます。私が認証ガイドで触れた「`hashed_password` を公開モデルに含めない」のと**全く同じ原則**で、SQLModel ではこれがフレームワークの標準作法になっています（[境界バリデーションの考え方は Pydantic ガイド](/blog/pydantic-v2-production-validation-type-safety)）。

---

## 5. 更新と削除：`exclude_unset` で「送られた値だけ」を反映する

更新（PATCH）には、**部分更新の罠**があります。「`age` だけ更新したいのに、送らなかった `name` が `None` で上書きされた」——これは部分更新で最も多い事故です。SQLModel は `exclude_unset` でこれを正しく扱います。

### 5.1 更新用モデルは「全フィールド任意」

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

### 5.2 PATCH パスオペレーション

```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` が部分更新の正解**：`model_dump(exclude_unset=True)` は、**クライアントが明示的に送ったフィールドだけ**を辞書化します。送らなかったフィールドは辞書に**現れない**ので、`sqlmodel_update` は既存値を保ちます。これにより「`age` だけ送れば `age` だけ変わる」が正しく成立します。逆に `exclude_unset` を**忘れる**と、未送信フィールドがデフォルト値（`None`）で全部潰れます——本番事故の典型なので、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}
```

> **本番では「物理削除」を疑う**：監査・誤操作復旧・参照整合性の観点で、**論理削除（`deleted_at` 列を立てる soft delete）**が適切な場面は多いです。`session.delete` で物理削除する前に、「本当に消してよいデータか」を要件と突き合わせてください。動画ローカライズ基盤でも、ジョブ履歴は**消さずに状態遷移で残す**設計にしました（後述）。

---

## 6. リレーション：`Relationship` と `foreign_key`

テーブル間のつながりは、**外部キー（`foreign_key`）**と**リレーション（`Relationship`）**の2点セットで表します。`Team`（チーム）と `Hero`（所属するヒーロー）を例にします。

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

役割を分けて理解します。

- **`Field(foreign_key="team.id")`** は**DB の外部キー制約**です。`team_id` が必ず実在する `team.id` を指すことを、**DB が保証**します（参照整合性）。
- **`Relationship(back_populates=...)`** は**Python オブジェクト上の便利な参照**です。`hero.team` で所属チームに、`team.heroes` で所属ヒーロー一覧に、**SQL を書かずにアクセス**できます。`back_populates` で両者を結ぶと、片方を変えれば他方も整合します（双方向）。

### 6.1 関連を含めて返す：`response_model` の入れ子

関連データを含めて返すには、**公開モデルを入れ子**にします。ここに循環参照の罠があるので、公式の作法に従います。

```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 の形で返る
```

> **循環参照を構造で断つ**：`HeroPublicWithTeam` の `team` を `TeamPublicWithHeroes` にすると、「ヒーロー→チーム→ヒーロー→…」と**無限に展開**します。だから**入れ子に使う側は、必ず1段浅い公開モデル（`TeamPublic` / `HeroPublic`）にする**——これが公式の作法です。「どこまで関連を含めて返すか」を**型で1段に固定**することで、レスポンスの肥大化も防げます。

---

## 7. 本番化：SQLite を卒業し、非同期・Alembic・プール・N+1に向き合う

ここまでが「動く DB アプリ」です。ここからが**落ちない・漏れない・速い**にするための差分。公式チュートリアルが SQLite で止めている先を埋めます。

### 7.1 SQLite → PostgreSQL

SQLite は学習・組み込みには最適ですが、**並行書き込み・ネットワーク越しのアクセス・本番の負荷**には向きません。本番は **PostgreSQL** が定番です。差し替えは接続URLとプール設定が中心です。

```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} は不要。
```

> **接続プールは「コスト」と「可用性」の両方に効く**：DB 接続の確立は重い処理です。プールで**接続を使い回す**ことで、レイテンシと DB 負荷を下げます。`pool_pre_ping=True` は、ロードバランサや DB のアイドル切断で**死んだ接続を掴む事故**を防ぐ、本番でほぼ必須の設定です。サーバーレスや RDS Proxy 経由など、環境ごとにプールの最適値は変わるので、**接続数の上限を DB 側の `max_connections` と必ず突き合わせて**ください。

### 7.2 非同期：AsyncEngine / AsyncSession（asyncpg）

FastAPI は非同期フレームワークです。DB アクセスが**I/O 待ちで他のリクエストを止めない**よう、本番では**非同期ドライバ**（PostgreSQL なら `asyncpg`）と **`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()
```

> **非同期にするなら最後まで非同期で（混在は事故）**：非同期ハンドラの中で**同期のブロッキング DB 呼び出し**をすると、イベントループを止めて**全リクエストが詰まります**。`async def` のハンドラでは `await session.execute(...)` を使い、`def`（同期）ハンドラでは同期セッションを使う——**同期と非同期を混ぜない**のが鉄則です。動画ローカライズ基盤では、API 層を **SQLAlchemy async** で統一し、重い GPU 処理は同期の Celery ワーカー側へ完全に分離しました（I/O は非同期、CPU/GPU 負荷はワーカーへ、という境界）。

### 7.3 マイグレーション：`create_all` を捨てて Alembic へ

第2.2章で予告したとおり、**本番で `create_all` は使いません**。スキーマは育つ——列を足し、型を変え、インデックスを張り直す——その変更履歴を**安全に・後戻り可能に**適用するのが Alembic です。

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

SQLModel のモデルは内部で SQLAlchemy のメタデータを持つので、Alembic の `target_metadata = SQLModel.metadata` を指すだけで**自動生成（autogenerate）**が効きます。本番では**ダウンタイム無しの移行手順**（列追加は後方互換、削除は二段階、など）が重要で、これは[Alembic ゼロダウンタイム移行ガイド](/blog/alembic-zero-downtime-migrations-sqlalchemy)に詳しくまとめています。

### 7.4 N+1 問題を潰す：`selectinload` / `joinedload`

リレーションの**最大の落とし穴が N+1 問題**です。「ヒーロー100件を取り、各ヒーローの `team` にアクセスしたら、チーム取得の SQL が**追加で100回**飛んだ」——これが N+1。1+N 回のクエリで DB を殴ります。

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

判断の目安：**1対多を含めるなら `selectinload`（追加1クエリ・IN句でまとめ取り）**、**多対1を含めるなら `joinedload`（JOIN で1クエリ）**が基本です。関連を `response_model` で返すエンドポイントでは、**必ず eager load を付ける**——「`response_model` に関連を入れたら eager load を足す」をセットで習慣化してください。

### 7.5 トランザクション境界と冪等な upsert

複数の書き込みを**まとめて成功か失敗か**にするのがトランザクションです。`commit` の単位が、そのまま「全部反映 or 全部巻き戻し」の境界になります。

```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()         # ここで初めて全変更が一括確定する（途中例外なら何も残らない）
```

**冪等性**も本番の必須要件です。「同じリクエストが二回来ても、結果が一回分」になるよう、自然キーでの **upsert（無ければ挿入・あれば更新）**を使います。

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

> **冪等性は「再開可能なパイプライン」の心臓**：[AI動画ローカライズ基盤](/case-studies/ai-video-localization-lipsync)では、**DB をジョブ状態の唯一の真実源**とし、全段階を非同期タスク化して**各段階の状態と失敗した段階を DB で管理**しました。長尺動画は**セグメントに分割**し、最後に完了したセグメントから**再開**できる設計です。さらに**キャッシュキーを「入力＋エンジン・チューニング」から導出**し、同条件の再実行は**冪等に既存結果を再利用**します。これらは全て「DB の状態を信頼の源にし、書き込みを冪等に保つ」という、この章の原則の実地応用でした。

---

## 8. テスト：テスト用エンジンと `dependency_overrides`

検証パスのない本番投入はありません。FastAPI × SQLModel は、**セッション依存を丸ごと差し替えられる**ので、DB 付き API も決定的にテストできます。

### 8.1 インメモリSQLite＋依存差し替え

```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          # 公開モデルが機密を漏らさないことを検証
```

ポイントは2つ。**(1) テスト用に独立したインメモリエンジン**を立てるので、本番 DB を汚しません。`StaticPool` で接続を共有しないと、インメモリ DB はクエリごとに作り直されてテーブルが消えます。**(2) `dependency_overrides[get_session]`** で、ハンドラの中身を一切変えずにセッションだけ差し替えます。

### 8.2 トランザクションのロールバックでテストを高速・隔離する

テストごとにテーブルを作り直す代わりに、**各テストをトランザクションで包み、終わったらロールバック**する戦略もあります。DB の状態がテスト間で**自動的に巻き戻る**ので、隔離が速く確実です。

```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` は『テストの最大の見返り』**：本番コードに「テスト用の分岐」を1行も入れずに、**外側からセッションを注入**できます。これは第2.3章で「セッションを依存に閉じ込めた」ことの直接の回収です。`finally`/`clear()` で**毎回オーバーライドを掃除**しないと、テスト間が汚染されて偽陽性・偽陰性を生むので、片付けは必須です。

---

## 9. まとめ：本番 FastAPI × SQLModel チートシート

迷ったときの早見表です。

- **正体**：SQLModel ＝ **Pydantic（検証）＋ SQLAlchemy（ORM）を同一作者が統合**。1クラスがテーブル＋スキーマを兼ね、FastAPI ネイティブで重複が少ない。複雑なクエリや既存資産が重いなら**生 SQLAlchemy**（必要なら SQLModel から降りられる）。
- **モデル**：`class Hero(SQLModel, table=True)`。`Field(default=None, primary_key=True)`／`Field(index=True)`。`id` は採番前 `None` を許す `int | None`。
- **エンジン/セッション**：エンジンはアプリに1つ（プール内蔵）。SQLite は `connect_args={"check_same_thread": False}`。セッションは `get_session` の **`yield` 依存**に閉じ、`SessionDep = Annotated[Session, Depends(get_session)]` で再利用。
- **CRUD**：作成は `add`→`commit`→`refresh`（採番した id を読み戻す）。単一は `session.get(Hero, id)`。一覧は `session.exec(select(Hero).offset().limit()).all()`、**`limit` は `Query(le=100)` で上限固定**。
- **境界**：`HeroBase`／`Hero(table=True)`／`HeroPublic`／`HeroCreate`／`HeroUpdate` を分ける。**入力に `id` を含めさせない・公開に機密を載せない**。`Hero.model_validate(入力)` と `response_model=公開` の**二重の関所**。
- **更新/削除**：PATCH は `hero.sqlmodel_update(data.model_dump(exclude_unset=True))`（送った値だけ反映）。削除は本番では論理削除も検討。
- **リレーション**：`Field(foreign_key="team.id")`＋`Relationship(back_populates=...)`。返すときは `HeroPublicWithTeam` 等の入れ子だが、**入れ子の中は1段浅い公開モデル**で循環を断つ。
- **本番化**：SQLite→**PostgreSQL**、**非同期（AsyncEngine/AsyncSession・asyncpg）は最後まで非同期で**。`create_all` は捨て **Alembic**。接続プール（`pool_pre_ping`）。**N+1 は `selectinload`/`joinedload`**。トランザクション境界と**冪等 upsert**。
- **テスト**：インメモリSQLite＋`StaticPool`、`dependency_overrides[get_session]` で隔離。ロールバック戦略でさらに速く。

---

SQLModel は「Pydantic を知っていれば、その日のうちに DB が載る」ほど立ち上がりが速いライブラリです。けれど本番品質は、ここでも**境界の設計**で決まります。**入力と公開でモデルを分け、セッションを依存に閉じ、`create_all` を Alembic に置き換え、関連には eager load を、書き込みには冪等性を**——どれも派手ではありませんが、この積み重ねが「漏れない・落ちない・変更しやすい DB 層」を作ります。

私は [AI動画ローカライズ・リップシンク基盤](/case-studies/ai-video-localization-lipsync)で、**FastAPI + Celery + SQLAlchemy async** を中核に、**DB をジョブ状態の唯一の真実源**とした再開可能・冪等な GPU パイプラインを設計しました。全段階の状態と失敗段階を DB で管理し、長尺動画はセグメント単位で再開し、同条件の再実行はキャッシュで冪等に畳む——「状態を DB に集約し、書き込みを冪等に保つ」という、この記事の原則を実地で回した案件です。生成AI（Claude Code）を相棒に、**一人で速く・安く**作りつつ、テストと検証ゲートで品質を担保するのが私の進め方です。

**「FastAPI に DB を載せたいが、SQLModel か生 SQLAlchemy か、モデル境界やマイグレーション、非同期や冪等性をどう設計すべきか」——その判断から実装・移行・運用まで、一気通貫で伴走します。** 要件整理の段階からでも、お気軽にご相談ください。

---

### 参考（公式ドキュメント）

- [SQL (Relational) Databases（FastAPI）](https://fastapi.tiangolo.com/tutorial/sql-databases/) — SQLModel での `table=True`・`get_session` 依存・CRUD・複数モデル
- [SQLModel（公式トップ）](https://sqlmodel.tiangolo.com/) — Pydantic × SQLAlchemy 統合の思想と基本の `Hero` モデル
- [SQLModel with FastAPI（チュートリアル）](https://sqlmodel.tiangolo.com/tutorial/fastapi/) — エンジン・セッション・起動時のテーブル作成
- [Multiple Models with FastAPI（SQLModel）](https://sqlmodel.tiangolo.com/tutorial/fastapi/multiple-models/) — `HeroBase`/`HeroCreate`/`HeroPublic`/`HeroUpdate` の境界設計
- [Relationships in FastAPI（SQLModel）](https://sqlmodel.tiangolo.com/tutorial/fastapi/relationships/) — `Relationship(back_populates=)`・`foreign_key`・`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 ドキュメント](https://docs.sqlalchemy.org/) — エンジン・接続プール・`selectinload`/`joinedload`・非同期
- [Alembic ドキュメント](https://alembic.sqlalchemy.org/) — マイグレーション（`create_all` の本番代替）