# FastAPI 本番運用ガイド：async の正しい使いどころ・Pydantic v2 境界バリデーション・DIと可観測性で落ちないAPIを作る

> FastAPI を本番品質で運用する実装ガイド。公式ドキュメントに忠実な async def / def の使い分け、Pydantic v2 の境界バリデーション、Depends による依存性注入、構造化ログと OpenTelemetry の可観測性、BackgroundTasks の限界とタスクキューへの逃がし方、テストとデプロイまでを実コードで解説します。

- 公開日: 2026-06-24
- 著者: 友田 陽大
- タグ: Python, FastAPI, アーキテクチャ設計, 可観測性, パフォーマンス, 型安全
- URL: https://tomodahinata.com/blog/fastapi-production-async-pydantic-observability-guide

## 要点

- 最大の落とし穴はasync def / defの取り違え。awaitするasyncライブラリはasync def、同期処理はdefでスレッドプールへ逃がす。迷ったらdef
- Pydantic v2で外部入力を境界で殺す。入力・出力モデルを分け、外部APIのJSONもmodel_validateで即モデル化する
- DBセッション・認証・設定はDepends＋Annotatedで注入し、後始末はyield依存で。テストはdependency_overridesで差し替える
- DBプール・モデル・HTTPクライアントはlifespanで起動時に一度だけ確保しapp.stateで共有する（on_eventは非推奨）
- 重い処理（数十秒以上・リトライ必須・CPU/GPU集約）はBackgroundTasksを卒業し、APIは202 Acceptedで薄く返してジョブ基盤へ逃がす

---

「FastAPI で API を立てたい」——要件としては一行です。`@app.get` を書けば 5 分で動くものができます。けれど**本番**に載せようとした瞬間、判断すべきことが一気に増えます。**`async def` と `def` のどちらで書くのか。外部入力をどこで検証するのか。DB セッションをどう注入するのか。重い AI ジョブをリクエストの中で回していいのか。落ちたとき、ログから原因を追えるのか。**

この記事は、FastAPI を**本番品質**で運用するための実装ガイドです。公式チュートリアルが「動かす」ところまでを丁寧に教えてくれるのに対し、ここでは「**落ちない・追える・変更しやすい**」を作るための判断基準とコードに焦点を当てます。題材として、私が国内大手放送事業者向けに構築した社内AIプラットフォーム（[テロップ誤字検出パイプライン](/case-studies/broadcaster-ai-content-platform)。FastAPI を API 層に据え、長時間の AI ジョブを Cloud Run Jobs / Cloud Workflows へ切り離して運用）での設計判断も交えます。

> **この記事のルール**：API 仕様・推奨パターンは **FastAPI / Pydantic 公式ドキュメント（2026年6月時点）** に基づきます。仕様は改定されるため、本番投入前に必ず公式ドキュメントで最新の挙動を確認してください。コードは実運用で使える形に整えていますが、**シークレット（DB URL・APIキー・署名鍵）は環境変数前提**です（ハードコード厳禁）。

---

## 0. FastAPI が「速い・安全」と言える根拠を一行で

設計判断の前に、FastAPI が何の上に立っているかを押さえます。公式は次の3点を土台に挙げています。

- **標準の Python 型ヒント**：パラメータの型を書くだけで、バリデーション・変換・OpenAPI ドキュメント生成が自動で付いてくる。
- **Pydantic**：データのバリデーションと解析。コア検証ロジックは **Rust** で書かれており、公式は「Python で最速級のデータ検証ライブラリの一つ」と述べています。
- **Starlette**：ASGI の Web 層。公式の表現で「**NodeJS や Go と同等**の高性能」を支える部分。

つまり FastAPI の「速い」「バグが減る」は、**型ヒント一本で検証・変換・ドキュメントを同時に得る**設計から来ています。本記事はこの設計思想を、本番運用の文脈で最大限に活かすことを目指します。

---

## 1. 最重要：`async def` と `def` の正しい使い分け

ここを間違えると、**スループットが出ない API** か、**たまにフリーズする API** ができあがります。FastAPI 最大の落とし穴であり、最大のレバレッジです。

### 1.1 公式が定める「急いでいる人向け」ルール

公式ドキュメント（`/async/`）の判断基準を、そのまま引用します。

1. **`await` で呼べと指示された外部ライブラリを使うなら → `async def`**

   ```python
   @app.get("/")
   async def read_results():
       results = await some_library()
       return results
   ```

2. **外部と通信するが `await` 非対応のライブラリ（多くの DB ライブラリが現状これ）なら → 普通の `def`**

   ```python
   @app.get("/")
   def results():
       results = some_library()
       return results
   ```

3. **何とも通信せず待つ必要がないなら → `async def`（中で `await` を使わなくてよい）**

4. **判断が付かないなら → 普通の `def`**

### 1.2 なぜ `def` でも遅くならないのか

普通の `def` でパスオペレーション関数を書くと、FastAPI はそれを**外部スレッドプールで実行して await** します（直接呼ぶとサーバーがブロックするため）。これはパスオペレーション関数だけでなく**依存（Depends）にも適用**されます。

ここが他の async フレームワークと決定的に違う点です。

- **`def` のブロッキング呼び出し**：FastAPI がスレッドプールに逃がすので、イベントループは止まらない。
- **`async def` の中のブロッキング呼び出し**：そのままイベントループ上で走るので、**全リクエストが固まる**。

### 1.3 一番多い事故：`async def` の中で同期ブロッキングを呼ぶ

これが現場で最も頻発する、そして最も気づきにくいバグです。

```python
import time
import requests  # 同期ライブラリ（await できない）

@app.get("/bad")
async def bad_endpoint():
    # ❌ async def の中で同期ブロッキング I/O を呼んでいる
    # この requests.get が返るまで、サーバー全体の処理が止まる
    res = requests.get("https://slow-upstream.example.com/data")
    time.sleep(1)  # ❌ これも同様にイベントループを丸ごと止める
    return res.json()
```

`async def` と宣言しているのに中身は同期です。1 リクエストの `requests.get` / `time.sleep` の間、**同じワーカーが捌くはずの他の全リクエストがブロックされます**。負荷試験で「同時接続を上げた途端にレイテンシが崩壊する」典型がこれです。

修正の方針は2つ。**(A) `def` にしてスレッドプールに任せる**か、**(B) `async` 対応ライブラリに替えて `await` する**か。

```python
import anyio
import httpx  # async 対応の HTTP クライアント

# (A) 同期ライブラリを使い続けるなら def にする（FastAPIがスレッドプールへ逃がす）
@app.get("/ok-sync")
def ok_sync_endpoint():
    res = requests.get("https://slow-upstream.example.com/data")
    return res.json()

# (B) async ネイティブのライブラリで正しく await する
@app.get("/ok-async")
async def ok_async_endpoint():
    async with httpx.AsyncClient() as client:
        res = await client.get("https://slow-upstream.example.com/data")
    return res.json()

# どうしても async def の中で重い同期処理を呼ぶしかないときは、明示的にスレッドへ逃がす
@app.get("/ok-offload")
async def ok_offload_endpoint():
    # anyio.to_thread.run_sync でブロッキング関数をスレッドプールへ
    data = await anyio.to_thread.run_sync(blocking_cpu_or_io_work)
    return data
```

### 1.4 決定表：`async def` か `def` か

| あなたの I/O | 関数の宣言 | 理由 |
| --- | --- | --- |
| `await` で呼ぶ async ライブラリ（`httpx`, `asyncpg`, async SDK） | `async def` | ネイティブに非ブロッキング。`await` できる |
| 同期 DB ドライバ・`requests`・`time.sleep`・同期 SDK | `def` | FastAPI がスレッドプールへ逃がし、ループを止めない |
| CPU バウンドな重い計算（画像処理・暗号・ML 推論） | `def`（または別プロセス/キュー） | スレッドでも GIL の影響あり。重ければ第7章のキューへ |
| 外部と通信しない純粋なロジック | `async def` | 待ちがないので何でも可。公式は async を推奨 |
| 判断が付かない | `def` | 公式の「迷ったら def」。事故率が最も低い |

> **設計の指針（KISS）**：「全部 `async def` にすれば速い」は誤りです。**`async def` は『中で本当に `await` するとき』にだけ使う**。中身が同期なら素直に `def` にして FastAPI に任せるのが、最も事故が少なく速い選択です。

---

## 2. Pydantic v2：外部入力を境界で殺す

API の本質は「**外の世界から来る信用できないデータ**を、内側の信用できる型に変換する関所」です。Pydantic v2 はその関所を型ヒント一本で作れます。**外部入力は一切信用しない**——これがセキュリティの第一原則です。

### 2.1 リクエスト／レスポンスモデルを分ける

入力用と出力用のモデルは**別物**にします。入力に `id` や `created_at` を含めさせない、出力にパスワードハッシュを漏らさない——これは責務分離（SRP）であり、セキュリティでもあります。

```python
from datetime import datetime
from pydantic import BaseModel, EmailStr, Field, field_validator

# --- 入力（クライアントが送ってよいものだけ） ---
class UserCreate(BaseModel):
    email: EmailStr                                  # 形式不正は422で弾かれる
    display_name: str = Field(min_length=1, max_length=50)
    age: int = Field(ge=0, le=150)                   # 範囲制約（負の年齢を入れさせない）

    @field_validator("display_name")
    @classmethod
    def no_control_chars(cls, v: str) -> str:
        # ドメイン固有のルールはvalidatorで明示的に検証する
        if any(ord(c) < 0x20 for c in v):
            raise ValueError("制御文字は使用できません")
        return v.strip()

# --- 出力（サーバーが返してよいものだけ。秘密は含めない） ---
class UserPublic(BaseModel):
    id: int
    email: EmailStr
    display_name: str
    created_at: datetime
    # password_hash は意図的に含めない → 漏洩を型で防ぐ
```

`Field` の制約（`min_length` / `max_length` / `ge` / `le` / `gt` など）と `field_validator` を使うと、**ビジネスルールがモデル定義に集約**されます。検証ロジックがハンドラ本体に散らばらないので、読みやすく変更しやすい（ETC）。

### 2.2 ハンドラで使う：`response_model` で出力も型で縛る

```python
from fastapi import FastAPI

app = FastAPI()

@app.post("/users", response_model=UserPublic, status_code=201)
async def create_user(payload: UserCreate) -> UserPublic:
    # payload はここに来た時点で「検証済み」。中で再検証は不要（DRY）
    user = await repository.insert_user(payload)
    # response_model=UserPublic により、余分なフィールドは自動で削ぎ落とされる
    return user
```

`payload: UserCreate` と書くだけで、不正なリクエストは**ハンドラに到達する前に 422 で弾かれます**。検証コードをハンドラに一行も書かなくていい——これが「型ヒント一本で検証が付く」の実体です。`response_model` を付ければ、出力側も契約どおりの形に整形され、秘密フィールドの漏洩を構造的に防げます。

### 2.3 「型でない場所」から来るデータ：`model_validate` / `model_validate_json`

外部 API のレスポンス、メッセージキューのペイロード、設定ファイル——FastAPI のハンドラ引数を通らないデータも、**内側に入れる前に必ず検証**します。

```python
import httpx

async def fetch_external_user(user_id: str) -> UserPublic:
    async with httpx.AsyncClient() as client:
        res = await client.get(f"https://upstream.example.com/users/{user_id}")
        res.raise_for_status()
    # 外部APIのJSONも「信用できない外部入力」。dictで持ち回らず即座にモデル化する
    return UserPublic.model_validate(res.json())     # dict → 検証済みモデル
    # JSON文字列を直接渡すなら model_validate_json(res.text) が一手少ない
```

> **Pydantic v2 の注意**：`strict` モード（型変換しない）と `lax` モード（`"123"` → `123` のように coerce する）があります。**信用できない外部入力には strict を検討**してください。`"true"` が `True` に化けるような暗黙変換が、思わぬバグの温床になります。検証に失敗すると `ValidationError` が**どのフィールドがなぜ落ちたか**を構造化して返すので、エラーログがそのまま原因解析になります。

---

## 3. 依存性注入（Depends）：DBセッション・認証・設定を注入する

`Depends` は FastAPI で最も価値の高い機能です。**「ハンドラが必要とするもの」を宣言すると、フレームワークが用意して注入してくれる**。DB セッション・認証ユーザー・設定を、ハンドラ本体から切り離せます（SRP）。

### 3.1 `Annotated` で型エイリアス化する（公式推奨）

公式は FastAPI 0.95.0 以降、`Annotated` を使う書き方を推奨しています。依存を**型エイリアスに畳んで再利用**するのが定石です。

```python
from typing import Annotated
from fastapi import Depends

# 設定（環境変数から読む。シークレットはここに集約し、ハードコードしない）
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    jwt_secret: str
    class Config:
        env_file = ".env"

from functools import lru_cache

@lru_cache  # 設定は毎回読み直さず、プロセス内で1つ（コスト効率）
def get_settings() -> Settings:
    return Settings()  # 環境変数 DATABASE_URL / JWT_SECRET を読む

SettingsDep = Annotated[Settings, Depends(get_settings)]
```

### 3.2 `yield` 依存でクリーンアップ（DBセッションの定番）

`yield` を使う依存は、**`yield` の前で確保し、レスポンス後に後始末**します。DB セッションのクローズはこのパターンの王道です。

```python
from collections.abc import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

# lifespan で作ったエンジンから sessionmaker を用意（第4章参照）
async def get_db(request: Request) -> AsyncGenerator[AsyncSession, None]:
    session_factory: async_sessionmaker = request.app.state.session_factory
    async with session_factory() as session:   # yield の前 = 確保
        yield session                           # ハンドラへ注入
        # yield の後 = 後始末。async with が確実にクローズする（リーク防止）

DbDep = Annotated[AsyncSession, Depends(get_db)]
```

### 3.3 認証も依存にする

認証は「全ハンドラで繰り返される横断的関心事」。依存に切り出せば、各ハンドラは `current_user` を受け取るだけで済みます（DRY）。

```python
from fastapi import HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

bearer = HTTPBearer()

async def get_current_user(
    creds: Annotated[HTTPAuthorizationCredentials, Depends(bearer)],
    settings: SettingsDep,
    db: DbDep,
) -> UserPublic:
    # トークン検証は外部入力の検証そのもの。失敗は401で即座に返す
    user = await verify_jwt(creds.credentials, settings.jwt_secret, db)
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="認証に失敗しました",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user

CurrentUser = Annotated[UserPublic, Depends(get_current_user)]


@app.get("/me", response_model=UserPublic)
async def read_me(current_user: CurrentUser) -> UserPublic:
    return current_user  # 認証ロジックはハンドラに一切ない
```

依存はハンドラと独立に `async def` でも `def` でも書けます（公式が明言）。**テスト時は丸ごと差し替えられる**（第8章 `dependency_overrides`）——これが DI の最大の見返りです。

---

## 4. lifespan：起動時に一度だけ重い資源を確保する

**DB プール・ML モデル・HTTP クライアントを、リクエストごとに作ってはいけません**。接続確立やモデルロードは秒単位のコストです。公式推奨の `lifespan` で、起動時に一度だけ確保し、ワーカー内で共有します。

```python
from contextlib import asynccontextmanager
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

@asynccontextmanager
async def lifespan(app: FastAPI):
    # --- 起動時（yield の前。一度だけ実行される） ---
    settings = get_settings()
    engine = create_async_engine(settings.database_url, pool_size=10, max_overflow=20)
    app.state.session_factory = async_sessionmaker(engine, expire_on_commit=False)
    app.state.http = httpx.AsyncClient(timeout=10.0)  # 使い回す共有クライアント
    # ML モデルなど重い資源もここでロードして app.state に置く

    yield  # ← ここでアプリが起動し、リクエストを受け始める

    # --- 終了時（yield の後。グレースフルシャットダウン時に一度だけ） ---
    await app.state.http.aclose()
    await engine.dispose()  # コネクションを綺麗に返す（リーク・接続枯渇の防止）

app = FastAPI(lifespan=lifespan)
```

> **重要**：旧来の `@app.on_event("startup")` / `@app.on_event("shutdown")` は**非推奨**です。`lifespan` なら起動と終了のロジックが一箇所にまとまり、`yield` を挟んで状態を共有できます。新規コードでは必ず `lifespan` を使ってください。

放送局向けプラットフォームでも、AI モデルや GCP クライアント・DB プールはすべて `lifespan` で確保し、`app.state` に載せて共有しました。リクエスト毎の確保をやめるだけで、p99 レイテンシとコールドな初動が劇的に安定します（コスト効率・パフォーマンス）。

---

## 5. エラーハンドリング：4xx は即時、5xx は構造化ログ

エラー処理の原則はシンプルです。**外部入力起因のエラー（クライアントが直せるもの）は 4xx で即座に返す。サーバー側の想定外（クライアントが直せないもの）は 5xx を返し、詳細は構造化ログにだけ出す**。クライアントにスタックトレースや内部例外メッセージを返してはいけません（情報漏洩）。

### 5.1 業務エラーは `HTTPException`

```python
from fastapi import HTTPException, status

@app.get("/users/{user_id}", response_model=UserPublic)
async def get_user(user_id: int, db: DbDep) -> UserPublic:
    user = await repository.find_user(db, user_id)
    if user is None:
        # 「存在しない」はクライアントが知るべき情報 → 404で即時
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="ユーザーが見つかりません")
    return user
```

### 5.2 ドメイン例外を集約ハンドラで HTTP に翻訳する

ハンドラの随所で `HTTPException` を投げ散らかすより、**ドメイン層は素のドメイン例外を投げ、境界で HTTP に翻訳**する方が綺麗です（関心の分離）。

```python
import logging
from fastapi import Request
from fastapi.responses import JSONResponse

logger = logging.getLogger("app")

class DomainError(Exception):
    """業務ルール違反。messageはユーザーに見せてよい前提で書く。"""
    def __init__(self, message: str, status_code: int = 400):
        self.message = message
        self.status_code = status_code

@app.exception_handler(DomainError)
async def domain_error_handler(request: Request, exc: DomainError):
    # 4xx: クライアント起因。messageはそのまま返してよい
    return JSONResponse(status_code=exc.status_code, content={"detail": exc.message})

@app.exception_handler(Exception)
async def unhandled_error_handler(request: Request, exc: Exception):
    # 5xx: 想定外。詳細はログにだけ残し、クライアントには汎用メッセージ
    logger.exception("unhandled error", extra={"request_id": getattr(request.state, "request_id", None)})
    return JSONResponse(status_code=500, content={"detail": "内部エラーが発生しました"})
```

`logger.exception` はスタックトレースを**ログにだけ**残し、レスポンスは「内部エラーが発生しました」で固定する。**何が落ちたかはサーバーが知り、クライアントには漏らさない**——これが本番のエラー設計です。

---

## 6. 可観測性：リクエストID・構造化ログ・OpenTelemetry

「落ちないAPI」の半分は「**落ちたときに即座に原因を追えるAPI**」です。本番でログを役立たせる最小装備は3つ——リクエストID・構造化ログ・分散トレーシング。

### 6.1 リクエストID付与ミドルウェア

1リクエストを貫く ID があると、複数行のログを横断して1件の処理を再構成できます。受信ヘッダがあれば引き継ぎ（上流からのトレース継続）、なければ採番します。

```python
import uuid
from starlette.middleware.base import BaseHTTPMiddleware

class RequestIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # 上流が付けた ID を引き継ぐ。なければ採番（分散トレースの連結）
        request_id = request.headers.get("x-request-id") or str(uuid.uuid4())
        request.state.request_id = request_id
        response = await call_next(request)
        response.headers["x-request-id"] = request_id  # 呼び出し側にも返す
        return response

app.add_middleware(RequestIDMiddleware)
```

### 6.2 構造化ログ（JSON）

本番のログは**機械可読**であるべきです。`request_id` / メソッド / パス / ステータス / 所要時間を JSON で吐けば、ログ基盤（Cloud Logging 等）で絞り込み・集計できます。

```python
import json, logging, time

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "level": record.levelname,
            "message": record.getMessage(),
            "request_id": getattr(record, "request_id", None),
        }
        return json.dumps(payload, ensure_ascii=False)

# アクセスログをミドルウェアで（所要時間つき）
class AccessLogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.perf_counter()
        response = await call_next(request)
        elapsed_ms = (time.perf_counter() - start) * 1000
        logger.info(
            "request",
            extra={"request_id": getattr(request.state, "request_id", None)},
        )
        # ⚠️ ここで body・メールアドレス・トークンなどPIIをログに入れない
        response.headers["x-response-time-ms"] = f"{elapsed_ms:.1f}"
        return response
```

> **PIIをログに残さない**：リクエストボディ・メールアドレス・認証トークン・氏名を構造化ログに流すのは事故です。記録するのは**メタデータ（ID・ステータス・所要時間・エラー種別）**だけ。内部統制案件では絶対条件であり、放送局向けプラットフォームでも徹底しました。

### 6.3 OpenTelemetry 連携の要点

複数サービス（FastAPI → DB → 外部API → ジョブ実行基盤）にまたがる処理は、ログだけでは追い切れません。**OpenTelemetry の自動計装**で、リクエストを貫くトレースを取ります。要点はこれだけです。

```python
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

# 起動時に一度だけ。HTTPサーバ・クライアント・DBのスパンが自動で繋がる
FastAPIInstrumentor.instrument_app(app)
```

`opentelemetry-instrumentation-fastapi` を入れて1行計装すれば、**どのスパンで時間を食ったか**がトレースで可視化されます。`request_id` をトレース ID と紐づければ、ログとトレースを相互参照できます。

---

## 7. 重い処理：BackgroundTasks の限界とタスクキューへの逃がし方

ここが本番設計の分水嶺です。「メール送信」「サムネ生成」「AI による長時間ジョブ」——これらをどこで処理するか。

### 7.1 `BackgroundTasks` でできること・できないこと

FastAPI の `BackgroundTasks` は、**レスポンスを返した後に同じプロセスで実行**される仕組みです。「軽く・短く・失敗しても致命的でない」副作用にだけ向いています。

```python
from fastapi import BackgroundTasks

@app.post("/users", status_code=201)
async def create_user(payload: UserCreate, bg: BackgroundTasks, db: DbDep):
    user = await repository.insert_user(db, payload)
    # 軽い後処理ならOK：ウェルカムメール送信など
    bg.add_task(send_welcome_email, user.email)
    return {"id": user.id}
```

**`BackgroundTasks` の限界**は明確です。

- **同一プロセス内で動く**：ワーカーの CPU・メモリを食う。重い処理を載せると API のレイテンシが悪化する。
- **永続化されない**：プロセスが落ちたら**タスクは消える**。リトライも進捗追跡もない。
- **スケールしない**：API のスケール（レプリカ数）と処理能力が結合してしまう。

### 7.2 決定表：`BackgroundTasks` か 本物のタスクキューか

| 観点 | `BackgroundTasks` で十分 | 本物のタスクキュー（Celery / Cloud Run Jobs / Workflows）へ |
| --- | --- | --- |
| 所要時間 | 数百ミリ秒〜数秒 | 数十秒〜数十分の長時間ジョブ |
| 失敗時 | 消えても許容できる | リトライ・再開・進捗追跡が必須 |
| リソース | 軽い I/O（メール・通知） | CPU/GPU 集約（動画・OCR・ASR・LLM） |
| スケール | API と一緒でよい | API と独立にスケールさせたい |
| 可視性 | 不要 | ジョブ単位の状態管理が要る |

判断基準は一行です。**「失敗したらやり直したい」「数十秒以上かかる」「CPU/GPUを食う」のどれかに当てはまったら、プロセス内 `BackgroundTasks` を卒業し、永続キュー / ジョブ実行基盤へ逃がす**（YAGNI の裏返し：必要になるまでキューを入れないが、必要になったら迷わず入れる）。

### 7.3 実例：FastAPI から長時間 AI ジョブを切り離す

放送局向けプラットフォームでは、テロップ誤字検出のために**動画から OCR（画面の文字）と ASR（発話の文字起こし）を抽出して突き合わせる**——どう見ても数分〜数十分かかる重いジョブでした。これを FastAPI のリクエスト内や `BackgroundTasks` で回すのは論外です。API が詰まり、デプロイのたびにジョブが消えます。

そこで構成をこう分離しました。

- **FastAPI（async）**：受付・認証・バリデーション・ジョブの起動とステータス返却に専念。リクエストは即座に返す。
- **Cloud Run Jobs**：実際の重い処理（OCR / ASR / 突き合わせ）を、API とは独立にスケールする実行基盤で走らせる。
- **Cloud Workflows**：複数ステップのオーケストレーション。独立な工程を**並列化**し、逐次 18 分かかっていた処理を約 30%短縮（並列 13 分）しました。

```python
@app.post("/jobs", status_code=202)  # 202 Accepted：受け付けたが完了はしていない
async def enqueue_job(req: JobRequest, current_user: CurrentUser, db: DbDep):
    # APIの仕事は「検証して、ジョブを起動して、追跡用IDを返す」まで
    job = await repository.create_job(db, owner=current_user.id, spec=req)
    await workflows_client.start_execution(job_id=job.id, spec=req.model_dump())
    # 重い処理はここで待たない。クライアントは job.id でステータスをポーリング
    return {"job_id": job.id, "status": "queued"}

@app.get("/jobs/{job_id}")
async def get_job_status(job_id: str, current_user: CurrentUser, db: DbDep):
    job = await repository.find_job(db, job_id, owner=current_user.id)
    if job is None:
        raise HTTPException(404, "ジョブが見つかりません")
    return {"job_id": job.id, "status": job.status, "progress": job.progress}
```

**API は「速く返す薄い層」、重さは外の実行基盤」**——この境界を引くだけで、API は落ちにくくなり、ジョブは独立にスケール・リトライできるようになります。`202 Accepted` + ステータスポーリング（または Webhook）が、非同期ジョブ API の素直な型です。

---

## 8. テスト：TestClient と httpx AsyncClient、依存のオーバーライド

検証パスのない本番投入はありません。FastAPI はテストが書きやすく設計されています。

### 8.1 同期テスト：`TestClient`

公式の基本形は `fastapi.testclient.TestClient` です。**テスト関数は普通の `def`**で、`await` 不要で呼べます（内部で ASGI を駆動）。

```python
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_create_user_validates_input():
    # 不正な入力（age が負）は 422 で弾かれるはず
    res = client.post("/users", json={"email": "x@example.com", "display_name": "Yu", "age": -1})
    assert res.status_code == 422

def test_create_user_ok():
    res = client.post("/users", json={"email": "yu@example.com", "display_name": "Yu", "age": 30})
    assert res.status_code == 201
    assert "id" in res.json()
```

### 8.2 非同期テスト：httpx `AsyncClient` + `ASGITransport`

DB を `await` で叩くような非同期コードを、HTTP 越しでなく直接テストしたいときは、公式が案内するとおり **httpx の `AsyncClient` を `ASGITransport` で**アプリに繋ぎます。

```python
import pytest
import httpx
from httpx import ASGITransport
from app.main import app

@pytest.mark.anyio
async def test_me_async():
    transport = ASGITransport(app=app)  # アプリを直接ドライブ（実ネットワーク不要）
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as ac:
        res = await ac.get("/me", headers={"Authorization": "Bearer faketoken"})
    assert res.status_code in (200, 401)
```

### 8.3 依存のオーバーライド：`dependency_overrides`

DI の最大の見返りがここです。**本物の DB や認証を、テスト用のフェイクに丸ごと差し替える**。外部依存を切り離せるので、テストが速く・決定的になります。

```python
async def override_get_db():
    # 本物の get_db の代わりに、テスト用のインメモリ/トランザクションロールバックなセッションを返す
    async with test_session_factory() as session:
        yield session

def fake_current_user() -> UserPublic:
    return UserPublic(id=1, email="test@example.com", display_name="Test", created_at=datetime.now())

# キーは「元の依存関数」、値は「差し替え関数」
app.dependency_overrides[get_db] = override_get_db
app.dependency_overrides[get_current_user] = fake_current_user

def test_read_me_with_fake_auth():
    res = client.get("/me")
    assert res.status_code == 200
    assert res.json()["email"] == "test@example.com"

# テスト後は必ずクリアして、他テストへの汚染を防ぐ
app.dependency_overrides.clear()
```

認証・DB・外部 API を依存に切り出しておくと、`dependency_overrides` で**ピンポイントに偽装**できます。第3章で認証を `Depends` にした投資が、ここで回収されます。

---

## 9. デプロイ：uvicorn / gunicorn とグレースフルシャットダウン

### 9.1 ワーカー構成

ASGI サーバの `uvicorn` を、本番ではプロセスマネージャの下で複数ワーカーで動かします。古典的には `gunicorn` + `uvicorn.workers.UvicornWorker`、もしくは `uvicorn --workers N` を使います。

```bash
# uvicorn 単体で複数ワーカー（シンプルな構成）
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4

# gunicorn をプロセスマネージャに、各ワーカーを uvicorn で（堅牢な定番）
gunicorn app.main:app \
  --worker-class uvicorn.workers.UvicornWorker \
  --workers 4 \
  --bind 0.0.0.0:8000 \
  --timeout 30 \
  --graceful-timeout 30
```

ワーカー数の目安は **CPU コア数を基準**に、ワークロード（I/O バウンドか CPU バウンドか）で調整します。やみくもに増やすとメモリを食うだけです——`lifespan` で確保した DB プールや HTTP クライアントは**ワーカーごとに複製される**ため、ワーカー数 × プールサイズが DB の最大接続数を超えないよう設計してください（接続枯渇の典型事故）。

### 9.2 グレースフルシャットダウン

デプロイのたびにリクエストを落とさないために、シャットダウンは**穏やかに**。`gunicorn` の `--graceful-timeout` は「処理中のリクエストを待つ猶予」、コンテナ基盤（Cloud Run / Kubernetes）なら `SIGTERM` を受けてから猶予内に既存リクエストを捌き切ってから終了します。このとき第4章の `lifespan` 終了処理（`engine.dispose()` / `http.aclose()`）が走り、**コネクションを綺麗に返してから**落ちます。これが「デプロイ中も 5xx を出さない」ための仕上げです。

> **コスト効率の一言**：I/O バウンドな API はワーカーあたりの同時実行が高く取れるため、`async` を正しく使えば**少ないインスタンスで高いスループット**が出せます。逆に CPU バウンドな処理を API プロセスに抱えると、コア数ぶんしか並列が効かず、無駄にインスタンスを増やす羽目になります。第7章の「重い処理は外へ」は、信頼性だけでなく**請求額**にも直結します。

---

## 10. まとめ：本番 FastAPI チートシート

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

- **`async def` か `def` か**：`await` する async ライブラリ → `async def`。同期ライブラリ・`requests`・`time.sleep` → `def`（FastAPI がスレッドプールへ逃がす）。**迷ったら `def`**。`async def` の中で同期ブロッキングを呼ばない。
- **バリデーション**：入力／出力モデルを分け、`Field` 制約と `field_validator` で**境界で殺す**。外部 API の JSON も `model_validate` で即モデル化。信用できない入力は strict を検討。
- **DI**：DB セッション・認証・設定は `Depends` + `Annotated`。後始末は `yield` 依存で。テストで `dependency_overrides` できる設計にしておく。
- **起動コスト**：DB プール・モデル・HTTP クライアントは `lifespan` で一度だけ確保し `app.state` で共有。`@app.on_event` は使わない。
- **エラー**：4xx は即時返却、5xx は汎用メッセージ + 構造化ログ。スタックトレースをクライアントに返さない。
- **可観測性**：リクエストID付与 → 構造化（JSON）ログ → OpenTelemetry 計装。**PII はログに出さない**。
- **重い処理**：数十秒以上 / リトライ必須 / CPU・GPU 集約なら、`BackgroundTasks` を卒業して Celery / Cloud Run Jobs / Workflows へ。API は `202 Accepted` で薄く返す。
- **テスト**：同期は `TestClient`、非同期は `httpx.AsyncClient` + `ASGITransport`。依存は `dependency_overrides` で差し替える。
- **デプロイ**：`gunicorn` + `UvicornWorker` で複数ワーカー、`--graceful-timeout` と `lifespan` 終了処理でグレースフルに。ワーカー数 × プールサイズ ≤ DB 最大接続数。

---

FastAPI は「5 分で動く」フレームワークですが、本番品質は**境界の設計**で決まります。**外部入力を型で殺し、横断的関心事を依存に逃がし、重さを外の実行基盤へ追い出し、すべてを ID で追えるようにする**——どれも派手ではありませんが、これらの積み重ねが「落ちない・追える・変更しやすい API」を作ります。

私は放送事業者向けの社内 AI プラットフォームで、FastAPI を**長時間 AI ジョブの薄い受付層**として設計し、重い処理を Cloud Run Jobs / Cloud Workflows へ分離、可観測性と冪等性を担保した本番運用に乗せました。生成 AI（Claude Code）を相棒に、**一人で速く・安く**作りつつ、検証ゲートで品質を担保するのが私の進め方です。

**「FastAPI でこの API を本番に載せたいが、async の使い分けやジョブ分離・可観測性をどう設計すればいいか」——その判断から実装・運用まで、一気通貫で伴走します。** 要件整理の段階からでも、お気軽にご相談ください。

---

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

- [FastAPI 公式トップ](https://fastapi.tiangolo.com/) — 型ヒント・Pydantic・Starlette を土台にした設計思想と機能一覧
- [Concurrency and async / await（FastAPI）](https://fastapi.tiangolo.com/async/) — `async def` と `def` の使い分けルール、スレッドプール挙動
- [Dependencies（FastAPI）](https://fastapi.tiangolo.com/tutorial/dependencies/) — `Depends` / `Annotated` による依存性注入
- [Lifespan Events（FastAPI）](https://fastapi.tiangolo.com/advanced/events/) — `@asynccontextmanager` による起動・終了処理
- [Testing（FastAPI）](https://fastapi.tiangolo.com/tutorial/testing/) — `TestClient` と httpx AsyncClient によるテスト
- [Pydantic 公式ドキュメント](https://docs.pydantic.dev/latest/) — Pydantic v2 のモデル・バリデーション・`Field`・`field_validator`