python-error-handling

Python のエラーハンドリング

適切な入力検証、意味のある例外、そして適切な失敗処理によって、堅牢な Python アプリケーションを構築します。優れたエラーハンドリングにより、デバッグが容易になり、システムの信頼性が高まります。

このスキルを使う場面

• ユーザー入力と API パラメーターの検証
• アプリケーションの例外階層の設計
• バッチ処理における部分的な失敗への対応
• 外部データからドメイン型への変換
• ユーザーフレンドリーなエラーメッセージの作成
• フェイルファスト検証パターンの実装

基本概念

1. 早期失敗

コストのかかる処理を行う前に、入力を早期に検証します。可能であれば、すべての検証エラーをまとめて報告します。

2. 意味のある例外

状況に応じた例外型を使用し、コンテキストを含めます。メッセージには、何が失敗したのか、なぜ失敗したのか、どう修正すればよいのかを説明します。

3. 部分的な失敗

バッチ処理では、1 件の失敗ですべてを中断させないようにします。成功と失敗を別々に記録します。

4. コンテキストの保持

例外を連鎖させ、デバッグに必要なエラーの全経路を保持します。

クイックスタート

def fetch_page(url: str, page_size: int) -> Page:
    if not url:
        raise ValueError("'url' is required")
    if not 1 <= page_size <= 100:
        raise ValueError(f"'page_size' must be 1-100, got {page_size}")
    # Now safe to proceed...

基本パターン

パターン 1：早期の入力検証

処理を開始する前に、API の境界ですべての入力を検証します。

def process_order(
    order_id: str,
    quantity: int,
    discount_percent: float,
) -> OrderResult:
    """Process an order with validation."""
    # Validate required fields
    if not order_id:
        raise ValueError("'order_id' is required")

    # Validate ranges
    if quantity <= 0:
        raise ValueError(f"'quantity' must be positive, got {quantity}")

    if not 0 <= discount_percent <= 100:
        raise ValueError(
            f"'discount_percent' must be 0-100, got {discount_percent}"
        )

    # Validation passed, proceed with processing
    return _process_validated_order(order_id, quantity, discount_percent)

パターン 2：早期のドメイン型への変換

システムの境界で、文字列や外部データを型付きのドメインオブジェクトにパースします。

from enum import Enum

class OutputFormat(Enum):
    JSON = "json"
    CSV = "csv"
    PARQUET = "parquet"

def parse_output_format(value: str) -> OutputFormat:
    """Parse string to OutputFormat enum.

    Args:
        value: Format string from user input.

    Returns:
        Validated OutputFormat enum member.

    Raises:
        ValueError: If format is not recognized.
    """
    try:
        return OutputFormat(value.lower())
    except ValueError:
        valid_formats = [f.value for f in OutputFormat]
        raise ValueError(
            f"Invalid format '{value}'. "
            f"Valid options: {', '.join(valid_formats)}"
        )

# Usage at API boundary
def export_data(data: list[dict], format_str: str) -> bytes:
    output_format = parse_output_format(format_str)  # Fail fast
    # Rest of function uses typed OutputFormat
    ...

パターン 3：複雑な検証には Pydantic を使用

構造化された入力検証とエラーメッセージの自動生成には、Pydantic モデルを使用します。

from pydantic import BaseModel, Field, field_validator

class CreateUserInput(BaseModel):
    """Input model for user creation."""

    email: str = Field(..., min_length=5, max_length=255)
    name: str = Field(..., min_length=1, max_length=100)
    age: int = Field(ge=0, le=150)

    @field_validator("email")
    @classmethod
    def validate_email_format(cls, v: str) -> str:
        if "@" not in v or "." not in v.split("@")[-1]:
            raise ValueError("Invalid email format")
        return v.lower()

    @field_validator("name")
    @classmethod
    def normalize_name(cls, v: str) -> str:
        return v.strip().title()

# Usage
try:
    user_input = CreateUserInput(
        email="[email protected]",
        name="john doe",
        age=25,
    )
except ValidationError as e:
    # Pydantic provides detailed error information
    print(e.errors())

パターン 4：エラーを標準例外に対応付ける

必要に応じてコンテキストを追加しながら、Python の組み込み例外型を適切に使用します。

失敗の種類	例外	例
無効な入力	ValueError	パラメーター値が不正
型の不一致	TypeError	文字列を期待していたが、int が渡された
項目が見つからない	KeyError	Dict のキーが見つからない
操作上の失敗	RuntimeError	サービスが利用できない
タイムアウト	TimeoutError	操作に時間がかかりすぎた
ファイルが見つからない	FileNotFoundError	パスが存在しない
アクセス拒否	PermissionError	アクセスが禁止されている

# Good: Specific exception with context
raise ValueError(f"'page_size' must be 1-100, got {page_size}")

# Avoid: Generic exception, no context
raise Exception("Invalid parameter")

高度なパターン

パターン 5：コンテキストを持つカスタム例外

構造化された情報を保持する、ドメイン固有の例外を作成します。

class ApiError(Exception):
    """Base exception for API errors."""

    def __init__(
        self,
        message: str,
        status_code: int,
        response_body: str | None = None,
    ) -> None:
        self.status_code = status_code
        self.response_body = response_body
        super().__init__(message)

class RateLimitError(ApiError):
    """Raised when rate limit is exceeded."""

    def __init__(self, retry_after: int) -> None:
        self.retry_after = retry_after
        super().__init__(
            f"Rate limit exceeded. Retry after {retry_after}s",
            status_code=429,
        )

# Usage
def handle_response(response: Response) -> dict:
    match response.status_code:
        case 200:
            return response.json()
        case 401:
            raise ApiError("Invalid credentials", 401)
        case 404:
            raise ApiError(f"Resource not found: {response.url}", 404)
        case 429:
            retry_after = int(response.headers.get("Retry-After", 60))
            raise RateLimitError(retry_after)
        case code if 400 <= code < 500:
            raise ApiError(f"Client error: {response.text}", code)
        case code if code >= 500:
            raise ApiError(f"Server error: {response.text}", code)

パターン 6：例外の連鎖

再送出する際に元の例外を保持し、デバッグの手がかりを維持します。

import httpx

class ServiceError(Exception):
    """High-level service operation failed."""
    pass

def upload_file(path: str) -> str:
    """Upload file and return URL."""
    try:
        with open(path, "rb") as f:
            response = httpx.post("https://upload.example.com", files={"file": f})
            response.raise_for_status()
            return response.json()["url"]
    except FileNotFoundError as e:
        raise ServiceError(f"Upload failed: file not found at '{path}'") from e
    except httpx.HTTPStatusError as e:
        raise ServiceError(
            f"Upload failed: server returned {e.response.status_code}"
        ) from e
    except httpx.RequestError as e:
        raise ServiceError(f"Upload failed: network error") from e

パターン 7：部分的な失敗を許容するバッチ処理

問題のある項目 1 件でバッチ全体を中断させないようにします。項目ごとに結果を記録します。

from dataclasses import dataclass

@dataclass
class BatchResult[T]:
    """Results from batch processing."""

    succeeded: dict[int, T]  # index -> result
    failed: dict[int, Exception]  # index -> error

    @property
    def success_count(self) -> int:
        return len(self.succeeded)

    @property
    def failure_count(self) -> int:
        return len(self.failed)

    @property
    def all_succeeded(self) -> bool:
        return len(self.failed) == 0

def process_batch(items: list[Item]) -> BatchResult[ProcessedItem]:
    """Process items, capturing individual failures.

    Args:
        items: Items to process.

    Returns:
        BatchResult with succeeded and failed items by index.
    """
    succeeded: dict[int, ProcessedItem] = {}
    failed: dict[int, Exception] = {}

    for idx, item in enumerate(items):
        try:
            result = process_single_item(item)
            succeeded[idx] = result
        except Exception as e:
            failed[idx] = e

    return BatchResult(succeeded=succeeded, failed=failed)

# Caller handles partial results
result = process_batch(items)
if not result.all_succeeded:
    logger.warning(
        f"Batch completed with {result.failure_count} failures",
        failed_indices=list(result.failed.keys()),
    )

パターン 8：長時間の処理における進捗報告

ビジネスロジックを UI に依存させずに、バッチ処理の進捗を確認できるようにします。

from collections.abc import Callable

ProgressCallback = Callable[[int, int, str], None]  # current, total, status

def process_large_batch(
    items: list[Item],
    on_progress: ProgressCallback | None = None,
) -> BatchResult:
    """Process batch with optional progress reporting.

    Args:
        items: Items to process.
        on_progress: Optional callback receiving (current, total, status).
    """
    total = len(items)
    succeeded = {}
    failed = {}

    for idx, item in enumerate(items):
        if on_progress:
            on_progress(idx, total, f"Processing {item.id}")

        try:
            succeeded[idx] = process_single_item(item)
        except Exception as e:
            failed[idx] = e

    if on_progress:
        on_progress(total, total, "Complete")

    return BatchResult(succeeded=succeeded, failed=failed)

ベストプラクティスのまとめ

1. 早期に検証する - コストのかかる処理の前に入力を確認します
2. 具体的な例外を使用する - 汎用的な Exception ではなく、ValueError や TypeError を使用します
3. コンテキストを含める - メッセージには、何が起きたのか、なぜ起きたのか、どう修正すればよいのかを説明します
4. 境界で型を変換する - 文字列を早期に enum やドメイン型にパースします
5. 例外を連鎖させる - raise ... from e を使用し、デバッグ情報を保持します
6. 部分的な失敗に対応する - 1 件のエラーでバッチを中断させないようにします
7. Pydantic を使用する - 構造化されたエラーを伴う複雑な入力検証に使用します
8. 失敗モードを文書化する - Docstring に発生し得る例外を記載します
9. コンテキストを付けてログに記録する - ID、件数、その他のデバッグ情報を含めます
10. エラー経路をテストする - 例外が正しく発生することを確認します

---

注意点

• raise X from None は原因チェーンを抑制しますが、raise X は __context__ を保持します — スタックトレースが異なります。下位の例外がノイズになる場合は、from None を使用してください。
• except 内の裸の raise は元の例外を再送出しますが、raise e は新しいトレースバックで再送出します — デバッグ時のヒント：元の発生箇所を保持するのは裸の raise だけです。
• python -O では assert が取り除かれます — 実行時の検証には必ず明示的な if + raise を使用し、assert は使用しないでください。
• 例外グループ（3.11 以降）には except ではなく except* が必要です — except ExceptionGroup はグループを捕捉しますが、展開しません。except* はサブ例外を選択的に処理します。
• finally は return の後でも実行されます。また、finally に独自の return があると戻り値を上書きします — finally から誤って戻り値を返すと、例外が握りつぶされます。