4. FastAPI を ASGI 視点で見る

4.1. FastAPI と Starlette の関係

FastAPI は Starlette を継承したクラスです。ソースコード上、fastapi.applications.FastAPI は starlette.applications.Starlette のサブクラスであり、Starlette が提供する以下の ASGI 基盤機能をそのまま受け継いでいます。

ルーティング（Router）
ミドルウェアスタック
例外ハンドリング
テストクライアント
WebSocket 対応
lifespan 管理

FastAPI が独自に追加しているのは、主に次の3つの層です。

追加層	概要
Pydantic による型検証と変換	パスパラメータ・クエリパラメータ・リクエストボディを型ヒントから自動バリデーション・変換します
依存性注入（Dependency Injection）	`Depends()` を通じて DB セッション・認証情報・設定値などをビュー関数の引数として注入します。Starlette には存在しない FastAPI 固有の仕組みです
OpenAPI スキーマ自動生成	型ヒントと docstring から OpenAPI 3.x のスキーマを自動生成し、`/docs`（Swagger UI）と `/redoc` で閲覧できます

注釈

3 章（最小の ASGI HTTP アプリ）で手書きした int(user_id) や json.loads(body) のエラーハンドリングが、def get_user(user_id: int) や def create_user(user: UserCreate) の宣言だけで済むようになります。

この関係を整理すると、ASGI サーバ（Uvicorn）→ Starlette（ASGI フレームワーク基盤）→ FastAPI（型検証・DI・OpenAPI の追加層）という階層になります。 3 章（最小の ASGI HTTP アプリ）で書いた手製ルーターが Starlette の Router に相当し、FastAPI はその上に Pydantic と DI を載せた構造です。

4.2. ASGI アプリとしての FastAPI

FastAPI インスタンスは ASGI アプリそのものです。app = FastAPI() は async def __call__(self, scope, receive, send) を持ち、Uvicorn から直接呼び出されます。

重要

FastAPI は単なる「便利なフレームワーク」ではなく、ASGI インタフェースを完全に実装した ASGI アプリです。Uvicorn などのサーバとの境界は scope / receive / send の3引数で統一されています。

from fastapi import FastAPI

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id, "name": "Taro"}

このコードが動作するとき、内部では次の処理が順に行われます。

各ステップの概要は以下の通りです。

Uvicorn がリクエストを受け取る: TCP 接続を受け取り、HTTP リクエストをパースして scope 辞書を構築し、await app(scope, receive, send) を呼びます。2 章（なぜ ASGI が必要になったのか）で学んだ async def application(scope, receive, send) と同じインタフェースです。
Starlette のミドルウェアスタックを通過する: scope を受け取り、登録されたミドルウェアを外側から内側へ順に通過させます。3 章（最小の ASGI HTTP アプリ）で実装した LoggingMiddleware や ErrorHandlingMiddleware のラップ構造と同じパターンです。
Router がルートを特定する: scope["path"] と scope["method"] を走査してマッチするルートを特定します。Starlette は Mount（サブアプリ）や WebSocket ルートも統合的に扱います。
Pydantic による型変換と DI 解決: ルートが確定するとパスパラメータ {user_id} は Pydantic を通じて int に変換され、Depends() で宣言された依存関係が解決されます。変換やバリデーションに失敗した場合は自動的に 422 Unprocessable Entity が返されます。
ビュー関数の実行とレスポンス生成: ビュー関数が実行され、戻り値（この例では dict）は JSONResponse に変換されます。JSONResponse は Starlette のクラスで、内部的には await send({"type": "http.response.start", ...}) と await send({"type": "http.response.body", ...}) を呼び出します。3 章（最小の ASGI HTTP アプリ）の send_json ヘルパーと同等の処理です。

Tip

async def ではなく def で定義されたビュー関数は、FastAPI が自動的にスレッドプール（anyio.to_thread.run_sync）で実行します。2.11 章（トラブルシューティングの観点）で触れた「async 関数内で同期ブロッキングを呼ぶとイベントループが停止する」問題への対策であり、既存の同期ライブラリをそのまま使えるようにする実用的な設計です。

@app.get("/users/{user_id}")
def get_user_sync(user_id: int):  # def（同期）→ 自動的にスレッドプールで実行
    user = db.query(User).get(user_id)  # 同期 ORM でもブロックしない
    return {"user_id": user.id, "name": user.name}

FastAPI が本当に ASGI アプリであることは、以下のように Uvicorn から直接起動できることで確認できます。

uvicorn main:app --host 127.0.0.1 --port 8000

ここで main:app は「main.py モジュールの app オブジェクト」を指し、Uvicorn は app(scope, receive, send) を呼び出すだけです。 3 章（最小の ASGI HTTP アプリ）で uvicorn hello_asgi:application としたのと構造的に同じです。FastAPI であっても生の ASGI アプリであっても、サーバとアプリの境界は scope / receive / send の3引数で統一されています。

次節では、FastAPI のリクエスト受信からレスポンス返却までの内部フローをさらに詳細に追跡し、Pydantic による型変換と依存性注入がどのタイミングで実行されるかを見ていきます。

4.3. ルーティングの流れ

4.3.1. path operation

FastAPI では「ルート」を path operation と呼びます。これは HTTP メソッド（operation）と URL パス（path）の組み合わせでエンドポイントを定義する考え方で、OpenAPI 仕様の用語をそのまま採用しています。

注釈

path operation という名称は OpenAPI 仕様の用語です。FastAPI はドキュメント生成と実装の整合性を保つために、OpenAPI の語彙をそのまま採用しています。

from fastapi import FastAPI

app = FastAPI()

@app.get("/users")
async def list_users():
    return [{"id": 1, "name": "Taro"}]

@app.post("/users")
async def create_user(user: UserCreate):
    return {"id": 3, "name": user.name}

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"id": user_id, "name": "Taro"}

@app.get("/users/{user_id}") は内部的に app.router.add_api_route("/users/{user_id}", get_user, methods=["GET"]) を呼び出します。このとき Starlette の Route オブジェクトではなく、FastAPI 独自の APIRoute が生成されます。 APIRoute は Route を継承しつつ、次の情報を保持するサブクラスです。

Pydantic モデルの解析結果
依存関係グラフ
レスポンスモデル
OpenAPI メタデータ

3 章（最小の ASGI HTTP アプリ）で自作した Router では Route クラスがパスパターンを正規表現に変換し、メソッドとパスを保持していました。 FastAPI の APIRoute はこれに加えて「引数をどこから（パス・クエリ・ボディ・ヘッダー）抽出し、どの型に変換するか」という情報を起動時に解析・キャッシュしています。この事前解析が、リクエスト到着時の高速な型変換を支えています。

登録された APIRoute は app.router.routes リストに格納され、上から順にマッチングされます。 3 章（最小の ASGI HTTP アプリ）の自作ルーターと同様に、登録順序が優先度を決定します。

Tip

複数のルートで同じプレフィックスを持つ場合（例: /users と /users/{user_id}）、より具体的なパスを先に登録しておくと意図しないマッチを防げます。

4.3.2. method dispatch

リクエストが到着すると、Starlette の Router.__call__ が scope["path"] に対してルートリストを先頭から走査します。パスパターンが一致する APIRoute が見つかると、scope に path_params が追加され、その APIRoute のハンドラに処理が委譲されます。

重要

Starlette のルーティングでは パスの一致だけ が先に判定され、HTTP メソッドの判定は APIRoute 内部で行われます。具体的には、APIRoute が生成する request_response ラッパー内で request.method と登録済みメソッドを照合し、一致しなければ 405 Method Not Allowed を返します。この「パス優先・メソッド後判定」という二段階構造を把握しておくと、405 エラーの原因がすぐに追跡できます。

scope["path"] = "/users/42"
scope["method"] = "GET"

Router.routes を走査:
  Route("/users")          → パス不一致 → skip
  Route("/users")          → パス不一致 → skip（POST 用だが、パスが "/users/42" と不一致）
  Route("/users/{user_id}") → パス一致 → path_params = {"user_id": "42"}
    → メソッド確認: GET ∈ ["GET"] → ✓ → get_user を実行

同じパスに対して GET と DELETE を別々に登録した場合、ルートリストには2つの APIRoute が存在します。パスが一致する最初のルートでメソッドも一致すればそのまま実行され、メソッドが一致しなければ 405 が返ります。 3 章（最小の ASGI HTTP アプリ）の自作ルーターではメソッドとパスを同時にチェックしていましたが、Starlette / FastAPI はパス優先・メソッド後判定という二段階構造を取っています。

4.3.3. path / query / header / cookie の抽出

FastAPI の最も特徴的な部分が、ビュー関数のシグネチャから自動的にパラメータの取得元と型を決定する仕組みです。この処理は APIRoute 生成時に関数シグネチャを解析して「依存関係モデル」を構築し、リクエスト到着時にそのモデルに従って値を抽出・変換・検証します。

FastAPI の引数判別ルールは次のとおりです。

引数の宣言方法	取得元
パスに `{param}` がある同名引数	パスパラメータ
`Body()` や Pydantic モデル	リクエストボディ
`Header()`	HTTP ヘッダー
`Cookie()`	Cookie
それ以外のスカラー型引数	クエリパラメータ

URL パスのプレースホルダ {user_id} と同名の引数は自動的にパスパラメータとして扱われます。

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # scope["path_params"]["user_id"] = "42" → int(42) に変換
    return {"user_id": user_id}

Starlette のルーターがパスマッチ時に scope["path_params"] へ文字列として格納した値を、FastAPI が Pydantic を通じて int に変換します。 3 章（最小の ASGI HTTP アプリ）では try: uid = int(user_id) except ValueError: ... と手動で行っていた処理が、型ヒント user_id: int の宣言だけで完了します。変換に失敗した場合（例: /users/abc）は自動的に 422 Unprocessable Entity が返されます。

パスに含まれず、Pydantic モデルでもない引数は自動的にクエリパラメータとして解釈されます。

Tip

デフォルト値のない引数は必須クエリパラメータとして扱われます。省略可能にするには skip: int = 0 のようにデフォルト値を設定するか、skip: int | None = None と宣言してください。

@app.get("/users")
async def list_users(skip: int = 0, limit: int = 10, active: bool = True):
    # GET /users?skip=20&limit=5&active=false
    # scope["query_string"] = b"skip=20&limit=5&active=false"
    # → skip=20, limit=5, active=False に変換
    return {"skip": skip, "limit": limit, "active": active}

scope["query_string"] のバイト列をパースし、各引数名に対応する値を型ヒントに従って変換します。デフォルト値があれば省略可能、なければ必須パラメータとして扱われ、欠落時は 422 が返ります。 2 章（なぜ ASGI が必要になったのか）で parse_qs(scope["query_string"].decode()) と手動で書いた処理が、シグネチャ宣言に吸収されています。

Header() を明示的に指定することで、HTTP ヘッダーから値を抽出します。

注釈

FastAPI は Python の変数名（アンダースコア区切り）を HTTP ヘッダー名（ハイフン区切り）に自動変換します。たとえば user_agent → user-agent、x_request_id → x-request-id となります。

from fastapi import Header

@app.get("/items")
async def read_items(user_agent: str = Header(default=None), x_request_id: str = Header(default=None)):
    # scope["headers"] から "user-agent" と "x-request-id" を抽出
    return {"user_agent": user_agent, "request_id": x_request_id}

2 章（なぜ ASGI が必要になったのか）の scope["headers"] はバイト列タプルのリストでしたが、FastAPI がその検索と変換を代行します。

Cookie() で宣言すると、リクエストの Cookie ヘッダーから値を取得します。

from fastapi import Cookie

@app.get("/me")
async def read_me(session_id: str = Cookie(default=None)):
    # Cookie: session_id=abc123 → session_id = "abc123"
    return {"session_id": session_id}

この判別ロジックは APIRoute 生成時に dependant オブジェクトとして解析・保存され、リクエスト到着時には解析済みの情報に従って値を取り出すだけです。

リクエスト到着
  │
  ├─ scope["path_params"]  ─→ パスパラメータ引数へ
  ├─ scope["query_string"]  ─→ クエリパラメータ引数へ
  ├─ scope["headers"]       ─→ Header() 引数へ
  ├─ scope["headers"] (cookie) ─→ Cookie() 引数へ
  └─ await receive() (body)  ─→ Body() / Pydantic モデル引数へ
        │
        ▼
   Pydantic で型変換・バリデーション
        │
    成功 → ビュー関数実行
    失敗 → 422 Unprocessable Entity（詳細なエラーメッセージ付き）

3 章（最小の ASGI HTTP アプリ）で手書きした「パスパラメータの正規表現抽出 → 型変換 → エラーハンドリング → ボディ読み取り → JSON パース → バリデーション」という一連の処理が、FastAPI では関数シグネチャの型ヒントに凝縮されています。しかしその裏側では、本章で追ってきた scope、receive、Starlette のルーティング、正規表現マッチングが同じように動いています。

次節では、この型変換の中核を担う Pydantic の検証フローと、リクエストボディの処理を詳しく追っていきます。

4.4. リクエストボディの解析

4.4.1. JSON body

FastAPI でリクエストボディを受け取る最も一般的な方法は、Pydantic モデルを引数の型ヒントとして宣言することです。

注釈

リクエストボディの受け取り方は、Content-Type によって3種類に分かれます。

application/json → Pydantic モデルまたは Body()
application/x-www-form-urlencoded → Form()
multipart/form-data → File() / UploadFile

from pydantic import BaseModel, EmailStr
from fastapi import FastAPI

app = FastAPI()

class UserCreate(BaseModel):
    name: str
    email: EmailStr
    age: int | None = None

@app.post("/users")
async def create_user(user: UserCreate):
    return {"id": 1, "name": user.name, "email": user.email}

この宣言だけで、FastAPI は次の処理を自動的に実行します。

内部で行われる処理は次の4段階です。

Starlette の Request オブジェクトが scope["headers"] から content-type を取得し、JSON として解釈可能かを判断します。
await receive() を more_body フラグが False になるまでループして全ボディを蓄積します（Starlette の Request.body() メソッドが内部でこのループを実行し、結果をキャッシュします）。
蓄積されたバイト列は json.loads() でデコードされます。
得られた辞書が Pydantic モデルの model_validate() に渡され、型変換・バリデーションが実行されます。

3 章（最小の ASGI HTTP アプリ）で書いたコードと比較すると、構造が同一であることが分かります。

# 第8章の手書き実装
body = await read_body(receive)               # receive ループ
content_type = get_header(scope, b"content-type")
if content_type != b"application/json":
    await send_error(send, 415, "Unsupported Media Type")
    return
try:
    data = json.loads(body)
except json.JSONDecodeError:
    await send_error(send, 400, "Invalid JSON")
    return
if "name" not in data or not isinstance(data["name"], str):
    await send_error(send, 422, "name is required")
    return

# FastAPI では上記すべてが型ヒント一行に集約される
async def create_user(user: UserCreate): ...

複数の JSON フィールドを個別に受け取ることも可能です。 Body() を使うと、単一のスカラー値やネストされた構造を明示的に指定できます。

Body(gt=0) のように Pydantic のバリデーション制約をインラインで指定できます。

from fastapi import Body

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    name: str = Body(),
    price: float = Body(gt=0),
    description: str | None = Body(default=None),
):
    return {"item_id": item_id, "name": name, "price": price}

この場合、FastAPI は {"name": "...", "price": ..., "description": "..."} という JSON ボディを期待し、各フィールドを個別に抽出・検証します。

4.4.2. フォーム

HTML フォームからの application/x-www-form-urlencoded 送信を受け取るには Form() を使います。

警告

Form() と Body() を同一エンドポイントに混在させることはできません。リクエストボディは一つの Content-Type しか持てないため、JSON かフォームかのいずれか一方を選ぶ必要があります。これは HTTP プロトコルの制約であり、FastAPI のルールではありません。

from fastapi import Form

@app.post("/login")
async def login(username: str = Form(), password: str = Form()):
    return {"username": username}

内部的には、Starlette の Request.form() が呼ばれます。このメソッドは Content-Type ヘッダーを確認し、application/x-www-form-urlencoded であれば await receive() でボディを読み取った後、urllib.parse.parse_qsl 相当の処理でキーと値のペアに分解します。 Vol.1「まずは 1 リクエストだけ処理するサーバを作る」で見た HTTP ボディのパースや、Werkzeug の url_decode と同じ仕組みが Starlette の内部に組み込まれています。

curl -X POST http://127.0.0.1:8000/login \
  -d "username=taro&password=secret123"
# → {"username": "taro"}

4.4.3. ファイルアップロード

ファイルを受け取るには File() または UploadFile を使います。

from fastapi import File, UploadFile

@app.post("/upload")
async def upload_file(file: UploadFile):
    contents = await file.read()
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": len(contents),
    }

@app.post("/upload-multiple")
async def upload_multiple(files: list[UploadFile]):
    return [{"filename": f.filename, "size": f.size} for f in files]

UploadFile は Starlette が提供するクラスで、内部的には multipart/form-data のパースを python-multipart ライブラリに委譲します。小さなファイルはメモリ上の SpooledTemporaryFile に保持され、閾値（デフォルト 1 MB）を超えるとディスクに書き出されます。これは Django の FILE_UPLOAD_MAX_MEMORY_SIZE による InMemoryUploadedFile と TemporaryUploadedFile の切り替え（1.3 章（リクエストは Django にどう渡るか））と同じ設計判断です。

UploadFile は非同期メソッド read(), seek(), write(), close() を持ち、await file.read(1024) のようにチャンク単位で読み取ることもできます。

注意

大きなファイルを一括で await file.read() するとメモリを圧迫します。ストリーム処理（チャンク読み取り）を推奨します。

@app.post("/upload-large")
async def upload_large(file: UploadFile):
    total_size = 0
    while chunk := await file.read(8192):
        total_size += len(chunk)
        # ここでチャンクをストレージへ書き出す
    return {"filename": file.filename, "total_size": total_size}

ファイルとフォームフィールドは同時に受け取ることができます。

@app.post("/profile")
async def update_profile(
    username: str = Form(),
    bio: str = Form(default=""),
    avatar: UploadFile | None = File(default=None),
):
    result = {"username": username, "bio": bio}
    if avatar:
        result["avatar_filename"] = avatar.filename
    return result

この場合、Content-Type は multipart/form-data となり、テキストフィールドとファイルが同一リクエスト内で送信されます。

4.4.4. バリデーションの入り口

JSON body・Form・File のすべてにおいて、値の抽出後に Pydantic によるバリデーションが実行されます。このバリデーションが FastAPI のリクエスト処理パイプラインのどこに位置するかを整理します。

ASGI サーバ (Uvicorn)
  │  scope, receive, send
  ▼
Starlette ミドルウェアスタック
  │
  ▼
Starlette Router → パスマッチング → scope["path_params"] 設定
  │
  ▼
APIRoute.handle() → request_response ラッパー
  │
  ├─ scope["path_params"] からパスパラメータ抽出
  ├─ scope["query_string"] からクエリパラメータ抽出
  ├─ scope["headers"] からヘッダー・Cookie 抽出
  ├─ await receive() → ボディ読み取り
  │    ├─ application/json → json.loads → dict
  │    ├─ application/x-www-form-urlencoded → parse → dict
  │    └─ multipart/form-data → python-multipart → dict + UploadFile
  │
  ▼
Pydantic model_validate() / フィールド単位の型変換
  │
  ├─ 成功 → 変換済みの値をビュー関数の引数として渡す → 実行
  └─ 失敗 → RequestValidationError → 422 レスポンス自動生成

バリデーションエラーが発生すると、FastAPI は RequestValidationError 例外を送出し、デフォルトの例外ハンドラが 422 ステータスコードと詳細なエラー情報を JSON 形式で返します。

curl -X POST http://127.0.0.1:8000/users \
  -H "Content-Type: application/json" \
  -d '{"name": "Taro", "email": "not-an-email"}'

# → 422
# {
#   "detail": [
#     {
#       "type": "value_error",
#       "loc": ["body", "email"],
#       "msg": "value is not a valid email address...",
#       "input": "not-an-email"
#     }
#   ]
# }

loc フィールドがエラーの位置を示します。各先頭要素の意味は次の通りです。

`loc` の先頭	意味
`"body"`	リクエストボディのフィールド
`"path"`	パスパラメータ
`"query"`	クエリパラメータ
`"header"`	HTTP ヘッダー
`"cookie"`	Cookie

この位置情報は、FastAPI がパラメータの取得元を APIRoute 生成時に解析して記録しているからこそ提供できるものです。

3 章（最小の ASGI HTTP アプリ）では手動で if "name" not in data のようなチェックを書き、send_error(send, 422, "...") で返していました。 FastAPI ではこの検証ロジックが Pydantic モデルの定義に集約され、エラーレスポンスの形式も OpenAPI 仕様に準拠した統一フォーマットで自動生成されます。ただし、422 が返ったときにその原因を追跡するには、リクエストボディがどのように receive → パース → Pydantic 検証というパイプラインを流れるかを理解していることが助けになります。

次節では、このバリデーションの中核を担う Pydantic の型変換と検証の仕組みをさらに掘り下げていきます。

4.5. Pydantic によるデータ検証

4.5.1. モデル定義

Pydantic モデルは Python のクラス構文と型ヒントだけでデータの構造・型・制約を宣言します。 FastAPI はこのモデルをリクエストボディの検証とレスポンスのシリアライズの両方に使います。

Tip

リクエスト用モデル（UserCreate）とレスポンス用モデル（UserResponse）を分けて定義するのがベストプラクティスです。クライアントが送るフィールドとサーバが返すフィールドを明確に区別することで、意図しないデータの漏洩を防げます。

from pydantic import BaseModel, Field, EmailStr
from datetime import datetime

class UserCreate(BaseModel):
    name: str = Field(min_length=1, max_length=100, examples=["Taro Yamada"])
    email: EmailStr
    age: int | None = Field(default=None, ge=0, le=150)
    tags: list[str] = Field(default_factory=list, max_length=10)

class UserResponse(BaseModel):
    id: int
    name: str
    email: EmailStr
    age: int | None = None
    tags: list[str] = []
    created_at: datetime

    model_config = {"from_attributes": True}

UserCreate はリクエスト受信時の検証に、UserResponse はレスポンス送信時のシリアライズに使います。この分離により、クライアントが送信するフィールド（name, email, age, tags）とサーバが返すフィールド（id, created_at を含む）を明確に区別できます。

Field() は各フィールドにバリデーション制約を付与します。よく使う引数は次の通りです。

引数	対象	説明
`min_length` / `max_length`	文字列	文字数の下限・上限
`ge` / `le` / `gt` / `lt`	数値	以上・以下・超過・未満
`pattern`	文字列	正規表現マッチ
`default_factory`	any	ミュータブルなデフォルト値の生成

これらの制約は OpenAPI スキーマにもそのまま反映され、/docs の Swagger UI 上でフィールドの許容範囲が確認できます。

Pydantic v2 では model_config でモデル全体の振る舞いを制御します。 from_attributes = True を設定すると、ORM オブジェクト（SQLAlchemy のモデルインスタンスなど）の属性から直接値を読み取れるようになります。

# ORM オブジェクトからレスポンスモデルへの変換
user_orm = db.query(User).get(1)
# user_orm.id, user_orm.name, ... の属性を読み取る
response = UserResponse.model_validate(user_orm)

ネストされたモデルも自然に定義することができます。

class Address(BaseModel):
    country: str
    city: str
    street: str | None = None

class UserWithAddress(BaseModel):
    name: str
    email: EmailStr
    address: Address

リクエストボディとして {"name": "Taro", "email": "taro@example.com", "address": {"country": "JP", "city": "Tokyo"}} を受け取ると、address フィールドも再帰的に検証されます。 3 章（最小の ASGI HTTP アプリ）で手書きした場合、ネストされた辞書のキー存在確認・型チェック・デフォルト値処理をすべて手動で行う必要がありましたが、Pydantic はモデル定義の構造をそのまま検証ロジックに変換します。

4.5.2. parse / validate

Pydantic v2 の検証は次の2つのメソッドで実行します。

model_validate() — 辞書入力
model_validate_json() — JSON 文字列入力

FastAPI の内部では、リクエストボディを json.loads() で辞書に変換した後、model_validate() を呼び出す流れが基本です。

# 辞書からの検証
data = {"name": "Taro", "email": "[email protected]", "age": 25}
user = UserCreate.model_validate(data)
# → UserCreate(name='Taro', email='[email protected]', age=25, tags=[])

# JSON 文字列からの検証（Rust 実装の高速パーサーを使用）
json_str = '{"name": "Taro", "email": "[email protected]", "age": 25}'
user = UserCreate.model_validate_json(json_str)

model_validate_json() は Pydantic v2 の Rust コア（pydantic-core）に実装された JSON パーサーを直接使用し、json.loads() + model_validate() よりも高速です。 FastAPI の将来のバージョンではこの最適化パスを活用する可能性がありますが、現時点では json.loads() を経由する実装が中心です。

型変換の過程で Pydantic は「厳密モード」と「寛容モード」の2種類の振る舞いを持ちます。デフォルトは寛容モード（lax mode）で、妥当な型変換を自動的に行います。

from pydantic import BaseModel

class Item(BaseModel):
    price: float
    quantity: int
    active: bool

# 寛容モード（デフォルト）
item = Item.model_validate({"price": "19.99", "quantity": "3", "active": "true"})
# → Item(price=19.99, quantity=3, active=True)
# 文字列 → float/int/bool に自動変換

# 厳密モードではこの変換を拒否
item = Item.model_validate(
    {"price": "19.99", "quantity": "3", "active": "true"},
    strict=True
)
# → ValidationError: price は float でなければならない

FastAPI はデフォルトで寛容モードを使用します。パスパラメータやクエリパラメータは HTTP 仕様上すべて文字列として到着するため、str → int や str → float の自動変換は実用上不可欠です。

注釈

JSON ボディについては型が明確であるため、意図しない型変換が起きていないか注意してください。厳密な型チェックが必要な場合は model_validate(..., strict=True) を使います。

バリデーション失敗時は ValidationError が送出されます。

from pydantic import ValidationError

try:
    user = UserCreate.model_validate({
        "name": "",
        "email": "not-an-email",
        "age": -5,
        "tags": ["a"] * 20,
    })
except ValidationError as e:
    print(e.error_count())  # 4
    for err in e.errors():
        print(err)

各エラーには次の情報が含まれます。

type — エラー種別
loc — フィールド位置のタプル
msg — 人間向けメッセージ
input — 入力値
ctx — 制約の詳細

上記の例では、name の min_length 違反・email のフォーマットエラー・age の ge=0 違反・tags の max_length 超過がそれぞれ個別のエラーとして報告されます。 Pydantic はすべてのフィールドを検証してからまとめてエラーを返すため、クライアントは一度のリクエストで全修正箇所を把握できます。

4.5.3. エラー応答生成

Pydantic の ValidationError が FastAPI のリクエスト処理パイプラインで発生すると、FastAPI はこれを RequestValidationError として捕捉し、デフォルトの例外ハンドラが 422 Unprocessable Entity レスポンスを生成します。

Tip

デフォルトのエラーレスポンス形式は @app.exception_handler(RequestValidationError) でカスタマイズできます。チームのエラーレスポンス規約がある場合はここで統一しましょう。

curl -X POST http://127.0.0.1:8000/users \
  -H "Content-Type: application/json" \
  -d '{"name": "", "email": "bad", "age": -5}'

# HTTP/1.1 422 Unprocessable Entity
# {
#   "detail": [
#     {
#       "type": "string_too_short",
#       "loc": ["body", "name"],
#       "msg": "String should have at least 1 character",
#       "input": "",
#       "ctx": {"min_length": 1}
#     },
#     {
#       "type": "value_error",
#       "loc": ["body", "email"],
#       "msg": "value is not a valid email address...",
#       "input": "bad"
#     },
#     {
#       "type": "greater_than_equal",
#       "loc": ["body", "age"],
#       "msg": "Input should be greater than or equal to 0",
#       "input": -5,
#       "ctx": {"ge": 0}
#     }
#   ]
# }

loc の先頭要素が "body" であることに注目してください。これは 4.3 章（ルーティングの流れ）で述べた通り、FastAPI がパラメータの取得元をルート定義時に解析し記録しているためです。パスパラメータのエラーなら ["path", "user_id"]、クエリパラメータなら ["query", "limit"] となります。ネストされたモデルの場合は ["body", "address", "city"] のように位置が深くなります。

デフォルトのエラーレスポンス形式は次のようにカスタマイズできます。

from fastapi import FastAPI, Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(RequestValidationError)
async def custom_validation_handler(request: Request, exc: RequestValidationError):
    errors = []
    for err in exc.errors():
        field = ".".join(str(loc) for loc in err["loc"])
        errors.append({
            "field": field,
            "message": err["msg"],
            "type": err["type"],
        })
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={
            "error": "Validation Failed",
            "error_count": len(errors),
            "details": errors,
        },
    )

この例外ハンドラは Starlette の例外ハンドリング機構を利用しており、@app.exception_handler で登録された関数は、ミドルウェアスタック内で例外が送出されたときに呼び出されます。 1 章（Django を WSGI 視点で見る）で Django の process_exception フックを見ましたが、FastAPI / Starlette では例外の型ごとにハンドラを登録する方式を採用しています。

レスポンス側のシリアライズにも Pydantic が関与します。response_model を指定すると、ビュー関数の戻り値がそのモデルでフィルタリング・変換されます。

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    new_user = {
        "id": 1,
        "name": user.name,
        "email": user.email,
        "age": user.age,
        "tags": user.tags,
        "created_at": datetime.now(),
        "internal_secret": "should_not_be_exposed",  # これは除外される
    }
    return new_user

response_model=UserResponse により、internal_secret のような UserResponse に定義されていないフィールドは自動的に除外されます。意図しないデータ漏洩を防ぐフィルタリング層として機能し、レスポンスの構造も OpenAPI スキーマに反映されます。

バリデーションの全体的な流れを3 章（最小の ASGI HTTP アプリ）との対比で整理すると、次のようになります。

第8章（手書き）                        FastAPI + Pydantic
──────────────────────────          ──────────────────────────
body = await read_body(receive)    → Starlette Request.body()
data = json.loads(body)            → json.loads (内部)
if "name" not in data: ...         → model_validate() が自動判定
if not isinstance(data["age"], int)  → 型ヒント int で自動変換・検証
if data["age"] < 0: ...            → Field(ge=0) が自動判定
await send_error(send, 422, msg)   → RequestValidationError → 422 自動生成
response_data = {必要なキーだけ}      → response_model でフィルタリング
await send_json(send, response_data) → JSONResponse が自動生成

3 章（最小の ASGI HTTP アプリ）のコードで20〜30行を費やしていた検証・エラーハンドリングが、モデル定義と型ヒントに集約されています。しかし内部では依然として receive によるボディ読み取り、JSON パース、フィールドごとの型チェックという同じ段階を踏んでいます。 422 エラーの loc を手がかりにどのパラメータがどの段階で失敗したかを特定できるのは、この内部構造を把握しているからこそです。

次節では、FastAPI のもう一つの中核機能である依存性注入（Dependency Injection）の仕組みを追います。

4.6. Dependency Injection の仕組み

4.6.1. Depends

FastAPI の依存性注入（DI）は、ビュー関数の引数に Depends() を宣言するだけで、必要なリソースや前処理の結果を自動的に注入する仕組みです。

注釈

Depends() には callable（関数・クラス・ジェネレータ関数）を渡すことができます。 FastAPI はリクエスト到着時にこの callable を呼び出し、戻り値をビュー関数の引数として注入します。

from fastapi import FastAPI, Depends, Query

app = FastAPI()

async def common_pagination(skip: int = Query(0, ge=0), limit: int = Query(10, ge=1, le=100)):
    return {"skip": skip, "limit": limit}

@app.get("/users")
async def list_users(pagination: dict = Depends(common_pagination)):
    return {"users": [...], "pagination": pagination}

@app.get("/articles")
async def list_articles(pagination: dict = Depends(common_pagination)):
    return {"articles": [...], "pagination": pagination}

上記の例では common_pagination が skip と limit のクエリパラメータを抽出・検証し、その結果が pagination 引数に渡されます。同じロジックを複数のエンドポイントで再利用でき、各エンドポイントでパラメータの検証コードを重複させる必要がありません。

Depends() の内部動作は、4.3 章（ルーティングの流れ）で述べた APIRoute 生成時のシグネチャ解析と直結しています。 FastAPI は @app.get("/users") が呼ばれた時点で list_users のシグネチャを inspect し、Depends() が指定された引数を発見すると、その callable のシグネチャもさらに再帰的に解析します。 common_pagination は skip: int = Query(...) と limit: int = Query(...) を持つので、FastAPI はこれらをクエリパラメータとして登録します。この解析結果は dependant オブジェクトとしてキャッシュされ、リクエスト到着時には解析済みの情報に従って値を取り出すだけです。

依存関数自体もパスパラメータ・ヘッダー・Cookie・リクエストボディなどを受け取ることができます。

from fastapi import Header, HTTPException

async def verify_token(authorization: str = Header()):
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid authorization header")
    token = authorization[7:]
    user = await decode_and_verify(token)
    if user is None:
        raise HTTPException(status_code=401, detail="Invalid or expired token")
    return user

@app.get("/me")
async def get_current_user(user: User = Depends(verify_token)):
    return {"id": user.id, "name": user.name}

verify_token が例外を送出した場合、ビュー関数は実行されません。 FastAPI が HTTPException を捕捉し、対応するステータスコードのレスポンスを自動生成します。この仕組みにより、認証・認可のロジックをビュー関数から完全に分離できます。

4.6.2. 依存関係グラフ

依存関数はさらに別の依存関数に依存することができます。 FastAPI はこの依存関係を有向非巡回グラフ（DAG）として解決します。

async def get_db_session():
    async with async_session_maker() as session:
        yield session

async def get_current_user(
    authorization: str = Header(),
    db: AsyncSession = Depends(get_db_session),
):
    token = authorization.replace("Bearer ", "")
    user = await db.execute(select(User).where(User.token == token))
    user = user.scalar_one_or_none()
    if user is None:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

async def require_admin(current_user: User = Depends(get_current_user)):
    if current_user.role != "admin":
        raise HTTPException(status_code=403, detail="Admin access required")
    return current_user

@app.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    admin: User = Depends(require_admin),
    db: AsyncSession = Depends(get_db_session),
):
    await db.execute(delete(User).where(User.id == user_id))
    await db.commit()
    return {"deleted": user_id}

このエンドポイントの依存関係グラフは次のようになります。

delete_user
  ├── user_id (パスパラメータ)
  ├── admin = Depends(require_admin)
  │     └── current_user = Depends(get_current_user)
  │           ├── authorization (ヘッダー)
  │           └── db = Depends(get_db_session)   ← ★
  └── db = Depends(get_db_session)               ← ★

ここで重要なのは、get_db_session が2箇所から参照されている点です。 FastAPI はデフォルトで同一リクエスト内の同じ依存関数を 1回だけ実行し、結果を再利用 します。 get_current_user に注入された db と delete_user に直接注入された db は同一のセッションインスタンスです。これにより、認証時のクエリとビジネスロジックのクエリが同一トランザクション内で実行されることが保証されます。

注釈

この再利用を無効にしたい場合は Depends(get_db_session, use_cache=False) と指定します。独立したセッションが必要なケース（例えば監査ログを別トランザクションで記録する場合）で使います。

解決順序は DAG のトポロジカルソートに従います。葉ノード（他に依存がないもの）から先に解決し、すべての依存が解決された後にビュー関数が呼ばれます。

警告

循環依存は APIRoute 生成時（サーバ起動時）に無限再帰を引き起こします。Python の型ヒント上で静的にチェックされるわけではないため、実行時エラーとなります。依存グラフが深くなる場合は構造を意識して設計してください。

4.6.3. request-scoped な値

FastAPI の DI はリクエストスコープで動作します。各リクエストが到着するたびに依存関数が新たに呼び出され、リクエスト処理が完了すると値は破棄されます。

この動作は Django の request オブジェクトがリクエストごとに生成される設計と同じ思想です。

import uuid

async def get_request_id():
    return str(uuid.uuid4())

@app.get("/items")
async def list_items(request_id: str = Depends(get_request_id)):
    # request_id はリクエストごとに新しい UUID
    return {"request_id": request_id, "items": [...]}

1 章（Django を WSGI 視点で見る）で見た WSGIRequest.__init__ が毎回 environ から新しいインスタンスを作るように、FastAPI の DI も毎回依存関数を実行します。

Request オブジェクト自体も依存として注入できます。

from fastapi import Request

@app.get("/debug")
async def debug_info(request: Request, request_id: str = Depends(get_request_id)):
    return {
        "request_id": request_id,
        "method": request.method,
        "url": str(request.url),
        "client": request.client.host if request.client else None,
        "headers": dict(request.headers),
    }

Starlette の Request は scope と receive をラップしたオブジェクトであり、2 章（なぜ ASGI が必要になったのか）で学んだ scope["method"]、scope["path"]、scope["headers"] のバイト列タプルを使いやすいプロパティとして提供します。 FastAPI は引数の型が Request であることを検出すると、Starlette の Request(scope, receive) を自動的に注入します。

4.6.4. cleanup を伴う dependency

データベースセッション・ファイルハンドル・外部 API クライアントなど、使用後にクリーンアップが必要なリソースには yield を使ったジェネレータ関数を用います。

async def get_db_session():
    session = async_session_maker()
    try:
        yield session
        await session.commit()
    except Exception:
        await session.rollback()
        raise
    finally:
        await session.close()

yield を使った依存関数の実行タイミングは次の通りです。

yield より前 → セットアップ（リクエスト処理前に実行）
yield された値 → ビュー関数に注入
yield より後 → クリーンアップ（レスポンス送信後に実行）

例外が発生した場合も finally ブロックが確実に実行されます。

この仕組みは Python の contextmanager と同じパターンです。 FastAPI は内部的にジェネレータを contextmanager としてラップし、__aenter__ でセットアップ、__aexit__ でクリーンアップを行います。 3 章（最小の ASGI HTTP アプリ）の lifespan（3.8 章（lifespan を扱う））がプロセスレベルの初期化・終了を担うのに対し、yield 依存はリクエストレベルのリソース管理を担います。

複数の yield 依存を組み合わせた場合のクリーンアップ順序は、依存関係グラフの逆順となります。

async def get_cache_client():
    client = await create_cache_client()
    try:
        yield client
    finally:
        await client.close()

async def get_db_session():
    session = async_session_maker()
    try:
        yield session
    finally:
        await session.close()

@app.get("/dashboard")
async def dashboard(
    db: AsyncSession = Depends(get_db_session),
    cache: CacheClient = Depends(get_cache_client),
):
    # db と cache を使ったビジネスロジック
    return {"data": ...}

この場合の実行順序は次の通りです。

get_db_session セットアップ
get_cache_client セットアップ
dashboard 実行
get_cache_client クリーンアップ
get_db_session クリーンアップ

リソースの取得順（グラフのトポロジカル順）と逆順で解放されるため、依存先が依存元より先に閉じられることはありません。

DI の全体フローを 3 章（最小の ASGI HTTP アプリ）のコードと対比すると、次のようになります。

第8章（手書き）                            FastAPI DI
──────────────────────────              ──────────────────────────
# ビュー関数の冒頭で毎回書く              # 宣言するだけ
token = get_header(scope, b"authorization")  → authorization: str = Header()
user = await verify(token)                   → current_user = Depends(get_current_user)
if user is None:                             → HTTPException(401) を依存関数内で送出
    await send_error(send, 401, "...")
    return
db = await create_session()                  → db = Depends(get_db_session)
try:                                         → yield で自動管理
    result = await db.execute(...)
    await db.commit()
except:
    await db.rollback()
    raise
finally:
    await db.close()

手書きでは認証チェック・セッション管理・エラーハンドリング・クリーンアップをビュー関数ごとに記述する必要がありましたが、FastAPI の DI ではこれらが依存関数として分離・再利用され、ビュー関数はビジネスロジックに集中できます。

Tip

DI が期待通りに動かないとき（依存関数が呼ばれない・クリーンアップが実行されない・同一セッションが共有されないなど）は、内部では scope["headers"] からの値抽出・Pydantic による型変換・ジェネレータの __aenter__/__aexit__ 呼び出しが行われていることを思い出してください。この内部構造の理解がデバッグの出発点になります。

次節では、FastAPI のレスポンス生成とミドルウェアの仕組みを追い、リクエスト受信からレスポンス送信までの全フローを完成させます。

4.7. レスポンス生成

4.7.1. dict から JSONResponse へ

FastAPI のビュー関数が dict や list を返すと、フレームワークが自動的に JSONResponse へ変換します。この変換は「魔法」ではなく、明確な処理ステップを持っています。

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"id": user_id, "name": "Taro", "email": "[email protected]"}

ビュー関数が値を返すと、FastAPI の request_response ラッパー内で戻り値の型が検査されます。戻り値がすでに Response オブジェクト（JSONResponse, HTMLResponse など）であればそのまま使われます。そうでなければ（dict, list, Pydantic モデルなど）、response_model が指定されていればそのモデルでフィルタリング・検証した後、jsonable_encoder() で JSON シリアライズ可能な形に変換し、JSONResponse でラップします。

JSONResponse は Starlette が提供するクラスで、その __call__ メソッドは3 章（最小の ASGI HTTP アプリ）で手書きした send_json ヘルパーと本質的に同じ処理を行います。

# Starlette の JSONResponse.__call__ が内部で行う処理（概念コード）
body = json.dumps(content, ensure_ascii=False).encode("utf-8")
await send({
    "type": "http.response.start",
    "status": self.status_code,
    "headers": [
        [b"content-type", b"application/json"],
        [b"content-length", str(len(body)).encode()],
        # その他のヘッダー
    ],
})
await send({
    "type": "http.response.body",
    "body": body,
})

3 章（最小の ASGI HTTP アプリ）で await send({"type": "http.response.start", ...}) と await send({"type": "http.response.body", ...}) を明示的に書いていた処理が、JSONResponse クラスに封じ込められています。ビュー関数が return {"id": 1} と書くだけで済むのは、FastAPI がこの変換パイプラインを自動実行するからです。

4.7.2. response_model

response_model を指定すると、ビュー関数の戻り値がそのモデルを通じてフィルタリング・検証されます。

from pydantic import BaseModel, EmailStr
from datetime import datetime

class UserResponse(BaseModel):
    id: int
    name: str
    email: EmailStr
    created_at: datetime

    model_config = {"from_attributes": True}

class UserInternal:
    """ORM や内部ロジックで使う構造（password_hash などを含む）"""
    def __init__(self, id, name, email, password_hash, created_at, internal_note):
        self.id = id
        self.name = name
        self.email = email
        self.password_hash = password_hash
        self.created_at = created_at
        self.internal_note = internal_note

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    user = UserInternal(
        id=user_id, name="Taro", email="[email protected]",
        password_hash="$2b$12$...", created_at=datetime.now(),
        internal_note="VIP customer",
    )
    return user

この場合のレスポンスには id, name, email, created_at だけが含まれ、password_hash と internal_note は自動的に除外されます。 response_model は出力のスキーマを強制するフィルタリング層であり、内部データが意図せずクライアントに露出することを防ぎます。

response_model が設定されている場合の変換フローは次の通りです。

ビュー関数の戻り値を model_validate() で検証
model_dump() で辞書に変換
jsonable_encoder() で JSON シリアライズ可能な型に変換
JSONResponse に渡してレスポンスを生成

Tip

response_model_exclude や response_model_include でフィールドを動的に制御することもできますが、実務では用途に応じた専用モデルを定義する方が、OpenAPI スキーマとコードの対応が明確になり保守性が高くなります。

class UserBrief(BaseModel):
    id: int
    name: str

class UserDetail(BaseModel):
    id: int
    name: str
    email: EmailStr
    created_at: datetime

@app.get("/users", response_model=list[UserBrief])
async def list_users():
    return [{"id": 1, "name": "Taro"}, {"id": 2, "name": "Hanako"}]

@app.get("/users/{user_id}", response_model=UserDetail)
async def get_user(user_id: int):
    return {"id": user_id, "name": "Taro", "email": "[email protected]", "created_at": datetime.now()}

一覧エンドポイントでは簡潔な UserBrief を、詳細エンドポイントでは UserDetail を返すことで、不要なデータの転送とシリアライズコストを削減できます。

4.7.3. シリアライズ

jsonable_encoder() は FastAPI が提供するユーティリティで、Python 標準の json.dumps() では直接シリアライズできない型を JSON 互換の型に変換します。

変換される主な型は次の通りです。

Python 型	JSON 変換後
`datetime`	ISO 8601 文字列（例: `"2025-01-15T10:30:00"`）
`date`	日付文字列（例: `"2025-01-15"`）
`Decimal`	浮動小数点数（例: `19.99`）
`UUID`	文字列（例: `"12345678-..."`）
`Enum`	値（例: `"active"`）
Pydantic モデル	`model_dump()` で辞書化後に変換

from fastapi.encoders import jsonable_encoder
from datetime import datetime, date
from decimal import Decimal
from uuid import UUID
from enum import Enum
from pydantic import BaseModel

class Status(Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"

class Item(BaseModel):
    name: str
    price: Decimal

data = {
    "created_at": datetime(2025, 1, 15, 10, 30, 0),   # → "2025-01-15T10:30:00"
    "date": date(2025, 1, 15),                          # → "2025-01-15"
    "price": Decimal("19.99"),                           # → 19.99
    "id": UUID("12345678-1234-5678-1234-567812345678"),  # → "12345678-..."
    "status": Status.ACTIVE,                             # → "active"
    "item": Item(name="Widget", price=Decimal("9.99")),  # → {"name": "Widget", "price": 9.99}
}

encoded = jsonable_encoder(data)
# すべて json.dumps() 可能な型に変換済み

Django の DjangoJSONEncoder が datetime, Decimal, UUID の変換を担っていた（1.7 章（レスポンス生成））のと同様に、jsonable_encoder は FastAPI / Pydantic のエコシステムで同じ役割を果たします。 Pydantic モデルは model_dump() で辞書に変換され、各フィールドの型アノテーションに従ってシリアライズされます。

jsonable_encoder は再帰的に動作するため、ネストされた辞書やリスト内の Pydantic モデル・datetime などもすべて変換されます。

注意

巨大なデータ構造に対しては、再帰の深さとパフォーマンスに注意が必要です。必要なフィールドだけを含む response_model を適切に設計することで、変換コストを抑えられます。

4.7.4. ステータスコード

FastAPI では status_code パラメータでデフォルトのステータスコードを宣言し、HTTPException で異常系のコードを返します。

ステータスコードの使い分けは次の通りです。

status_code デコレータ引数 → 正常系のデフォルトステータスを宣言（OpenAPI スキーマにも反映）
HTTPException → 異常系を明示的に送出（{"detail": "..."} 形式の JSON レスポンスが自動生成）

from fastapi import HTTPException, status

@app.post("/users", status_code=status.HTTP_201_CREATED, response_model=UserResponse)
async def create_user(user: UserCreate):
    if await user_exists(user.email):
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"User with email {user.email} already exists",
        )
    new_user = await save_user(user)
    return new_user

@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(user_id: int):
    if not await user_exists_by_id(user_id):
        raise HTTPException(status_code=404, detail="User not found")
    await remove_user(user_id)
    # 204 の場合、戻り値なし（ボディなし）

HTTPException は Starlette の例外クラスを FastAPI が拡張したもので、headers パラメータでレスポンスヘッダーを追加できます。認証系のエンドポイントで WWW-Authenticate ヘッダーを返す場合などに使います。

raise HTTPException(
    status_code=401,
    detail="Invalid credentials",
    headers={"WWW-Authenticate": "Bearer"},
)

4.7.5. カスタムレスポンス

JSON 以外のレスポンスを返す場合、Starlette が提供する Response サブクラスを直接使います。

主なレスポンスクラスと用途は次の通りです。

クラス	用途
`HTMLResponse`	HTML テキストを返す
`PlainTextResponse`	プレーンテキストを返す
`RedirectResponse`	リダイレクト（301/302/307 など）
`StreamingResponse`	ストリーミング・SSE
`FileResponse`	ファイルダウンロード

from fastapi.responses import (
    HTMLResponse,
    PlainTextResponse,
    RedirectResponse,
    StreamingResponse,
    FileResponse,
)

@app.get("/", response_class=HTMLResponse)
async def index():
    return "<html><body><h1>Hello, World!</h1></body></html>"

@app.get("/health", response_class=PlainTextResponse)
async def health():
    return "OK"

@app.get("/old-path")
async def redirect():
    return RedirectResponse(url="/new-path", status_code=301)

StreamingResponse は 2.8 章（ASGI の HTTP モデル）で見た more_body=True のチャンク送信を抽象化したクラスです。

import asyncio

async def generate_sse():
    for i in range(10):
        yield f"data: {{\"count\": {i}}}\n\n"
        await asyncio.sleep(1)
    yield "data: {\"message\": \"done\"}\n\n"

@app.get("/stream")
async def sse_endpoint():
    return StreamingResponse(
        generate_sse(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

StreamingResponse の __call__ メソッド内部では、http.response.start を一度送った後、ジェネレータから値を取り出すたびに http.response.body を more_body=True で送り、最後に more_body=False で締めくくります。 3 章（最小の ASGI HTTP アプリ）で SSE アプリを手書きしたときの for ループと同じ構造です。

注釈

X-Accel-Buffering: no は Nginx のバッファリングを無効化し、チャンクが即座にクライアントへ転送されるようにするヘッダーです。 SSE を Nginx 経由で配信する場合は必ず付与してください。1.10 章（どこまでが Django の責務で、どこからがサーバの責務か）のリバースプロキシ設定と関連しています。

FileResponse はファイルパスを指定するだけで次の処理を自動実行します。

Content-Type の自動判定
Content-Length の設定
Content-Disposition の付与

@app.get("/download/{filename}")
async def download_file(filename: str):
    file_path = f"/data/exports/{filename}"
    return FileResponse(
        path=file_path,
        filename=filename,
        media_type="application/octet-stream",
    )

内部的には Starlette の FileResponse が aiofiles でファイルをチャンク読みし、StreamingResponse と同様に send を繰り返します。 Django の FileResponse が wsgi.file_wrapper でカーネルの sendfile を利用していた（1.7 章（レスポンス生成））のに対し、Starlette の実装はユーザースペースでの非同期読み取りが基本です。

レスポンス生成の全体フローをまとめると次の通りです。

ビュー関数の戻り値
  │
  ├─ Response オブジェクト → そのまま使用
  │
  └─ dict / list / Pydantic モデル / ORM オブジェクト
       │
       ├─ response_model あり → model_validate() でフィルタリング・検証
       │    → model_dump() で辞書化
       │
       ├─ jsonable_encoder() で JSON 互換型に変換
       │    datetime → ISO文字列, Decimal → float, UUID → 文字列, ...
       │
       └─ JSONResponse(content=encoded, status_code=status_code)
            │
            ├─ json.dumps(content).encode("utf-8")
            ├─ await send({"type": "http.response.start", "status": ..., "headers": ...})
            └─ await send({"type": "http.response.body", "body": ...})

3 章（最小の ASGI HTTP アプリ）の手書きコードでは json.dumps → send_json → http.response.start / http.response.body を明示的に書いていました。 FastAPI ではこのパイプラインがビュー関数の return 文の後ろに隠れていますが、構造は同一です。

Tip

レスポンスが期待通りに生成されないとき（フィールドが欠落する・datetime が文字列にならない・ストリーミングが途切れるなど）は、この変換パイプラインのどの段階で問題が起きているかを追跡してみてください。内部構造を学ぶ実務上の価値はここにあります。

次節では、FastAPI のミドルウェアと例外ハンドリングの仕組みを追い、リクエスト受信からレスポンス返却までの全フローを完成させます。

4.8. middleware と exception handler

4.8.1. Starlette middleware

FastAPI のミドルウェアは Starlette のミドルウェア機構そのものです。 3.6 章（ASGI ミドルウェアを書く）で手書きした ASGI ミドルウェアと同じ「アプリをラップして scope / receive / send を傍受する」構造が、FastAPI でも使われています。

FastAPI ではミドルウェアを次の2つのスタイルで登録できます。

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import time

app = FastAPI()

# スタイル1: @app.middleware("http") デコレータ
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    elapsed = time.time() - start
    response.headers["X-Process-Time"] = f"{elapsed:.4f}"
    return response

@app.middleware("http") は内部的に BaseHTTPMiddleware を使います。各引数の役割は次の通りです。

request — Starlette の Request オブジェクト（scope と receive をラップ）
call_next — 次のミドルウェアまたはルーターを呼び出す関数（3.6 章（ASGI ミドルウェアを書く）の await self.app(scope, receive, send) に相当）
response — Response オブジェクトとして返され、ヘッダーの追加や差し替えが可能

# スタイル2: add_middleware() で Starlette/サードパーティのミドルウェアクラスを登録
from starlette.middleware.trustedhost import TrustedHostMiddleware
from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware

app.add_middleware(TrustedHostMiddleware, allowed_hosts=["example.com", "*.example.com"])
app.add_middleware(HTTPSRedirectMiddleware)

add_middleware() はクラスとそのコンストラクタ引数を受け取り、起動時にミドルウェアスタックを構築します。 3.6 章（ASGI ミドルウェアを書く）で app = LoggingMiddleware(ErrorHandlingMiddleware(Router())) と手動でラップしたのと同じネスト構造が、add_middleware の呼び出し順で決まります。

ミドルウェアの実行順序は登録順の「外側から内側」です。

警告

@app.middleware("http") では、後に登録されたミドルウェアが外側になります。これは Starlette がミドルウェアをスタック（後入れ先出し）として積み上げるためです。 Django の MIDDLEWARE リストが先頭から外側であったのとは逆の感覚になるため、注意が必要です。

app = FastAPI()

@app.middleware("http")
async def middleware_a(request: Request, call_next):
    print("A: before")
    response = await call_next(request)
    print("A: after")
    return response

@app.middleware("http")
async def middleware_b(request: Request, call_next):
    print("B: before")
    response = await call_next(request)
    print("B: after")
    return response

# 出力順序:
# B: before  ← 後に登録されたものが外側
# A: before
# (ビュー関数実行)
# A: after
# B: after

add_middleware() も同様の挙動です。

純粋な ASGI ミドルウェアクラスもそのまま使えます。 3.6 章（ASGI ミドルウェアを書く）で作った SecurityHeadersMiddleware を FastAPI に組み込む場合は次のようになります。

from middleware import SecurityHeadersMiddleware

app.add_middleware(SecurityHeadersMiddleware)

注意

BaseHTTPMiddleware は使いやすい反面、内部で receive を一度消費してリクエストボディをメモリに読み込む設計のため、大きなファイルアップロードやストリーミングリクエストでメモリを圧迫する可能性があります。パフォーマンスが重要な場合や receive / send を細かく制御したい場合は、3.6 章（ASGI ミドルウェアを書く）で書いたような純粋 ASGI ミドルウェアの方が適しています。

4.8.2. 例外の変換

FastAPI の例外ハンドリングは、Starlette の ServerErrorMiddleware と ExceptionMiddleware の2層構造で実現されています。

各ミドルウェアの役割は次の通りです。

ミドルウェア	役割
`ServerErrorMiddleware`	最外層。未捕捉例外を 500 に変換
ユーザー登録ミドルウェア	`add_middleware` / `@app.middleware` で登録したもの
`ExceptionMiddleware`	例外ハンドラのディスパッチ。HTTPException・ValidationError → JSON

クライアント
  ↕
ServerErrorMiddleware          ← 最外層：未捕捉例外を 500 に変換
  ↕
ユーザー登録ミドルウェア        ← add_middleware / @app.middleware
  ↕
ExceptionMiddleware            ← 例外ハンドラのディスパッチ
  ↕
Router → APIRoute → ビュー関数

ExceptionMiddleware は登録された例外ハンドラを管理し、ビュー関数やミドルウェアから送出された例外の型に応じて適切なハンドラを呼び出します。 FastAPI はデフォルトで次の2つの例外ハンドラを登録しています。

# FastAPI が内部で登録するデフォルトハンドラ（概念コード）
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return JSONResponse(status_code=422, content={"detail": exc.errors()})

@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail},
                        headers=exc.headers)

HTTPException はビュー関数や依存関数（Depends）から送出され、ExceptionMiddleware が捕捉して JSON レスポンスに変換します。 1 章（Django を WSGI 視点で見る）で Django の Http404 → 404、PermissionDenied → 403 への変換を見ましたが、FastAPI では HTTPException(status_code=404) のように任意のステータスコードを指定できます。

RequestValidationError は 4.4 章（リクエストボディの解析）・4.5 章（Pydantic によるデータ検証）で見た通り、Pydantic のバリデーション失敗時に自動送出されます。

カスタム例外ハンドラを登録することで、アプリケーション固有のエラー形式を統一することができます。

class BusinessLogicError(Exception):
    def __init__(self, code: str, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code

@app.exception_handler(BusinessLogicError)
async def business_error_handler(request: Request, exc: BusinessLogicError):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "code": exc.code,
                "message": exc.message,
            }
        },
    )

@app.post("/transfer")
async def transfer(amount: float, from_account: str, to_account: str):
    if amount <= 0:
        raise BusinessLogicError(code="INVALID_AMOUNT", message="Amount must be positive")
    if from_account == to_account:
        raise BusinessLogicError(code="SAME_ACCOUNT", message="Cannot transfer to the same account")
    # ...

例外ハンドラの解決順序は、例外クラスの継承階層に従います。 BusinessLogicError のハンドラが登録されていなければ、Exception のハンドラが呼ばれ、それもなければ ServerErrorMiddleware が 500 を返します。 Django の process_exception がミドルウェアの逆順で呼ばれるのに対し、FastAPI / Starlette は例外の型でディスパッチするアプローチを採用しています。

ServerErrorMiddleware は最外層に位置し、ExceptionMiddleware やユーザーミドルウェアで捕捉されなかった例外をすべて受け止めます。 DEBUG=True のときはスタックトレースを含む HTML エラーページを返し、DEBUG=False のときは素の “Internal Server Error” を返します。これは Django の DEBUG 設定による詳細エラーページの表示（1.8 章（Django が面倒を見てくれているもの））と同じ設計判断です。

例外がどの層で処理されるかを整理すると次の通りです。

例外の種類                     捕捉する層                レスポンス
──────────────────────     ──────────────────────     ──────────
RequestValidationError     ExceptionMiddleware        422 + detail
HTTPException              ExceptionMiddleware        指定した status_code + detail
BusinessLogicError         ExceptionMiddleware        カスタムハンドラの戻り値
(send 済み後の例外)          ServerErrorMiddleware      ログ出力のみ（送信不可）
未登録の Exception          ServerErrorMiddleware      500 (DEBUG で詳細表示)

注釈

3.6 章（ASGI ミドルウェアを書く）の ErrorHandlingMiddleware で response_started フラグを確認したのと同様に、http.response.start が送信済みの場合は新たなレスポンスを生成できません。 Starlette の ServerErrorMiddleware もこの状態を検出し、ログに記録するだけで処理を終えます。

4.8.3. CORS などの位置づけ

Cross-Origin Resource Sharing（CORS）は、ブラウザが異なるオリジンへのリクエストを許可するかどうかを制御する仕組みです。 FastAPI では Starlette の CORSMiddleware を add_middleware() で登録します。

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://frontend.example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
    max_age=3600,
)

CORSMiddleware はリクエストの Origin ヘッダーを検査し、プリフライト（OPTIONS）リクエストにはビュー関数を呼ばずにレスポンスを返し、通常リクエストにはレスポンスに Access-Control-Allow-Origin などのヘッダーを付加します。 3.6 章（ASGI ミドルウェアを書く）で SecurityHeadersMiddleware が send をラップしてヘッダーを注入したのと同じパターンで、CORSMiddleware は send イベントの http.response.start に CORS ヘッダーを追加しています。

重要

CORS ミドルウェアの配置位置は重要です。認証ミドルウェアよりも外側（先に処理される位置）に置かなければ、プリフライトリクエストが認証で弾かれてしまいます。プリフライトは認証トークンを持たない OPTIONS リクエストであるため、認証チェックを通過できません。

# 正しい順序
app.add_middleware(CORSMiddleware, ...)         # 外側：プリフライトをここで処理
app.add_middleware(AuthenticationMiddleware)      # 内側：通常リクエストのみ到達

Starlette / FastAPI が提供する主要なミドルウェアを整理すると次の通りです。

ミドルウェア	役割
`CORSMiddleware`	CORS ヘッダー付与・プリフライト応答
`TrustedHostMiddleware`	`Host` ヘッダー検証（Django の `ALLOWED_HOSTS` に相当）
`HTTPSRedirectMiddleware`	HTTP → HTTPS リダイレクト
`GZipMiddleware`	レスポンス圧縮（Nginx の `gzip on` に相当）

これらのミドルウェアは Django が settings.py の MIDDLEWARE リストと各種設定（ALLOWED_HOSTS, CSRF_COOKIE_SECURE など）で提供していた機能を、明示的なクラスとコンストラクタ引数で設定する方式に置き換えたものです。 Django が「設定ファイルに書けば自動で有効になる」batteries-included アプローチを取るのに対し、FastAPI / Starlette は「必要なミドルウェアを明示的に登録する」最小構成アプローチを取ります。どちらが優れているかではなく、設計思想の違いです。

ミドルウェアと例外ハンドラを含めた FastAPI の全体フローは次の通りです。

Uvicorn (ASGI サーバ)
  │ scope, receive, send
  ▼
ServerErrorMiddleware          ← 未捕捉例外 → 500
  │
CORSMiddleware                 ← CORS ヘッダー付与 / プリフライト応答
  │
GZipMiddleware                 ← レスポンス圧縮
  │
カスタムミドルウェア             ← ロギング、認証、リクエスト ID など
  │
ExceptionMiddleware            ← HTTPException / ValidationError → JSON
  │
Router                         ← パスマッチング
  │
APIRoute                       ← パラメータ抽出、Pydantic 検証、DI 解決
  │
ビュー関数                      ← ビジネスロジック
  │
戻り値 → JSONResponse          ← response_model フィルタ → send

1.5 章（middleware chain の流れ）で Django のミドルウェアチェーンを「タマネギ構造」として追跡しましたが、FastAPI でもまったく同じ構造が ASGI のイベントモデル上で実現されています。違いは次の点です。

Django: process_request / process_view / process_response / process_exception のフックメソッドで段階を分ける
FastAPI / Starlette: call_next を中心とした単一の前後処理パターンに統一

次節では、FastAPI の OpenAPI 自動生成と、ここまで学んだ型ヒント・DI・レスポンスモデルがドキュメントにどう反映されるかを見ていきます。

4.9. OpenAPI はどう生成されるのか

4.9.1. 型情報

FastAPI の OpenAPI スキーマ自動生成は、ここまでの章で見てきた型ヒント・Pydantic モデル・Depends()・response_model の情報をすべて集約した結果です。新たな仕組みが加わるのではなく、既存の宣言が別の出力形式（JSON Schema / OpenAPI 3.x）に変換されます。

注釈

OpenAPI スキーマは FastAPI がコードを解析して自動生成します。コードが唯一の情報源（single source of truth）として機能するため、実装とドキュメントの乖離が構造的に起きにくい設計です。

from pydantic import BaseModel, Field, EmailStr
from fastapi import FastAPI, Query, Path, Header

app = FastAPI()

class UserCreate(BaseModel):
    """ユーザー作成リクエスト"""
    name: str = Field(min_length=1, max_length=100, examples=["Taro Yamada"])
    email: EmailStr
    age: int | None = Field(default=None, ge=0, le=150, description="年齢（任意）")

class UserResponse(BaseModel):
    """ユーザーレスポンス"""
    id: int
    name: str
    email: EmailStr
    age: int | None = None
    created_at: str

@app.post("/users", response_model=UserResponse, status_code=201,
          summary="ユーザーを作成する", tags=["users"])
async def create_user(user: UserCreate):
    """新しいユーザーを作成し、作成されたユーザー情報を返します。"""
    return {"id": 1, "name": user.name, "email": user.email,
            "age": user.age, "created_at": "2025-01-15T10:30:00"}

この定義から FastAPI が生成する OpenAPI スキーマの該当部分は、次のような構造になります。

{
  "paths": {
    "/users": {
      "post": {
        "summary": "ユーザーを作成する",
        "description": "新しいユーザーを作成し、作成されたユーザー情報を返します。",
        "tags": ["users"],
        "requestBody": {
          "required": true,
          "content": {
            "application/json": {
              "schema": { "$ref": "#/components/schemas/UserCreate" }
            }
          }
        },
        "responses": {
          "201": {
            "content": {
              "application/json": {
                "schema": { "$ref": "#/components/schemas/UserResponse" }
              }
            }
          },
          "422": {
            "content": {
              "application/json": {
                "schema": { "$ref": "#/components/schemas/HTTPValidationError" }
              }
            }
          }
        }
      }
    }
  },
  "components": {
    "schemas": {
      "UserCreate": {
        "title": "UserCreate",
        "description": "ユーザー作成リクエスト",
        "type": "object",
        "required": ["name", "email"],
        "properties": {
          "name": { "type": "string", "minLength": 1, "maxLength": 100,
                     "examples": ["Taro Yamada"] },
          "email": { "type": "string", "format": "email" },
          "age":  { "type": "integer", "minimum": 0, "maximum": 150,
                     "description": "年齢（任意）", "default": null }
        }
      },
      "UserResponse": { "..." : "..." }
    }
  }
}

Pydantic モデルの各フィールドと JSON Schema の対応は次の通りです。

Pydantic の記述	JSON Schema
`Field(min_length=1, max_length=100)`	`minLength` / `maxLength`
`Field(ge=0, le=150)`	`minimum` / `maximum`
`EmailStr`	`format: "email"`
`int \| None = None`	`nullable` 相当
`description="..."`	`description`
`examples=[...]`	`examples`

4.3 章（ルーティングの流れ）で見たパスパラメータ・クエリパラメータ・ヘッダーも同様にスキーマへ変換されます。

@app.get("/users/{user_id}", response_model=UserResponse, tags=["users"])
async def get_user(
    user_id: int = Path(ge=1, description="ユーザーID"),
    fields: str | None = Query(default=None, description="取得するフィールド（カンマ区切り）"),
    x_request_id: str | None = Header(default=None),
):
    return {"id": user_id, "name": "Taro", "email": "[email protected]", "created_at": "..."}

Path(), Query(), Header() の情報はスキーマの parameters セクションに in: path, in: query, in: header としてそれぞれ出力されます。型ヒント int は type: integer に、str | None は type: string + 任意パラメータとして表現されます。 FastAPI が APIRoute 生成時に解析した「どの引数がどこから来るか」という情報（4.3 章（ルーティングの流れ））が、ここでスキーマの in フィールドとして表出してきます。

4.9.2. route metadata

各エンドポイントのデコレータに渡すメタデータが OpenAPI ドキュメントの構造を決定します。

@app.get(
    "/users",
    response_model=list[UserResponse],
    summary="ユーザー一覧を取得する",
    description="ページネーション付きでユーザー一覧を返します。",
    tags=["users"],
    responses={
        200: {"description": "ユーザー一覧"},
        401: {"description": "認証エラー", "content": {
            "application/json": {
                "example": {"error": {"code": "UNAUTHORIZED", "message": "Invalid token"}}
            }
        }},
    },
    deprecated=False,
    operation_id="listUsers",
)
async def list_users(
    skip: int = Query(0, ge=0, description="スキップする件数"),
    limit: int = Query(10, ge=1, le=100, description="取得する最大件数"),
    current_user: User = Depends(get_current_user),
):
    """
    認証済みユーザーのみアクセス可能です。

    - **skip**: ページネーションのオフセット
    - **limit**: 1ページあたりの最大件数
    """
    return [...]

デコレータで使用できる主なメタデータは次の通りです。

引数	役割
`summary`	エンドポイント一覧での短い説明
`description` / docstring	詳細説明
`tags`	エンドポイントのグループ化（Swagger UI でセクション分け）
`responses`	正常系以外のレスポンスパターンの宣言
`operation_id`	クライアントコード生成ツールが関数名として使用
`deprecated=True`	UI 上で取り消し線を表示

Depends(get_current_user) で注入される認証依存が HTTPBearer や OAuth2PasswordBearer を使っている場合、セキュリティスキーマも自動生成されます。

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    user = await verify_token(token)
    if user is None:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

HTTPBearer() は Depends() として機能すると同時に、OpenAPI の securitySchemes セクションに type: http, scheme: bearer を登録します。このエンドポイントの OpenAPI 定義には security: [{"HTTPBearer": []}] が付与され、Swagger UI にロックアイコンと「Authorize」ボタンが表示されます。 DI の仕組み（4.6 章（Dependency Injection の仕組み））とスキーマ生成が統合されている好例です。

4.9.3. docs UI の仕組み

FastAPI が /docs と /redoc で対話的なドキュメントを提供する仕組みは、非常にシンプルです。

FastAPI の起動時に、app.openapi() メソッドが呼ばれます。このメソッドは app.routes に登録されたすべての APIRoute を走査し、各ルートの APIRoute が保持する次の情報を集約して、OpenAPI 3.x 準拠の Python 辞書を構築します。

シグネチャ解析結果（dependant オブジェクト）
response_model
メタデータ（summary, tags, responses など）
セキュリティ依存（HTTPBearer, OAuth2PasswordBearer など）

構築された辞書はキャッシュされ、2回目以降は再計算されません。

# FastAPI が内部で自動登録するエンドポイント（概念コード）
@app.get("/openapi.json", include_in_schema=False)
async def openapi_schema():
    return app.openapi()  # キャッシュ済みの辞書を JSONResponse で返す

@app.get("/docs", include_in_schema=False)
async def swagger_ui_html():
    return HTMLResponse(f"""
    <!DOCTYPE html>
    <html>
    <head>
        <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/swagger-ui-dist/swagger-ui.css">
    </head>
    <body>
        <div id="swagger-ui"></div>
        <script src="https://cdn.jsdelivr.net/npm/swagger-ui-dist/swagger-ui-bundle.js"></script>
        <script>
            SwaggerUI({{ url: "/openapi.json", dom_id: "#swagger-ui" }})
        </script>
    </body>
    </html>
    """)

@app.get("/redoc", include_in_schema=False)
async def redoc_html():
    return HTMLResponse(f"""
    <!DOCTYPE html>
    <html>
    <head>
        <script src="https://cdn.jsdelivr.net/npm/redoc/bundles/redoc.standalone.js"></script>
    </head>
    <body>
        <redoc spec-url="/openapi.json"></redoc>
    </body>
    </html>
    """)

/docs は Swagger UI の JavaScript ライブラリを CDN から読み込む HTML ページを返しているだけです。 Swagger UI は /openapi.json を fetch で取得し、クライアントサイドでレンダリングします。 /redoc も同様に ReDoc のライブラリを読み込む HTML です。 FastAPI 自体がドキュメントの UI をレンダリングしているわけではなく、OpenAPI JSON を生成し、それを読み取る既存のフロントエンドツールへ HTML を提供しているだけです。

Tip

スキーマさえ正確であれば UI は差し替え可能です。Swagger UI や ReDoc 以外のツール、あるいは openapi-generator によるクライアント SDK 自動生成も、同じ /openapi.json を入力として使えます。

スキーマをカスタマイズする場合は app.openapi() をオーバーライドします。

from fastapi.openapi.utils import get_openapi

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema
    openapi_schema = get_openapi(
        title="My API",
        version="2.0.0",
        description="本書のサンプル API です。",
        routes=app.routes,
    )
    openapi_schema["info"]["x-logo"] = {"url": "https://example.com/logo.png"}
    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

OpenAPI 生成の全体的な流れをまとめると次の通りです。

開発者が書くコード                    OpenAPI スキーマへの変換
─────────────────────              ──────────────────────────
class UserCreate(BaseModel):       → components/schemas/UserCreate (JSON Schema)
  name: str = Field(min_length=1)    → properties.name.minLength: 1
  email: EmailStr                    → properties.email.format: "email"

@app.post("/users",                → paths./users.post
    response_model=UserResponse,     → responses.201.schema.$ref
    status_code=201,                 → responses のキー "201"
    tags=["users"],                  → tags: ["users"]
    summary="ユーザー作成")           → summary: "ユーザー作成"

async def create_user(             → requestBody + parameters
    user: UserCreate                 → requestBody.content.application/json.schema
):
    """docstring"""                 → description

Depends(get_current_user)          → security + securitySchemes
  └ HTTPBearer()                     → securitySchemes.HTTPBearer.type: http

4 章（FastAPI を ASGI 視点で見る）を通じて見てきた型ヒント → パラメータ抽出（4.3 章（ルーティングの流れ））、Pydantic モデル → バリデーション（4.5 章（Pydantic によるデータ検証））、Depends → DI 解決（4.6 章（Dependency Injection の仕組み））、response_model → レスポンスフィルタリング（4.7 章（レスポンス生成））のすべてが、OpenAPI スキーマ生成にも再利用されています。コードが唯一の情報源（single source of truth）として機能し、実装とドキュメントの乖離が構造的に起きにくい設計です。

注意

この仕組みには限界もあります。次のようなケースはスキーマに正確に反映されません。

動的に生成されるレスポンス（条件によって構造が変わる）
WebSocket エンドポイント（OpenAPI 3.0 では未サポート）
複雑な認可ロジック（ロールごとのレスポンス差異）

これらの場合は responses パラメータで手動補完するか、ドキュメントの外部補足が必要です。

次節では FastAPI のトラブルシューティング観点を整理し、ここまで学んだ内部構造の知識がどのような問題解決に直結するかを確認します。

4.10. トラブルシューティングの観点

4.10.1. 422 エラーの意味

FastAPI を使い始めて最初に遭遇しやすい問題が 422 Unprocessable Entity です。このステータスコードは「リクエストの構文は正しいが、内容がバリデーションに通らなかった」ことを意味します。

ステータス	意味
400 Bad Request	リクエスト自体が不正（構文エラーなど）
422 Unprocessable Entity	構文は正しいが内容がバリデーションに通らなかった
500 Internal Server Error	サーバ側の障害

422 が返されたとき、レスポンスボディの detail 配列が原因の特定に直結します。

{
  "detail": [
    {
      "type": "missing",
      "loc": ["body", "email"],
      "msg": "Field required",
      "input": {"name": "Taro"}
    }
  ]
}

loc を読めば、どのパラメータがどの取得元で失敗したかを即座に判別できます。

`loc` の例	意味
`["body", "email"]`	リクエストボディの `email` フィールド
`["path", "user_id"]`	パスパラメータ `user_id`
`["query", "limit"]`	クエリパラメータ `limit`
`["header", "authorization"]`	`Authorization` ヘッダー
`["body", "address", "city"]`	ネストされたモデルの `address.city`

この先頭要素は、4.3 章（ルーティングの流れ）で解説した FastAPI のパラメータ取得元の自動判別がそのまま反映されたものです。

実務で遭遇しやすい 422 のパターンを以下にまとめます。

Content-Type ヘッダーの欠落（最も多い原因）

curl -X POST /users -d '{"name": "Taro"}' のように -H "Content-Type: application/json" を忘れると、Starlette はボディを JSON としてパースしません。結果として Pydantic に空のデータが渡され、すべての必須フィールドが missing エラーになります。エラーメッセージだけ見ると「フィールドが足りない」ように見えますが、実際にはボディのパース段階で失敗しています。

パスパラメータの型不一致

/users/{user_id} に int を宣言しているのに /users/abc にアクセスすると、loc: ["path", "user_id"] で type: int_parsing エラーが返ります。 Starlette のルーターはパスの文字列マッチングだけを行い、型変換は FastAPI / Pydantic が担当するため、パスは一致するがパラメータの変換で失敗するという状況が起きます。

クエリパラメータのデフォルト値忘れ

limit: int = Query() と書くと必須パラメータになり、クエリ文字列に含めないと 422 が返ります。 limit: int = Query(10) のようにデフォルト値を設定するか、limit: int | None = Query(default=None) で任意パラメータにしてください。

4.10.2. validation の責務

422 エラーを適切に扱うためには、バリデーションがアプリケーションのどの層で行われるべきかを整理する必要があります。

バリデーションの責務は次のように分担するのが FastAPI の設計意図に沿っています。

層	責務	実装方法
Pydantic モデル	入力データの形式・型が正しいか	`Field(min_length=1, ge=0)`, `EmailStr` など
ビュー関数 / 依存関数	データの意味が正しいか（ビジネスロジック）	`HTTPException(409/404/403)`

FastAPI / Pydantic のバリデーションは「入力データの形式と型が正しいか」を検証する層です。フィールドの存在確認・型変換・範囲チェック・正規表現マッチングなどがここに該当します。

一方で「そのメールアドレスが既に登録されているか」「指定されたユーザー ID が存在するか」「現在のユーザーにその操作の権限があるか」はビジネスロジックの検証であり、Pydantic モデルの責務ではありません。これらはビュー関数や依存関数内で検証し、HTTPException で適切なステータスコードを返してください。

class UserCreate(BaseModel):
    name: str = Field(min_length=1, max_length=100)  # ← 形式の検証
    email: EmailStr                                    # ← 形式の検証

@app.post("/users", status_code=201)
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db_session)):
    # Pydantic による形式検証はここに到達する前に完了している
    existing = await db.execute(select(User).where(User.email == user.email))
    if existing.scalar_one_or_none():
        raise HTTPException(status_code=409, detail="Email already registered")  # ← ビジネスロジックの検証
    new_user = User(name=user.name, email=user.email)
    db.add(new_user)
    await db.commit()
    return new_user

この責務分離が曖昧になると、次の問題が起きます。

Pydantic モデル内で DB アクセスを行う（@field_validator 内で DB クエリを実行する）→ モデルが外部リソースに依存し、テストが困難になり、OpenAPI スキーマにもその検証が反映されません。
ビュー関数内でフィールドの型チェックを手動で行う → Pydantic の自動検証と重複し、エラーレスポンスの形式も不統一になります。

4.10.3. dependency の循環や重複

4.6 章（Dependency Injection の仕組み）で依存関係グラフの構造を見ましたが、実務ではグラフが複雑になるにつれて問題が顕在化します。

循環依存は、依存関数 A が B に依存し、B が A に依存するケースです。

async def get_service_a(b = Depends(get_service_b)):
    return ServiceA(b)

async def get_service_b(a = Depends(get_service_a)):  # 循環
    return ServiceB(a)

FastAPI は APIRoute 生成時（サーバ起動時）にシグネチャを再帰的に解析するため、この循環は無限再帰を引き起こします。エラーメッセージは RecursionError: maximum recursion depth exceeded となり、どの依存関係が循環しているかが直接示されません。依存グラフが深くなると原因の特定に時間がかかるため、依存関数を追加する際はグラフの構造を意識してください。

解決策としては、次のいずれかを取ります。

共通の依存を抽出して両方から参照する
インターフェースを分離して一方向の依存にリファクタリングする

意図しない重複実行は、use_cache=False を不用意に使った場合や、同じ機能を持つ異なる依存関数を別々に定義した場合に起きます。

# 異なる関数だが同じことをしている
async def get_db_1():
    async with async_session_maker() as session:
        yield session

async def get_db_2():
    async with async_session_maker() as session:
        yield session

@app.get("/items")
async def list_items(
    db1: AsyncSession = Depends(get_db_1),   # セッション A
    db2: AsyncSession = Depends(get_db_2),   # セッション B（別物）
):
    # db1 と db2 は異なるセッション → トランザクションが分離される
    ...

FastAPI のキャッシュは同一リクエスト内で「同じ callable オブジェクト」に対して適用されます。 get_db_1 と get_db_2 は異なる関数オブジェクトなので、それぞれ独立に実行されます。 4.6 章（Dependency Injection の仕組み）で説明した通り、同一関数であればキャッシュが効いて1回だけ実行されますが、見た目は同じでも関数が別であれば別セッションが生成されます。これがトランザクション不整合の原因になることがあります。

Tip

依存関数のデバッグには、関数の冒頭にログを入れるのが最も直接的です。同じ session_id が複数回表示されればキャッシュが効いており、異なる ID が表示されれば別インスタンスが生成されていることがわかります。

依存関数のデバッグには、関数の冒頭にログを入れるのが最も直接的です。

async def get_db_session():
    session = async_session_maker()
    print(f"[DI] get_db_session called, session_id={id(session)}")
    try:
        yield session
    finally:
        print(f"[DI] get_db_session cleanup, session_id={id(session)}")
        await session.close()

4.10.4. async endpoint 内での blocking I/O

2.11 章（トラブルシューティングの観点）と 3.10 章（現場で起きる問題）で触れたこの問題は、FastAPI の実務で最も深刻かつ発見が遅れやすいバグです。

import requests  # 同期ライブラリ

@app.get("/external")
async def get_external_data():
    # この呼び出しが完了するまでイベントループ全体がブロックされる
    response = requests.get("https://api.example.com/data", timeout=5)
    return response.json()

async def で定義されたエンドポイントは Uvicorn のイベントループ上で直接実行されます。 requests.get() は同期ブロッキング関数なので、この呼び出しが完了するまで（最大5秒間）、同じワーカープロセス上の他のすべてのリクエストが処理されません。 10の同時リクエストがあれば、最後のリクエストは最大50秒待たされることになります。

警告

この問題は、単体テストや低負荷の開発環境では顕在化しません。 1リクエストずつ処理する限り正常に動作し、負荷が上がって初めてレスポンスタイムが劣化します。本番環境に出てから発覚するケースが多いため、設計段階から意識してください。

対処法は次の3つです。

① 非同期ライブラリへの置き換え（推奨）

主要な I/O ライブラリには非同期版が存在します。

requests → httpx（async with httpx.AsyncClient() as client）
同期 ORM → asyncpg や SQLAlchemy の async 拡張
open() → aiofiles

import httpx

@app.get("/external")
async def get_external_data():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data", timeout=5)
    return response.json()

② def（同期関数）での定義

4 章（FastAPI を ASGI 視点で見る）で述べた通り、FastAPI は def で定義されたエンドポイントを自動的にスレッドプール（anyio.to_thread.run_sync）で実行します。イベントループはブロックされません。

@app.get("/external")
def get_external_data():  # async ではなく def
    response = requests.get("https://api.example.com/data", timeout=5)
    return response.json()

この方法はスレッドプールのサイズ（デフォルト40スレッド）が上限になりますが、既存の同期ライブラリをそのまま使えるため、移行コストが最も低い選択肢です。

③ run_in_executor による明示的なオフロード

async def 内で部分的に同期処理を呼ぶ必要がある場合に使います。

import asyncio
from functools import partial

@app.get("/external")
async def get_external_data():
    loop = asyncio.get_event_loop()
    response = await loop.run_in_executor(
        None,  # デフォルトのスレッドプールを使用
        partial(requests.get, "https://api.example.com/data", timeout=5),
    )
    return response.json()

問題の検出方法として、2.11 章（トラブルシューティングの観点）で触れた PYTHONASYNCIODEBUG=1 環境変数が有効です。イベントループ上で100ms以上ブロックする処理が警告として出力されます。本番環境では Uvicorn のアクセスログでレスポンスタイムを監視し、特定のエンドポイントだけが遅い場合はそのエンドポイント内のブロッキング呼び出しを疑います。

blocking I/O の予防原則

async def の中では、すべての I/O に await を付けます。付けられないなら def に変えるか run_in_executor で逃がしてください。エンドポイントが async def で定義されている場合、その内部で呼び出すすべての I/O 関数が await 可能かどうかを確認することが、この問題の予防策になります。

3 章（最小の ASGI HTTP アプリ）から 4 章（FastAPI を ASGI 視点で見る）を通じて、生の ASGI から FastAPI までの全層を追いかけてきました。次章では Uvicorn の内部構造を掘り下げ、ASGI サーバがどのように scope を構築し application を呼び出しているかを見ていきます。