⟐ Claude Code/2026-04-07上級

Claude Code × Python FastAPI 本番開発2026 — 設計・型安全コード生成・pytest自動化・Dockerデプロイまで

Claude Code を AI ペアプログラマーとして使い、FastAPI の本番 API サーバーを設計からデプロイまで組み上げた記録です。Pydantic v2・pytest・Docker・CI/CD の実装パターンを扱います。

Claude Code²⁵⁶ Python¹⁷FastAPIpytestDocker⁵API開発バックエンド⁴ 本番運用³⁶

✦ プレミアム記事

個人開発の API サーバーに Redis キャッシュを入れた翌週、レスポンスタイムのグラフがまったく変わっていないことに気づきました。エラーは出ていません。ログも静かです。ただ、キャッシュが一度も当たっていませんでした。

原因はキャッシュキーでした。デコレータがキーを str(kwargs) から組み立てており、その kwargs に DB セッションが混ざっていたのです。セッションの repr にはメモリアドレスが入ります。リクエストごとに別のキーが生まれ、Redis は書き込まれるだけの倉庫になっていました。

FastAPI は型安全で速く、非同期も素直に書けます。Claude Code と組み合わせれば、設計からデプロイまでの骨格は驚くほど短い時間で立ち上がります。けれど「動いているように見えるコード」は、こういう静かな場所に残ります。

以下では、FastAPI の本番 API サーバーを Claude Code で組み上げる工程を実装パターン中心にお伝えします。あわせて、このワークフローで生まれやすい欠陥を手元で実際に動かして測った結果も添えました。生成を速める手と、生成物を疑う手。私はこの両方が要ると考えています。

Python と FastAPI の基礎には触れているけれど、本番品質の設計・テスト・デプロイの手前で止まっている——そうした状況を想定して書いています。

前提知識・準備

必要な環境

本記事のコードを実行するには、以下の環境が必要です。

Python 3.12 以上
Claude Code CLI（最新版）
Docker Desktop
Git

Claude Code のインストールと初期設定については、Claude Code プロジェクト初期化完全自動化ガイドをご参照ください。

プロジェクト構成の全体像

今回構築する API サーバーの構成は以下の通りです。

fastapi-app/
├── CLAUDE.md              # Claude Code 向け設計指針
├── app/
│   ├── main.py            # FastAPI アプリ本体
│   ├── api/
│   │   ├── v1/
│   │   │   ├── router.py  # v1 ルーター
│   │   │   ├── users.py   # ユーザーエンドポイント
│   │   │   └── items.py   # アイテムエンドポイント
│   ├── core/
│   │   ├── config.py      # 環境変数・設定管理
│   │   ├── security.py    # 認証・JWT
│   │   └── database.py    # SQLAlchemy セッション管理
│   ├── models/
│   │   ├── user.py        # SQLAlchemy モデル
│   │   └── item.py
│   └── schemas/
│       ├── user.py        # Pydantic v2 スキーマ
│       └── item.py
├── tests/
│   ├── conftest.py        # pytest フィクスチャ
│   ├── unit/
│   └── integration/
├── Dockerfile
├── docker-compose.yml
└── .github/workflows/ci.yml

CLAUDE.md の設計 — AIに正確なコンテキストを与える

Claude Code がプロジェクトを正確に理解するには、CLAUDE.md の設計が最重要です。これは Claude Code に対する「技術仕様書」であり、適切に書かれた CLAUDE.md があるかどうかで、生成されるコードの品質が劇的に変わります。

CLAUDE.md の記述内容

以下の CLAUDE.md を fastapi-app/ のルートに配置します。

# FastAPI Production API — CLAUDE.md
 
## 技術スタック
- Python 3.12 / FastAPI 0.115+ / Uvicorn
- SQLAlchemy 2.0 (async) + asyncpg (PostgreSQL)
- Pydantic v2（フィールドバリデーション厳格モード）
- JWT 認証（python-jose + passlib bcrypt）
- pytest-asyncio + httpx（非同期テスト）
- Docker + docker-compose
 
## コーディング規約
- すべての関数・クラスに型ヒントを付ける
- async/await を一貫して使用（同期コードを混在させない）
- エンドポイントのレスポンスモデルは必ず Pydantic スキーマで定義
- DBセッションは依存性注入（Depends）で取得する
- HTTPException は app/core/exceptions.py に集約する
 
## テスト規約
- 新機能実装時は必ず pytest テストを同時作成する
- ユニットテスト: app/core/ の純粋関数
- 統合テスト: 実際の DB（テスト用 SQLite in-memory）を使用
- カバレッジ目標: 80% 以上
 
## 禁止事項
- グローバル変数への状態保存
- print デバッグ（logging モジュールを使う）
- ハードコードされた認証情報（必ず環境変数）

この CLAUDE.md があることで、Claude Code は「このプロジェクトでは非同期 SQLAlchemy を使う」「テストは pytest-asyncio で書く」といったコンテキストを自動的に把握し、一貫性のあるコードを生成します。

✦

ここまでお読みいただきありがとうございます。

この記事の続きを読む

この先には、実装コードやベンチマーク結果など、実務でお役に立てる内容をご用意しています。このサイトは広告を掲載しておらず、サーバーや開発にかかる費用はメンバーの皆様のご支援で成り立っています。もしお役に立てていましたら、ご支援いただけますと大変ありがたいです。

この記事で得られること

✦FastAPI の設計・実装・テストを Claude Code に任せるためのプロンプトの組み立て方

✦pytest・Docker・GitHub Actions CI/CD までを、動作を確認した実装コードつきで通す手順

✦生成された Redis キャッシュ・例外階層・レート制限の欠陥を実測で特定し、修正版と測り直した結果まで確認する

Stripe による安全な決済 · いつでもキャンセル可能

✦

この記事を購入する

この先の内容をすべてお読みいただけます。一度のご購入で、いつでも何度でもアクセスできます。このサイトは広告を掲載しておらず、皆さまのご支援がサーバー費用などの運営を支えています。

または

メンバーシップなら全記事が読み放題 →

Step 1: プロジェクト初期化 — Claude Code で骨格を生成する

まず Claude Code を使ってプロジェクト全体の骨格を生成します。

# プロジェクトディレクトリ作成
mkdir fastapi-app && cd fastapi-app
# CLAUDE.md を先に配置（上記内容で）
 
# Claude Code を起動
claude

Claude Code が起動したら、以下のプロンプトを入力します。

CLAUDE.md の仕様に従って、FastAPI 本番 API の完全なプロジェクト骨格を生成してください。

要件:
1. app/core/config.py — pydantic-settings を使った環境変数管理
2. app/core/database.py — SQLAlchemy 2.0 async エンジン + セッションファクトリ
3. app/core/security.py — JWT アクセストークン生成・検証
4. app/main.py — lifespan イベント・CORS・ミドルウェア設定済み
5. requirements.txt — バージョン固定済み
6. pyproject.toml — ruff + mypy 設定

各ファイルを実際に作成してください。

生成されたコードの例

Claude Code が生成する app/core/config.py は以下のようになります。

# app/core/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache
 
class Settings(BaseSettings):
    """アプリケーション設定。環境変数から自動読み込み。"""
 
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
    )
 
    # アプリ設定
    app_name: str = "FastAPI Production API"
    app_version: str = "1.0.0"
    debug: bool = False
 
    # DB 設定
    database_url: str = "postgresql+asyncpg://user:password@localhost/dbname"
    database_pool_size: int = 10
    database_max_overflow: int = 20
 
    # JWT 設定
    secret_key: str  # 必須: 本番では 256bit ランダム文字列
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30
 
    # CORS 設定
    allowed_origins: list[str] = ["http://localhost:3000"]
 
@lru_cache
def get_settings() -> Settings:
    """シングルトンで設定を返す（DI用）。"""
    return Settings()

@lru_cache でシングルトン化することで、Depends(get_settings) で注入したときに毎回ファイルを読み直すコストを避けられます。Claude Code はこういった本番品質のベストプラクティスを自動で組み込んでくれます。

Step 2: モデルとスキーマ設計 — Pydantic v2 × SQLAlchemy 2.0

Claude Code を使ってモデル設計を行う際のポイントは、「要件を自然言語で詳細に伝える」ことです。

app/models/user.py と app/schemas/user.py を作成してください。

ユーザーモデルの要件:
- id: UUID（自動生成）
- email: 一意制約、255文字以内
- hashed_password: str
- is_active: bool（デフォルト True）
- is_superuser: bool（デフォルト False）
- created_at / updated_at: タイムスタンプ自動管理

Pydantic スキーマは以下の3種類:
- UserCreate: パスワード入力用（プレーンテキスト）
- UserRead: APIレスポンス用（パスワード非公開）
- UserUpdate: 部分更新用（全フィールドOptional）

SQLAlchemy 2.0 の Mapped[T] 型アノテーション方式で書くこと。

生成された app/models/user.py の核心部分：

# app/models/user.py
import uuid
from datetime import datetime
from sqlalchemy import String, Boolean, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
class Base(DeclarativeBase):
    pass
 
class User(Base):
    __tablename__ = "users"
 
    id: Mapped[uuid.UUID] = mapped_column(
        primary_key=True, default=uuid.uuid4, index=True
    )
    email: Mapped[str] = mapped_column(
        String(255), unique=True, index=True, nullable=False
    )
    hashed_password: Mapped[str] = mapped_column(nullable=False)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    is_superuser: Mapped[bool] = mapped_column(Boolean, default=False)
    created_at: Mapped[datetime] = mapped_column(
        default=func.now(), server_default=func.now()
    )
    updated_at: Mapped[datetime] = mapped_column(
        default=func.now(), onupdate=func.now(), server_default=func.now()
    )

SQLAlchemy 2.0 の Mapped[T] アノテーションは、IDE の型補完とも完全に連携し、mypy によるスタティック解析も通ります。Claude Code はこの最新スタイルを正確に把握して生成します。

Step 3: APIエンドポイント開発 — ルーティング設計とビジネスロジック

エンドポイントの実装は、CRUD の単純な実装から始めて、段階的に複雑な要件を追加するのがコツです。

ユーザー CRUD エンドポイントの生成

app/api/v1/users.py を実装してください。

エンドポイント一覧:
- POST /users/ — 新規ユーザー登録（重複メールチェック付き）
- GET /users/me — 現在のログインユーザー取得
- GET /users/{user_id} — ユーザー詳細取得（スーパーユーザーのみ）
- PATCH /users/{user_id} — ユーザー情報更新
- DELETE /users/{user_id} — ユーザー削除（論理削除: is_active=False）

共通要件:
- すべて async def で実装
- DB操作は SQLAlchemy async Session を Depends で取得
- 認証済みユーザーの確認は get_current_user 依存性を使用
- 適切な HTTPException（400/401/403/404）を返す
- レスポンスは UserRead スキーマで型付け

生成されたエンドポイントの例：

# app/api/v1/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
 
from app.core.database import get_db
from app.core.security import get_current_user, get_password_hash
from app.models.user import User
from app.schemas.user import UserCreate, UserRead, UserUpdate
 
router = APIRouter(prefix="/users", tags=["users"])
 
@router.post("/", response_model=UserRead, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_in: UserCreate,
    db: AsyncSession = Depends(get_db),
) -> UserRead:
    """新規ユーザーを登録する。メールアドレス重複時は 400 を返す。"""
    # 重複チェック
    result = await db.execute(select(User).where(User.email == user_in.email))
    existing = result.scalar_one_or_none()
    if existing:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="このメールアドレスは既に登録されています。",
        )
    # パスワードハッシュ化
    hashed_pw = get_password_hash(user_in.password)
    db_user = User(email=user_in.email, hashed_password=hashed_pw)
    db.add(db_user)
    await db.commit()
    await db.refresh(db_user)
    return db_user
 
@router.get("/me", response_model=UserRead)
async def read_current_user(
    current_user: User = Depends(get_current_user),
) -> UserRead:
    """現在の認証済みユーザー情報を返す。"""
    return current_user

このコードは response_model=UserRead を指定することで、hashed_password が自動的にレスポンスから除外されます。Claude Code に型安全な設計を徹底させるには、CLAUDE.md でその旨を明記する点が肝心です。

Step 4: pytest 自動テスト — 非同期 API のテストを組む

FastAPI + SQLAlchemy async の組み合わせのテストは、設定が煩雑になりがちです。Claude Code に任せることで、この複雑な conftest.py をすぐに生成できます。

conftest.py の生成プロンプト

tests/conftest.py を作成してください。

要件:
- テスト用 SQLite in-memory DB（aiosqlite）を使用
- 各テストの前後でDBを初期化（isolation 保証）
- pytest-asyncio の asyncio_mode = "auto"
- httpx.AsyncClient でエンドポイントをテスト
- テスト用ユーザー生成フィクスチャ（通常ユーザー + スーパーユーザー）
- JWT トークン生成フィクスチャ

pytest.ini も合わせて作成すること。

生成された tests/conftest.py：

# tests/conftest.py
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.pool import StaticPool
 
from app.main import app
from app.core.database import Base, get_db
from app.core.security import get_password_hash, create_access_token
from app.models.user import User
 
# テスト用インメモリ SQLite エンジン
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
 
test_engine = create_async_engine(
    TEST_DATABASE_URL,
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,  # in-memory 共有のため必須
)
 
@pytest_asyncio.fixture(autouse=True)
async def setup_db():
    """各テスト前にテーブルを作成し、テスト後に削除する。"""
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
 
@pytest_asyncio.fixture
async def db_session():
    """テスト用 DB セッションを提供する。"""
    async with AsyncSession(test_engine) as session:
        yield session
 
@pytest_asyncio.fixture
async def client(db_session: AsyncSession):
    """DB を差し替えた FastAPI テストクライアントを提供する。"""
    app.dependency_overrides[get_db] = lambda: db_session
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as ac:
        yield ac
    app.dependency_overrides.clear()
 
@pytest_asyncio.fixture
async def normal_user(db_session: AsyncSession) -> User:
    """テスト用通常ユーザーを作成する。"""
    user = User(
        email="test@example.com",
        hashed_password=get_password_hash("testpassword123"),
    )
    db_session.add(user)
    await db_session.commit()
    await db_session.refresh(user)
    return user
 
@pytest_asyncio.fixture
def user_token(normal_user: User) -> str:
    """テスト用アクセストークンを生成する。"""
    return create_access_token(subject=str(normal_user.id))

統合テストの例

# tests/integration/test_users.py
import pytest
from httpx import AsyncClient
 
@pytest.mark.asyncio
async def test_create_user_success(client: AsyncClient):
    """正常なユーザー登録のテスト。"""
    response = await client.post(
        "/api/v1/users/",
        json={"email": "new@example.com", "password": "securepass123"},
    )
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "new@example.com"
    assert "hashed_password" not in data  # パスワードが非公開であることを確認
 
@pytest.mark.asyncio
async def test_create_user_duplicate_email(client: AsyncClient, normal_user):
    """重複メールアドレスで 400 が返ることを確認。"""
    response = await client.post(
        "/api/v1/users/",
        json={"email": normal_user.email, "password": "anypassword"},
    )
    assert response.status_code == 400
    assert "既に登録" in response.json()["detail"]
 
@pytest.mark.asyncio
async def test_read_current_user(client: AsyncClient, user_token: str):
    """認証済みユーザーが /me エンドポイントにアクセスできることを確認。"""
    response = await client.get(
        "/api/v1/users/me",
        headers={"Authorization": f"Bearer {user_token}"},
    )
    assert response.status_code == 200
    assert response.json()["email"] == "test@example.com"

このテスト設計のポイントは、本番DBに依存せず、完全にインメモリで動作することです。app.dependency_overrides で本番の get_db をテスト用セッションに差し替える手法は、FastAPI の依存性注入システムを最大限に活用しています。

カバレッジレポートを確認するには：

# カバレッジ付きテスト実行
pytest --cov=app --cov-report=html tests/
# htmlcov/index.html でカバレッジを視覚的に確認
 
# 期待する出力例:
# ==================== 12 passed in 1.83s ====================
# TOTAL: 84%

Step 5: 例外設計とミドルウェア — エラー応答を1か所に集約する

本番 API で最初に破綻するのは、たいていエラー応答の一貫性です。あるエンドポイントは {"detail": "..."} を返し、別のエンドポイントは素の 500 を返す。フロントエンドは文字列を目視で分岐するはめになります。

Claude Code はこの手の「量は多いが判断は少ない」基盤コードの生成が得意です。

app/core/exceptions.py にアプリケーション例外の階層を作ってください。

要件:
- 基底クラス AppException は status_code / detail / error_code を持つ
- サブクラス: NotFoundError, RequestValidationFailure, AuthenticationError, PermissionDeniedError
- AppException を JSON レスポンスへ変換する FastAPI の例外ハンドラを含める
- 追跡用に request_id を全エラー応答へ含める
- Python 組み込み例外と同名のクラス名は使わないこと

最後の1行が肝心です。この指示を入れずに生成させると、PermissionError や ValidationError という組み込み名と衝突するクラス名がほぼ確実に出てきます。理由は後述します。

# app/core/exceptions.py
from fastapi import Request
from fastapi.responses import JSONResponse
import uuid
 
class AppException(Exception):
    """アプリケーション例外の基底クラス。"""
 
    def __init__(
        self,
        status_code: int,
        detail: str,
        error_code: str = "INTERNAL_ERROR",
    ):
        self.status_code = status_code
        self.detail = detail
        self.error_code = error_code
        super().__init__(detail)
 
class NotFoundError(AppException):
    def __init__(self, resource: str, resource_id: str):
        super().__init__(
            status_code=404,
            detail=f"{resource} (id: {resource_id}) が見つかりません。",
            error_code="NOT_FOUND",
        )
 
class RequestValidationFailure(AppException):
    # 組み込みの ValidationError / pydantic.ValidationError と名前を分ける
    def __init__(self, detail: str = "リクエスト内容が不正です。"):
        super().__init__(422, detail, "VALIDATION_FAILED")
 
class AuthenticationError(AppException):
    def __init__(self, detail: str = "認証が必要です。"):
        super().__init__(401, detail, "AUTHENTICATION_REQUIRED")
 
class PermissionDeniedError(AppException):
    # 組み込みの PermissionError と名前を分ける（理由は下記）
    def __init__(self, detail: str = "権限が不足しています。"):
        super().__init__(403, detail, "PERMISSION_DENIED")
 
async def app_exception_handler(request: Request, exc: AppException) -> JSONResponse:
    """AppException 系を一貫した JSON エラー応答へ変換します。"""
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "code": exc.error_code,
                "detail": exc.detail,
                "request_id": getattr(request.state, "request_id", str(uuid.uuid4())),
            }
        },
    )

クラス名を組み込み例外と衝突させてはいけない理由

生成されたコードに class PermissionError(AppException) が含まれていても、テストは通ります。API は 403 を正しく返します。問題はそのモジュールの中で 組み込みの PermissionError が名前ごと消えている ことです。

手元で確認しました（Python 3.10.12）。同じモジュールに独自 PermissionError を定義した状態で、書き込み権限のないディレクトリへファイルを開きます。

class PermissionError(AppException):  # 組み込みを覆い隠している
    ...
 
try:
    open("/tmp/ro_dir/f.txt", "w")   # 実際に EACCES が起きる
except PermissionError:
    print("捕捉できた")
except Exception as e:
    print("すり抜けた ->", type(e).__name__, ":", e)

出力は次のとおりでした。

すり抜けた -> PermissionError : [Errno 13] Permission denied: '/tmp/ro_dir/f.txt'

同じコードから独自クラスの定義だけを外すと、except PermissionError は当然のように捕捉します。つまりこのシャドーイングは、ファイル書き込みやプロセス起動の権限エラーを握りつぶす側ではなく、捕捉し損ねる側に倒れます。しかも例外の型名は表示上まったく同じなので、ログを見ても気づけません。

同じ理屈が ValidationError にも当てはまります。pydantic の ValidationError を import しているモジュールで同名クラスを定義すれば、後から書いたほうが勝ちます。生成プロンプトの段階で名前を指定してしまうのが、いちばん安いのです。

リクエストログのミドルウェア

# app/core/middleware.py
import time
import uuid
import logging
from fastapi import Request, Response
 
logger = logging.getLogger(__name__)
 
async def logging_middleware(request: Request, call_next) -> Response:
    """全リクエストを相関 ID と所要時間つきで記録します。"""
    request_id = str(uuid.uuid4())[:8]
    start_time = time.perf_counter()
 
    # ハンドラ側から参照できるよう request.state に載せる
    request.state.request_id = request_id
 
    response = await call_next(request)
 
    duration_ms = (time.perf_counter() - start_time) * 1000
    response.headers["X-Request-ID"] = request_id
    logger.info(
        "HTTP request",
        extra={
            "request_id": request_id,
            "method": request.method,
            "path": request.url.path,
            "status_code": response.status_code,
            "duration_ms": round(duration_ms, 2),
        },
    )
    return response

app/main.py への登録はこの2行です。

from app.core.middleware import logging_middleware
from app.core.exceptions import AppException, app_exception_handler
 
app.middleware("http")(logging_middleware)
app.add_exception_handler(AppException, app_exception_handler)

ミドルウェアで request.state.request_id を先に載せておくと、例外ハンドラ側の getattr(request.state, "request_id", ...) が同じ ID を拾います。エラー応答の request_id と、その直前のアクセスログが一本の線でつながる。障害調査でいちばん効くのはこの一致です。レスポンスヘッダにも同じ ID を返しておくと、利用者からの問い合わせをログ検索一発で追えます。

Step 6: バックグラウンド処理とレート制限

レスポンスを待たせない処理の逃がし方

ウェルカムメールの送信、レポート生成、キャッシュの温め直し。リクエスト・レスポンスの往復に巻き込む必要のない処理は BackgroundTasks へ逃がします。

# app/api/v1/users.py（抜粋）
from fastapi import BackgroundTasks
from app.services.email import send_welcome_email
 
@router.post("/", response_model=UserRead, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_in: UserCreate,
    background_tasks: BackgroundTasks,
    db: AsyncSession = Depends(get_db),
) -> UserRead:
    """ユーザー登録後、ウェルカムメールをバックグラウンドで送ります。"""
    db_user = User(email=user_in.email, hashed_password=hash_password(user_in.password))
    db.add(db_user)
    await db.commit()
    await db.refresh(db_user)
 
    # レスポンスを返した後に実行される
    background_tasks.add_task(send_welcome_email, email=db_user.email)
 
    return db_user

ひとつ注意があります。BackgroundTasks はワーカープロセスの中で動くだけで、キューではありません。デプロイやクラッシュでプロセスが落ちれば、実行待ちのタスクは黙って消えます。失われては困る処理——決済後の在庫引き当てや、再送のない通知——は最初から Celery + Redis のような永続キューに置きます。Claude Code へは「Celery と Redis ブローカーを追加し、tasks.py にメール送信タスクの例と、FastAPI のエンドポイントから起動する書き方を示してください」と頼めば、docker-compose の追記まで一度に出てきます。

slowapi によるレート制限

レート制限は、たいてい荒らされてから思い出す機能です。ログインのような当てられ方をする経路には最初から入れておきます。

# requirements.txt に slowapi を追加
 
# app/main.py
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
 
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
 
# 対象エンドポイントに適用する
@router.post("/auth/token")
@limiter.limit("5/minute")  # IP あたり 1 分 5 回まで
async def login(request: Request, form_data: OAuth2PasswordRequestForm = Depends()):
    ...

@limiter.limit を付けた関数には、本文で使っていなくても request: Request を必ず宣言します。slowapi は引数名 request を頼りに呼び出し元を特定するためです。手元で外して試したところ、デコレータの評価時点、つまり モジュールの import 時に Exception: No "request" or "websocket" argument on function "<function bad at ...>" で落ちました。本番で黙って壊れるのではなく起動そのものが止まる挙動なので、この点はむしろ親切です。

上の 2/minute 相当で試した実測は、1回目 200 / 2回目 200 / 3回目 429 でした。制限に達したリクエストは _rate_limit_exceeded_handler が 429 に変換します。

なお get_remote_address は接続元 IP をそのまま見ます。Fly.io や Railway のようにロードバランサ越しで動かす場合、全リクエストが同じ IP に見えて制限が全体に効いてしまいます。X-Forwarded-For を信頼する key_func へ差し替えるか、プロキシ側で制限をかけるか、どちらかを先に決めておきます。

Claude Code Hooks でテストを自動化する

Claude Code Hooks 自動化テクニック集で詳しく解説されていますが、Hooks を使うことでファイル保存時に自動でテストを実行できます。

`.claude/settings.json` の設定

{
  "hooks": {
    "PostToolUse": [
      {
        "matcher": "Write|Edit",
        "hooks": [
          {
            "type": "command",
            "command": "cd /path/to/fastapi-app && python -m pytest tests/ -x -q --tb=short 2>&1 | tail -20"
          }
        ]
      }
    ]
  }
}

この設定により、Claude Code がファイルを編集するたびにテストが自動実行されます。テストが失敗すると Claude Code は失敗を認識し、自動的に修正を試みます。これにより、「コード生成 → テスト失敗 → 自動修正」のサイクルが確立され、開発速度と品質が同時に向上します。

Step 7: Docker化 — 本番グレードのコンテナ設計

Dockerfile の生成

本番グレードの Dockerfile を作成してください。

要件:
- マルチステージビルド（builder + runtime）
- Python 3.12-slim ベースイメージ
- 非root ユーザーで実行（セキュリティ）
- .dockerignore で不要ファイルを除外
- イメージサイズを最小化（200MB以下を目標）
- ヘルスチェック設定

生成された Dockerfile：

# ── ステージ1: 依存関係ビルド ──
FROM python:3.12-slim AS builder
 
WORKDIR /build
 
# ビルド依存のみインストール
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc libpq-dev \
    && rm -rf /var/lib/apt/lists/*
 
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt
 
# ── ステージ2: 本番ランタイム ──
FROM python:3.12-slim AS runtime
 
# セキュリティ: 非rootユーザー作成
RUN useradd --create-home --shell /bin/bash appuser
WORKDIR /home/appuser/app
 
# ビルドステージの site-packages のみコピー
COPY --from=builder /root/.local /home/appuser/.local
 
# アプリコードをコピー
COPY --chown=appuser:appuser app/ ./app/
 
USER appuser
 
# PATH に .local/bin を追加
ENV PATH=/home/appuser/.local/bin:$PATH
 
# ヘルスチェック
HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \
    CMD python -c "import httpx; httpx.get('http://localhost:8000/health')" || exit 1
 
EXPOSE 8000
 
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

docker-compose.yml（開発環境）

# docker-compose.yml
version: "3.9"
 
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://postgres:password@db:5432/appdb
      - SECRET_KEY=dev-secret-key-change-in-production
      - DEBUG=true
    depends_on:
      db:
        condition: service_healthy
    volumes:
      - ./app:/home/appuser/app/app  # ホットリロード用
 
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: appdb
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: password
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 5s
      timeout: 5s
      retries: 5
 
volumes:
  postgres_data:

# 開発環境の起動
docker compose up -d
 
# API の動作確認
curl http://localhost:8000/docs
# → Swagger UI が表示されれば成功
 
# ログ確認
docker compose logs -f api

Step 8: GitHub Actions CI/CD パイプライン

Claude Code HTTP Hooks × GitHub Actions 完全統合ガイドでも解説されていますが、ここでは FastAPI 特有の CI/CD 設定を紹介します。

`.github/workflows/ci.yml`

name: CI/CD Pipeline
 
on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]
 
jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ["3.12"]
 
    steps:
      - uses: actions/checkout@v4
 
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}
          cache: "pip"
 
      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest pytest-asyncio pytest-cov httpx aiosqlite
 
      - name: Run linting (ruff)
        run: ruff check app/ tests/
 
      - name: Run type checking (mypy)
        run: mypy app/
 
      - name: Run tests with coverage
        run: |
          pytest tests/ --cov=app --cov-report=xml --cov-fail-under=80
 
      - name: Upload coverage to Codecov
        uses: codecov/codecov-action@v4
 
  deploy:
    needs: test
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
 
    steps:
      - uses: actions/checkout@v4
 
      - name: Deploy to Fly.io
        uses: superfly/flyctl-actions/setup-flyctl@master
 
      - run: flyctl deploy --remote-only
        env:
          FLY_API_TOKEN: ${{ secrets.FLY_API_TOKEN }}

このパイプラインは、PR ごとに自動でテスト・型チェック・Lint を実行し、main ブランチへのマージ後に自動デプロイします。カバレッジが 80% を下回るとデプロイが失敗する設定になっており、品質の維持が自動化されます。

Step 9: マルチエージェントでコードレビューを自動化する

Claude Code のマルチエージェント並列実行の応用として、FastAPI 開発ではレビューエージェントを別プロセスで動かす設計が効果的です。

レビューエージェントの設定

.claude/review_agent.md を作成します。

# FastAPI コードレビュー専門エージェント
 
あなたは FastAPI のセキュリティ・パフォーマンス専門のコードレビュアーです。
 
## レビュー観点（優先度順）
1. **セキュリティ**: SQLインジェクション・認証バイパス・権限昇格のリスク
2. **N+1問題**: SELECT文のループ実行（selectinload/joinedload の提案）
3. **型安全性**: Pydantic バリデーションの抜け漏れ
4. **非同期安全性**: sync ブロッキング処理の混入（requests → httpx 等）
5. **エラーハンドリング**: 適切な HTTPException の使用
 
## レビュー出力形式
各問題について:
- 場所: ファイル名:行番号
- 重大度: CRITICAL / WARNING / INFO
- 問題: 何が問題か
- 修正案: 具体的なコード例付き

Claude Code のマルチエージェント機能でこのエージェントをサブエージェントとして起動することで、メインの開発セッションと並行してコードレビューを継続的に行えます。

パフォーマンスの落とし穴 — 生成されたコードを実際に測る

ここからは、このワークフローで生成されやすいコードのうち、動いているように見えて動いていないものを扱います。以下の挙動はすべて手元の環境（Python 3.10.12 / FastAPI 0.141.1 / SQLAlchemy 2.0.52）で実行して確かめました。

接続プールの設定

FastAPI と SQLAlchemy の非同期構成で最初に詰まるのは、負荷がかかったときのコネクションプール枯渇です。

# app/core/database.py（本番向け設定）
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.core.config import get_settings
 
settings = get_settings()
 
engine = create_async_engine(
    settings.database_url,
    pool_size=20,            # 常時維持する接続数
    max_overflow=10,         # pool_size を超えて一時的に許す接続数
    pool_pre_ping=True,      # 使う前に接続の生死を確認する
    pool_recycle=3600,       # 1 時間を超えた接続は張り直す
    echo=settings.debug,     # SQL ログはデバッグ時のみ
)
 
async_session_factory = async_sessionmaker(
    engine,
    expire_on_commit=False,  # commit 後の遅延ロードを防ぐ
    class_=AsyncSession,
)

expire_on_commit=False は非同期では特に重要です。既定値のままだと commit() の後にモデルの属性へ触れた瞬間に再取得が走り、非同期コンテキストでは MissingGreenlet になります。レスポンス直前でシリアライズするときに踏みやすい場所です。

pool_size はワーカープロセスごとの値です。uvicorn を 4 ワーカーで起動すれば、データベース側から見える接続数は最大 4 × (20 + 10) = 120 になります。マネージド Postgres の接続上限は思ったより低いので、pool_size を上げる前に上限を確認します。

N+1 クエリ

Claude Code に「プロジェクト内の SQLAlchemy クエリを見直して、リレーションへアクセスしたときに N+1 が起きるものを指摘し、selectinload か joinedload を付けてください」と頼むと、一括で洗い出してくれます。

# BEFORE: ユーザーごとに items が個別クエリになる
users = await db.execute(select(User))
for user in users.scalars():
    print(user.items)  # ここでユーザー数だけ SELECT が飛ぶ
 
# AFTER: eager loading で追加クエリを 1 回にまとめる
from sqlalchemy.orm import selectinload
 
stmt = select(User).options(selectinload(User.items))
result = await db.execute(stmt)
users = result.scalars().all()

Redis キャッシュのデコレータ — 生成コードをそのまま使うと当たりません

冒頭で触れた話に戻ります。この種のプロンプトからは、次のようなキャッシュデコレータがよく生成されます。

def cache(ttl_seconds: int = 300, key_prefix: str = ""):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            cache_key = f"{key_prefix}:{func.__name__}:{str(kwargs)}"  # ここが問題
            cached = await redis_client.get(cache_key)
            if cached:
                return json.loads(cached)
            result = await func(*args, **kwargs)
            await redis_client.setex(cache_key, ttl_seconds, json.dumps(result))  # ここも問題
            return result
        return wrapper
    return decorator

読む分には筋が通っています。実際に FastAPI へ載せて叩くと、二段構えで壊れました。

1つめ。応答モデルを返すエンドポイントでは 500 になります。

TypeError: Object of type ItemRead is not JSON serializable

json.dumps は Pydantic モデルを直列化できません。response_model=list[ItemRead] を宣言した典型的なエンドポイントは、キャッシュを入れた瞬間に全リクエストが失敗します。

2つめ。辞書を返すようにして直列化を通しても、キャッシュは一度も当たりませんでした。

同じ URL へ 2 回リクエストしたときのキーがこれです。

items:get_popular_items:{'db': <FakeSession object at 0x735ea09ee680>}
items:get_popular_items:{'db': <FakeSession object at 0x735ea086ceb0>}

kwargs には Depends(get_db) で注入されたセッションが入っています。__repr__ を持たないオブジェクトの既定表現にはメモリアドレスが含まれるため、リクエストごとにキーが変わります。計測結果は 2 リクエストで ハンドラ実行 2 回・キャッシュエントリ 2 件、つまりヒット率 0% でした。エラーも警告も出ません。Redis の使用量だけが伸びていきます。

修正の方針は 2 つです。キーに含める引数を明示すること。直列化を Pydantic 側に任せること。

# app/core/cache.py
import json
from functools import wraps
from pydantic import TypeAdapter
 
def cache(ttl_seconds: int = 300, key_prefix: str = "", model=None, cache_kwargs: tuple = ()):
    """cache_kwargs に挙げた引数だけをキーに使い、model があれば Pydantic で直列化します。"""
    adapter = TypeAdapter(model) if model is not None else None
 
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            key_parts = [f"{k}={kwargs[k]!r}" for k in cache_kwargs if k in kwargs]
            cache_key = f"{key_prefix}:{func.__name__}:" + "&".join(key_parts)
 
            cached = await redis_client.get(cache_key)
            if cached is not None:
                return adapter.validate_json(cached) if adapter else json.loads(cached)
 
            result = await func(*args, **kwargs)
            payload = adapter.dump_json(result) if adapter else json.dumps(result)
            await redis_client.setex(cache_key, ttl_seconds, payload)
            return result
        return wrapper
    return decorator

使うときは、キーに効かせたい引数を宣言します。

@router.get("/items/popular", response_model=list[ItemRead])
@cache(ttl_seconds=60, key_prefix="items", model=list[ItemRead], cache_kwargs=("limit",))
async def get_popular_items(limit: int = 10, db: AsyncSession = Depends(get_db)):
    ...

同じ条件で測り直した結果です。/items/popular へ 6 回、?limit=5 へ 1 回、合計 7 リクエスト。ハンドラ実行は 2 回、生成されたキーは items:get_popular_items:limit=10 と items:get_popular_items:limit=5 の 2 つだけでした。意図どおりです。

cache_kwargs を明示する形にすると、書く側は毎回「このエンドポイントの応答は何で決まるのか」を考えることになります。手間が増えたように見えますが、キャッシュの取り違えはその問いを飛ばしたときに起きます。私はこの手間のほうを選びます。

なお、ユーザーごとに内容が変わるエンドポイントへこのデコレータを当てるときは、cache_kwargs に利用者を特定する引数を必ず含めてください。含め忘れると、他人の応答が返ります。キャッシュのバグの中でいちばん高くつく種類です。

本番デプロイ前チェックリスト

Fly.io や Railway へのデプロイ前に、以下を必ず確認します。

セキュリティ

SECRET_KEY が 256bit 以上のランダム文字列（openssl rand -hex 32 で生成）
ALLOWED_ORIGINS が本番ドメインのみに限定されている
デバッグモード（DEBUG=false）になっている
依存パッケージに既知の脆弱性がない（pip audit で確認）

パフォーマンス

データベース接続プールサイズが適切（本番: pool_size=20）
重い処理が BackgroundTasks または Celery で非同期化されている
N+1 クエリがない（SQLAlchemy の echo=True でログ確認）

可観測性

構造化ログが設定されている（structlog 推奨）
/health エンドポイントが DB 接続確認を含んでいる
Sentry や Datadog 等のエラー追跡が設定されている

Fly.io へのデプロイは以下のコマンドで完結します。

# Fly.io の初期設定（初回のみ）
fly launch --name my-fastapi-app --region nrt
 
# 環境変数の設定
fly secrets set SECRET_KEY="$(openssl rand -hex 32)"
fly secrets set DATABASE_URL="postgresql+asyncpg://..."
 
# デプロイ実行
fly deploy
 
# デプロイ確認
fly status
fly logs

まとめ

CLAUDE.md で前提を固め、Pydantic と SQLAlchemy でモデルを起こし、pytest と Hooks で自動検証の輪を閉じる。Docker と GitHub Actions まで通せば、個人開発でも本番相当の FastAPI サーバーが手に入ります。ここまでの工程で Claude Code が担った比重は、正直かなり大きいものでした。

一方で、この記事で実際に測ったのは次の3点です。

生成された Redis キャッシュデコレータは response_model 付きのエンドポイントで 500 になり、辞書へ直しても 2 リクエストで 2 キーが生まれてヒット率 0% でした
独自の PermissionError を定義したモジュールでは、except PermissionError が組み込みの権限エラーを取り逃がしました
@limiter.limit を付けた関数から request 引数を外すと、import の時点で例外になりました

3つとも、コードを読んでいるだけでは気づけませんでした。動かして初めて出てきた挙動です。

生成の速さと、生成物を測る手間。この2つは相反しません。個人開発で一人ぶんの手しかない身としては、速く出せるようになったぶんを、そのまま測る側へ回せるようになったと感じています。

まず1つ、手元のプロジェクトでキャッシュデコレータのキーを print してみてください。同じ URL を2回叩いて、同じキーが出るかどうか。それだけで判断がつきます。

お読みいただきありがとうございました。