モダンPython実践 #2 ルーティング、Pydantic モデル、依存性注入

プログラミング言語 Python FastAPI Pydantic
読了 6分

#1 はじめ方とセットアップ では app/main.py 一ファイルにルートを書きました。今回はそれを 拡張可能な構造 に解いていきます。3 つの道具が核心:

  • APIRouter — ルートをモジュール別に分離
  • Pydantic モデル — 入力 / 出力スキーマ、自動検証
  • Depends — 依存性注入で共通ロジック (DB セッション、認証ユーザーなど) を解く

APIRouter — ルートのモジュール化 #

ルートが増えると 1 つのファイルにすべて書くのは現実的ではありません。APIRouter がモジュール別に分離してくれます。

app/api/todos.py 
from fastapi import APIRouter

router = APIRouter(prefix="/todos", tags=["todos"])

@router.get("/")
def list_todos():
    return [{"id": 1, "title": "FastAPI 学習"}]

@router.get("/{todo_id}")
def get_todo(todo_id: int):
    return {"id": todo_id, "title": "..."}
app/main.py 
from fastapi import FastAPI
from app.api import todos

app = FastAPI(title="Todo API")
app.include_router(todos.router)

3 つの効用:

  • prefix="/todos" — すべてのルートの先頭に自動付与。@router.get("/") が GET /todos/ になる
  • tags=["todos"] — Swagger UI でグルーピング
  • モジュール分離 — auth.py、users.py のようにドメイン単位で分けやすい

Pydantic v2 モデル — スキーマの正体 #

Pydantic は dataclass + 検証 + シリアライズ が合わさったライブラリです。中級 #1 で dataclass は強い検証の場面に向いていないと言いました。その答えが Pydantic です。

app/schemas/todo.py 
from datetime import datetime
from pydantic import BaseModel, Field

class TodoCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    description: str | None = None

class TodoUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    description: str | None = None
    done: bool | None = None

class TodoOut(BaseModel):
    id: int
    title: str
    description: str | None
    done: bool
    created_at: datetime

    model_config = {"from_attributes": True}

核心パターン:

  • 入力用 (TodoCreate、TodoUpdate) と 出力用 (TodoOut) を分離
  • 入力はクライアントが送れるもの、出力はサーバーが公開するもの
  • id、created_at のようにサーバーが作るフィールドは 出力にだけ
  • パスワードのように外部に公開してはいけないフィールドは 出力で除外

この分離が セキュリティと明確さ を同時に押さえます。

from_attributes=True — ORM オブジェクトから直接 #

model_config = {"from_attributes": True} オプションは SQLAlchemy モデルのようなオブジェクトから 属性アクセス でデータを読めるようにします (旧 v1 の orm_mode)。#3 で詳しく。

Pydantic v2 のビルトイン検証 #

よく使う検証 
from pydantic import BaseModel, EmailStr, HttpUrl, Field
from datetime import datetime

class UserCreate(BaseModel):
    email: EmailStr                                       # メール形式
    username: str = Field(min_length=3, max_length=30, pattern=r"^[a-zA-Z0-9_]+$")
    password: str = Field(min_length=8)
    website: HttpUrl | None = None                        # URL
    birth_year: int = Field(ge=1900, le=2030)             # 範囲
    bio: str = Field(max_length=500, default="")

EmailStr、HttpUrl は別パッケージが必要です。[standard] オプションなら既に入っています。Field() のオプション:

  • min_length、max_length — 長さ
  • pattern — 正規表現
  • ge、le、gt、lt — 数値の範囲
  • default、default_factory — デフォルト値
  • description、example — Swagger ドキュメント化

ユーザー定義の検証 #

@field_validator / @model_validator 
from pydantic import BaseModel, field_validator, model_validator

class TodoCreate(BaseModel):
    title: str
    priority: int = 1

    @field_validator("title")
    @classmethod
    def title_no_html(cls, v: str) -> str:
        if "<" in v or ">" in v:
            raise ValueError("HTML タグは許可されません")
        return v.strip()

    @model_validator(mode="after")
    def check_priority_with_title(self):
        if self.priority > 5 and "緊急" not in self.title:
            raise ValueError("優先度 5 を超える場合はタイトルに '緊急' が必要")
        return self
  • @field_validator("name") — 単一フィールドの検証 / 変換
  • @model_validator(mode="after") — すべてのフィールドが埋まった後にモデル単位で検証

検証失敗時には自動で 422 Unprocessable Entity レスポンスが返ります。

ルートで Pydantic を使う #

app/api/todos.py 
from datetime import datetime, timezone
from fastapi import APIRouter, HTTPException
from app.schemas.todo import TodoCreate, TodoUpdate, TodoOut
from app.schemas.common import Page

router = APIRouter(prefix="/todos", tags=["todos"])

# 一時的なインメモリストア (#3 で DB に置き換え)
_db: dict[int, TodoOut] = {}
_next_id = 1

@router.post("/", response_model=TodoOut, status_code=201)
def create_todo(payload: TodoCreate) -> TodoOut:
    global _next_id
    todo = TodoOut(
        id=_next_id,
        title=payload.title,
        description=payload.description,
        done=False,
        created_at=datetime.now(timezone.utc),
    )
    _db[_next_id] = todo
    _next_id += 1
    return todo

@router.get("/{todo_id}", response_model=TodoOut)
def read_todo(todo_id: int) -> TodoOut:
    if todo_id not in _db:
        raise HTTPException(status_code=404, detail="Todo not found")
    return _db[todo_id]

@router.patch("/{todo_id}", response_model=TodoOut)
def update_todo(todo_id: int, payload: TodoUpdate) -> TodoOut:
    if todo_id not in _db:
        raise HTTPException(status_code=404, detail="Todo not found")
    current = _db[todo_id]
    update_data = payload.model_dump(exclude_unset=True)
    updated = current.model_copy(update=update_data)
    _db[todo_id] = updated
    return updated

@router.delete("/{todo_id}", status_code=204)
def delete_todo(todo_id: int) -> None:
    if todo_id not in _db:
        raise HTTPException(status_code=404, detail="Todo not found")
    del _db[todo_id]

チェックポイント:

  • payload: TodoCreate — POST/PATCH のボディが自動で検証 / パースされる
  • response_model=TodoOut — レスポンスがこのスキーマでシリアライズ。他のフィールドがあっても自動でフィルタリング
  • status_code=201 — レスポンスコードを明示
  • payload.model_dump(exclude_unset=True) — クライアントが明示したフィールドだけを取得 (PATCH の核心)

exclude_unset — PATCH の本質 #

PATCH リクエストでクライアントが title だけを送ったとき、description=None として処理してはいけません (上書き)。明示されていないフィールドはそのまま残す のが PATCH の意味であり、exclude_unset=True がそれを解決します。

model_copy で部分更新 #

current.model_copy(update={...}) が新しいインスタンスを作りつつ、一部のフィールドだけを更新します。イミュータブルオブジェクトのパターン。

依存性注入 — Depends #

同じコードを複数のルートで繰り返さないために、依存性 に切り出します。

app/api/deps.py 
from typing import Annotated
from fastapi import Depends, HTTPException, Header

def verify_token(authorization: Annotated[str | None, Header()] = None) -> dict:
    if authorization is None:
        raise HTTPException(401, "Authorization header required")
    if not authorization.startswith("Bearer "):
        raise HTTPException(401, "Invalid auth scheme")
    token = authorization.split(" ", 1)[1]
    # 実際の検証は #4 で
    return {"sub": "user123"}

CurrentUser = Annotated[dict, Depends(verify_token)]
ルートで使用 
@router.post("/", response_model=TodoOut)
def create_todo(payload: TodoCreate, user: CurrentUser) -> TodoOut:
    print("user:", user["sub"])
    ...

Depends(fn) が入ったところに fn が返した値 が注入されます。2 つの効用:

  1. 共通ロジックの分離 — 認証、権限、DB セッション、ページネーションなどを一カ所に置き、ルートは受け取るだけ
  2. テスト時の差し替え — app.dependency_overrides[verify_token] = mock_verify で一行で差し替え。#6 で詳しく

依存性も依存性を持てる #

ネストされた依存性 
def get_db_session() -> Iterator[Session]:
    with SessionLocal() as session:
        yield session

DBSession = Annotated[Session, Depends(get_db_session)]

def get_current_user(token: Annotated[str, Depends(...)], db: DBSession) -> User:
    user = db.query(User).filter(User.id == token).first()
    if not user:
        raise HTTPException(401)
    return user

get_current_user が get_db_session に依存。FastAPI が依存性グラフを自動で解き、同じリクエストの中で 1 度だけ実行します。リクエスト単位のキャッシュ も自動。

クラス形式の依存性 — パラメータ付き依存性 #

page params クラス 
from dataclasses import dataclass

@dataclass
class Pagination:
    skip: int = 0
    limit: int = 100

PaginationDep = Annotated[Pagination, Depends()]

@router.get("/")
def list_todos(p: PaginationDep) -> list[TodoOut]:
    return list(_db.values())[p.skip : p.skip + p.limit]

Depends() の空括弧にクラスだけを渡すと、そのクラスのフィールドがクエリパラメータに 自動変換されます。ページネーションのように複数のルートで繰り返されるパラメータを一カ所にまとめるのに便利です。

標準的なレスポンスモデル — ページネーション、エラー #

app/schemas/common.py 
from typing import Generic, TypeVar
from pydantic import BaseModel

T = TypeVar("T")

class Page(BaseModel, Generic[T]):
    items: list[T]
    total: int
    skip: int
    limit: int

class ErrorResponse(BaseModel):
    detail: str
    code: str | None = None
ルートで 
@router.get("/", response_model=Page[TodoOut])
def list_todos(p: PaginationDep) -> Page[TodoOut]:
    items = list(_db.values())
    return Page(
        items=items[p.skip : p.skip + p.limit],
        total=len(items),
        skip=p.skip,
        limit=p.limit,
    )

Page[TodoOut] のように ジェネリックレスポンス で型安全を保ちつつ、メタデータを一緒に返します。中級 #2 の Generic がそのまま動作。

例外ハンドラ — 一貫したエラー形式 #

app/main.py 
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

class NotFoundError(Exception):
    def __init__(self, resource: str, key: str):
        self.resource = resource
        self.key = key

app = FastAPI()

@app.exception_handler(NotFoundError)
async def not_found_handler(request: Request, exc: NotFoundError):
    return JSONResponse(
        status_code=404,
        content={"detail": f"{exc.resource} {exc.key} なし", "code": "not_found"},
    )

ルートで:

@router.get("/{todo_id}")
def read_todo(todo_id: int) -> TodoOut:
    if todo_id not in _db:
        raise NotFoundError("Todo", str(todo_id))
    return _db[todo_id]

ドメイン例外を作り、それが HTTP レスポンスに変換される処理を 1 箇所にまとめます。ルートがすっきりします。

CORS — フロントエンドと分離運用 #

ブラウザアプリが別ドメインの API を呼び出すには CORS の設定が必要です。

CORS ミドルウェア 
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://myapp.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

allow_origins=["*"] は開発段階でだけ。プロダクションでは明示的なドメインリストに絞ってください。

まとめ #

今回つかんだもの:

  • APIRouter — prefix + tags でモジュール別ルート、app.include_router
  • Pydantic v2 — BaseModel、Field 検証、EmailStr/HttpUrl、@field_validator
  • 入力用 / 出力用スキーマの分離 — セキュリティ + 明確さ
  • model_config = {"from_attributes": True} — ORM オブジェクトから直接
  • response_model= でレスポンスのシリアライズ / フィルタリング
  • PATCH は model_dump(exclude_unset=True) + model_copy(update=...)
  • Depends — 認証、DB セッション、ページネーションのような共通ロジックを解く
  • 依存性グラフの自動解決 + リクエスト単位のキャッシュ
  • Page[T] のようなジェネリックレスポンスモデル
  • @app.exception_handler でドメイン例外 → HTTP レスポンス
  • CORS ミドルウェア

次回(#3 DB 連携 — SQLAlchemy 2.x + Alembic)ではこのインメモリ dict を本物の DB に置き換えます。SQLAlchemy 2.x の新スタイル、async session、Alembic でマイグレーションまで。

X