fastapi-patterns

FastAPI のパターン

使用する場面

以下の場合にこのスキルを有効にしてください：

• FastAPI ミドルウェア（CORS、ロギング、タイミング、エラー処理）を設定する
• 複雑な依存性注入チェーンを作成する
• 接続管理を伴う WebSocket エンドポイントを実装する
• OpenAPI ドキュメント（タグ、例、非推奨化）をカスタマイズする
• JWT 認証とロールベースのアクセス依存関係を設定する
• バックグラウンドタスク（軽量または分散型）を実装する
• アプリケーションのライフサイクル（lifespan による起動／終了）を管理する
• レート制限またはリクエストのスロットリングを設定する

以下の場合、このスキルは使用しないでください：

• 基本的なエンドポイントの CRUD、リポジトリ、またはサービスのパターン（python-backend-expert を使用）
• FastAPI エンドポイントのテストを書く（pytest-patterns を使用）
• API コントラクトの設計またはスキーマの計画（api-design-patterns を使用）
• アーキテクチャ上の意思決定（system-architecture を使用）

指示

ミドルウェアスタック

ミドルウェアは LIFO（後入れ先出し）順に実行されます。最後に追加したミドルウェアが最外層になります。

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# Order matters: added last = executed first (outermost)
app.add_middleware(TimingMiddleware)          # 3rd added = runs 1st
app.add_middleware(RequestLoggingMiddleware)  # 2nd added = runs 2nd
app.add_middleware(                           # 1st added = runs 3rd (innermost)
    CORSMiddleware,
    allow_origins=["https://app.example.com"],  # NEVER use "*" in production
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

ASGI ミドルウェア（推奨）

パフォーマンスが重要な処理には、純粋な ASGI ミドルウェアを使用してください：

from starlette.types import ASGIApp, Receive, Scope, Send
import time

class TimingMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        start = time.perf_counter()
        await self.app(scope, receive, send)
        duration = time.perf_counter() - start
        # Log or record the duration

BaseHTTPMiddleware（より簡単ですが低速）

リクエストボディまたはレスポンスを読み取り／変更する必要があるミドルウェアにのみ使用してください：

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

class RequestIdMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
        request.state.request_id = request_id
        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id
        return response

使い分け：

• ASGI ミドルウェア： パフォーマンスが重要で、リクエスト／レスポンスのボディを読み取る必要がない場合
• BaseHTTPMiddleware： Request／Response オブジェクトへのアクセスが必要で、より簡単な API を使いたい場合

認証依存関係

JWT トークンの検証

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    session: AsyncSession = Depends(get_async_session),
) -> User:
    try:
        payload = jwt.decode(token, settings.jwt_secret, algorithms=["HS256"])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid token")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

    user = await session.get(User, user_id)
    if user is None or not user.is_active:
        raise HTTPException(status_code=401, detail="User not found or inactive")
    return user

ロールベースのアクセス制御（ファクトリーパターン）

def require_role(*roles: str):
    """Factory that creates a dependency requiring specific roles."""
    async def check_role(user: User = Depends(get_current_user)) -> User:
        if user.role not in roles:
            raise HTTPException(
                status_code=403,
                detail=f"Requires one of: {', '.join(roles)}",
            )
        return user
    return check_role

# Usage in routes
@router.delete("/users/{user_id}", dependencies=[Depends(require_role("admin"))])
async def delete_user(user_id: int, ...) -> None:
    ...

@router.patch("/posts/{post_id}")
async def update_post(
    post_id: int,
    user: User = Depends(require_role("admin", "editor")),
) -> PostResponse:
    ...

依存性注入チェーン

キャッシュの動作

FastAPI は単一のリクエスト内で依存関係の結果をキャッシュします。同じ依存関係が複数回呼び出されても、同じインスタンスが返されます：

# get_async_session is called once per request, even if used by multiple deps
async def get_user_service(session: AsyncSession = Depends(get_async_session)) -> UserService:
    return UserService(session)

async def get_post_service(session: AsyncSession = Depends(get_async_session)) -> PostService:
    return PostService(session)  # Same session instance as user_service

キャッシュを無効にする（呼び出すたびに新しいインスタンスを取得する）には、use_cache=False を使用します：

session: AsyncSession = Depends(get_async_session, use_cache=False)

Yield 依存関係（リソースのクリーンアップ）

async def get_http_client() -> AsyncGenerator[httpx.AsyncClient, None]:
    async with httpx.AsyncClient(timeout=30.0) as client:
        yield client
    # Client is automatically closed after the request

テストでの依存関係のオーバーライド

# In tests
from app.main import app
from app.dependencies.auth import get_current_user

async def mock_current_user() -> User:
    return User(id=1, email="[email protected]", role="admin")

app.dependency_overrides[get_current_user] = mock_current_user

バックグラウンドタスク

FastAPI BackgroundTasks（軽量）

サーバーの再起動後も存続する必要がないタスクに使用します：

from fastapi import BackgroundTasks

@router.post("/users", status_code=201)
async def create_user(
    data: UserCreate,
    background_tasks: BackgroundTasks,
    service: UserService = Depends(get_user_service),
) -> UserResponse:
    user = await service.create_user(data)
    background_tasks.add_task(send_welcome_email, user.email)
    return UserResponse.model_validate(user)

async def send_welcome_email(email: str) -> None:
    """Runs after the response is sent. Creates its own session."""
    async with async_session_factory() as session:
        async with session.begin():
            # Send email, log activity, etc.
            ...

ルール：

• バックグラウンドタスクではリクエストのセッションを再利用せず、新しいセッションを作成してください
• バックグラウンドタスクは同じプロセス内で実行され、再試行も永続化もありません
• 信頼性、再試行、または分散処理が必要なタスクには、Celery などを使用してください

WebSocket パターン

from fastapi import WebSocket, WebSocketDisconnect

class ConnectionManager:
    def __init__(self) -> None:
        self.active: dict[int, list[WebSocket]] = {}

    async def connect(self, user_id: int, ws: WebSocket) -> None:
        await ws.accept()
        self.active.setdefault(user_id, []).append(ws)

    def disconnect(self, user_id: int, ws: WebSocket) -> None:
        if user_id in self.active:
            self.active[user_id].remove(ws)
            if not self.active[user_id]:
                del self.active[user_id]

    async def send_to_user(self, user_id: int, message: dict) -> None:
        for ws in self.active.get(user_id, []):
            await ws.send_json(message)

manager = ConnectionManager()

@router.websocket("/ws")
async def websocket_endpoint(ws: WebSocket, token: str) -> None:
    # Auth via query parameter: /ws?token=xxx
    user = await verify_ws_token(token)
    if not user:
        await ws.close(code=4001)
        return

    await manager.connect(user.id, ws)
    try:
        while True:
            data = await ws.receive_json()
            # Process incoming messages
            await handle_message(user.id, data)
    except WebSocketDisconnect:
        manager.disconnect(user.id, ws)

WebSocket 認証の方法：

1. クエリパラメーター：ws://host/ws?token=xxx（最も簡単ですが、トークンがログに表示されます）
2. 最初のメッセージ：接続後、最初のメッセージとしてトークンを送信します（より安全です）
3. Cookie：既存のセッション Cookie を使用します（同じドメインである必要があります）

アプリケーションのライフサイクル

非推奨の on_event ではなく、lifespan コンテキストマネージャーを使用します：

from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # Startup: initialize resources
    await init_database()
    redis = await aioredis.from_url(settings.redis_url)
    app.state.redis = redis

    yield  # Application runs here

    # Shutdown: cleanup resources
    await redis.close()
    await dispose_engine()

app = FastAPI(lifespan=lifespan)

ライフサイクルで行う処理：

• データベース接続プールの初期化と破棄
• Redis／キャッシュ接続の設定と終了処理
• HTTP クライアントプールの作成
• バックグラウンドスケジューラーの起動／終了
• 起動時のキャッシュのウォームアップ

例外ハンドラー

一貫したエラーレスポンスを返すため、グローバル例外ハンドラーを登録します：

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

@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "detail": "Validation error",
            "code": "VALIDATION_ERROR",
            "field_errors": [
                {"field": e["loc"][-1], "message": e["msg"], "code": e["type"]}
                for e in exc.errors()
            ],
        },
    )

OpenAPI のカスタマイズ

app = FastAPI(
    title="My API",
    version="1.0.0",
    description="API description with **markdown** support",
    openapi_tags=[
        {"name": "Users", "description": "User management operations"},
        {"name": "Auth", "description": "Authentication endpoints"},
    ],
    docs_url="/docs",        # Swagger UI
    redoc_url="/redoc",      # ReDoc
    openapi_url="/openapi.json",
)

例

JWT 認証の依存関係チェーン

トークンから認証済みユーザーまでの認証チェーン全体：

Request with Authorization: Bearer <token>
    ↓
oauth2_scheme (extracts token from header)
    ↓
get_current_user (decodes JWT, loads user from DB)
    ↓
require_role("admin") (checks user.role)
    ↓
Route handler (receives verified admin user)

チェーン内の各依存関係は、dependency_overrides を使って個別にテストできます。

エッジケース

• ミドルウェアと依存関係： ロギング、タイミング、CORS などの横断的関心事にはミドルウェアを使用します。ルートごとのロジック（認証、ページネーションのパラメーター、機能フラグ）には依存関係を使用します。

• ASGI と BaseHTTPMiddleware： パフォーマンスを重視する場合は ASGI ミドルウェアを推奨します。BaseHTTPMiddleware はレスポンスボディ全体をメモリに読み込むため、ストリーミングや大きなレスポンスで問題が発生します。

• Lifespan と on_event： 常に lifespan コンテキストマネージャーを使用してください。@app.on_event("startup") と @app.on_event("shutdown") は FastAPI 0.109+ で非推奨になっています。

• サブアプリケーション間での Depends キャッシュ： 依存関係のキャッシュは、単一のアプリインスタンス内でリクエストごとに機能します。app.mount() を使用してサブアプリケーションをマウントする場合、各サブアプリはそれぞれ独自の依存関係解決スコープを持ちます。

• WebSocket のスケーリング： 1 つのサーバーインスタンスがすべての WebSocket 接続を保持します。複数インスタンスでデプロイする場合は、Redis pub/sub を使用してインスタンス間でメッセージをブロードキャストします。

ミドルウェアの実装例については、references/middleware-examples.md を参照してください。 高度な DI パターンについては、references/dependency-injection-patterns.md を参照してください。