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

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

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

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

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

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

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

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

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

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

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

from fastapi import FastAPI

app = FastAPI()

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

import httpx

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

from typing import Annotated
from fastapi import Depends

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

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

from functools import lru_cache

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

SettingsDep = Annotated[Settings, Depends(get_settings)]

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

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

DbDep = Annotated[AsyncSession, Depends(get_db)]

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

bearer = HTTPBearer()

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

CurrentUser = Annotated[UserPublic, Depends(get_current_user)]


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

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

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

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

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

app = FastAPI(lifespan=lifespan)

from fastapi import HTTPException, status

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

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

logger = logging.getLogger("app")

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

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

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

import uuid
from starlette.middleware.base import BaseHTTPMiddleware

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

app.add_middleware(RequestIDMiddleware)

import json, logging, time

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

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

from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

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

from fastapi import BackgroundTasks

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

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

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

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

client = TestClient(app)

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

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

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

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

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

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

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

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

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

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

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