error-handling

エラー処理

Let it crash の哲学に基づく、Python API 向けの本番環境対応エラー処理です。

設計思想

Let it crash — 防御的なコードは書かず、例外を自然に伝播させ、境界で処理します。

# BAD - Too defensive, obscures errors
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    try:
        user = await user_service.get(user_id)
        if not user:
            raise HTTPException(404, "Not found")
        return user
    except DatabaseError as e:
        raise HTTPException(500, "Database error")
    except Exception as e:
        logger.exception("Unexpected error")
        raise HTTPException(500, "Internal error")

# GOOD - Let exceptions propagate, handle at boundary
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await user_service.get(user_id)
    if not user:
        raise UserNotFoundError(user_id)
    return user

基本原則

1. 下位で送出し、上位で捕捉する — エラーが発生した箇所で例外を送出し、API の境界で処理します
2. ドメイン例外 — 汎用的な例外ではなく、意味を表す例外を作成します
3. グローバルハンドラー — @app.exception_handler() を使用して、エラー形式を一元的に整えます
4. 裸の except は使わない — 必ず具体的な例外を捕捉します
5. コンテキストを保持する — raise ... from error を使って、元のトレースバックを保持します

クイックスタート

1. ドメイン例外を定義する

from enum import StrEnum

class ErrorCode(StrEnum):
    USER_NOT_FOUND = "user_not_found"
    INVALID_CREDENTIALS = "invalid_credentials"
    RATE_LIMITED = "rate_limited"

class DomainError(Exception):
    """Base exception for all domain errors."""
    def __init__(self, code: ErrorCode, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code
        super().__init__(message)

class UserNotFoundError(DomainError):
    def __init__(self, user_id: int):
        super().__init__(
            code=ErrorCode.USER_NOT_FOUND,
            message=f"User {user_id} not found",
            status_code=404
        )

2. エラーレスポンスのスキーマを定義する

from pydantic import BaseModel

class ErrorDetail(BaseModel):
    code: str
    message: str
    request_id: str | None = None

class ErrorResponse(BaseModel):
    error: ErrorDetail

3. グローバルハンドラーを登録する

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

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

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": {"code": "http_error", "message": str(exc.detail)}}
    )

@app.exception_handler(RequestValidationError)
async def validation_error_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={"error": {"code": "validation_error", "message": "Invalid request"}}
    )

@app.exception_handler(Exception)
async def generic_error_handler(request: Request, exc: Exception):
    # Log full error internally
    logger.exception("Unhandled error")
    # Return safe message to client
    return JSONResponse(
        status_code=500,
        content={"error": {"code": "internal_error", "message": "Internal server error"}}
    )

4. ルートで使用する

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await user_service.get(user_id)
    if not user:
        raise UserNotFoundError(user_id)
    return user

例外を捕捉するタイミング

次の場合に限り、例外を捕捉します。

状況	例
再試行が必要な場合	一時的な失敗に対する tenacity.retry()
変換が必要な場合	サードパーティ SDK のエラーをドメインエラーとしてラップする
クリーンアップが必要な場合	finally またはコンテキストマネージャーを使用する
コンテキストの追加が必要な場合	raise DomainError(...) from original

Python + FastAPI の統合

レイヤー	責務
サービス／ドメイン	ドメイン例外を送出する（UserNotFoundError）
ルート	例外を伝播させる（try/except は使わない）
例外ハンドラー	HTTP レスポンスに変換する
ミドルウェア	リクエストコンテキスト（request_id、タイミング）を追加する

よくあるパターン

サードパーティ SDK のラップ

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

class ExternalServiceError(DomainError):
    def __init__(self, service: str, original: Exception):
        super().__init__(
            code=ErrorCode.EXTERNAL_SERVICE_ERROR,
            message=f"{service} unavailable",
            status_code=503
        )
        self.__cause__ = original

@retry(stop=stop_after_attempt(3), wait=wait_exponential())
async def call_payment_api(data: dict):
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.post("https://api.payment.com/charge", json=data)
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as e:
        raise ExternalServiceError("Payment API", e) from e

バックグラウンドタスクのエラー処理

from fastapi import BackgroundTasks

async def safe_background_task(task_func, *args, **kwargs):
    try:
        await task_func(*args, **kwargs)
    except Exception as e:
        logger.exception(f"Background task failed: {e}")
        # Optional: send to dead letter queue or alerting

@app.post("/orders")
async def create_order(order: Order, background_tasks: BackgroundTasks):
    result = await order_service.create(order)
    background_tasks.add_task(safe_background_task, send_confirmation_email, result.id)
    return result

トラブルシューティング

問題	原因	修正方法
レスポンスにスタックトレースが含まれる	汎用ハンドラーがない	@app.exception_handler(Exception) を追加する
元のエラーが失われる	from がない	raise NewError() from original を使用する
バリデーションエラーが漏れる	デフォルトのハンドラー	RequestValidationError ハンドラーをオーバーライドする
失敗が黙って見過ごされる	例外が握りつぶされている	例外を伝播させ、境界で処理する

参考資料

• Python パターン — 例外の設計、捕捉するタイミング、SDK のラップ
• FastAPI パターン — HTTPException、グローバルハンドラー、ミドルウェア
• Pydantic パターン — ValidationError、バリデーターでの送出
• Asyncio パターン — TaskGroup、タイムアウト、バックグラウンドタスク
• FastAPI ドキュメント：エラーの処理
• Pydantic ドキュメント：エラー処理