Pythonの独自例外の定義

はじめに

標準的な例外処理は、Pythonに組み込まれた例外クラスを使用して一般的なエラー状況を処理します。しかし、実際のアプリケーション開発では、ビジネスロジックやドメイン固有のエラー条件を表現するために、独自の例外クラスを定義することが非常に重要です。

独自例外を適切に定義することで、プログラムにおけるエラー処理がより明確で効果的になります。まず、状況に応じた意味のあるエラーメッセージを提供できるようになり、問題の原因を理解しやすくなります。また、特定のエラー条件を明確に識別できるため、処理の分岐がわかりやすくなります。さらに、コード全体の可読性と保守性が向上し、開発者間で一貫したエラーハンドリングを実現できる点も大きな利点です。

基本的な独自例外の定義

最もシンプルな独自例外

Pythonで独自例外を定義する最も基本的な方法は、Exceptionクラスを継承することです。次の例は、validate_age()関数で年齢が0未満の場合に例外を発生させ、try-except構文でその例外を捕捉してエラーメッセージを表示します。

class MyCustomError(Exception):
    """独自例外の基本的な例"""
    pass

# 使用例
def validate_age(age):
    if age < 0:
        raise MyCustomError("年齢は0以上である必要があります")
    return age

try:
    validate_age(-5)
except MyCustomError as e:
    print(f"カスタムエラーが発生: {e}")

この例では、MyCustomErrorという独自例外を定義し、年齢が負の値の場合にこの例外を発生させています。

例外メッセージのカスタマイズ

より詳細な情報を伝えるために、例外クラスにメッセージを組み込む方法があります。次のコードは、入力値の検証エラーを扱うための独自例外ValidationErrorを定義しています。

class ValidationError(Exception):
    """バリデーションエラーを表す独自例外"""

    def __init__(self, field, value, message):
        self.field = field
        self.value = value
        self.message = message
        super().__init__(f"フィールド '{field}' の値 '{value}' が無効です: {message}")

# 使用例
def validate_email(email):
    if "@" not in email:
        raise ValidationError("email", email, "メールアドレスの形式が正しくありません")
    return True

try:
    validate_email("invalid-email")
except ValidationError as e:
    print(f"エラー詳細 - フィールド: {e.field}, 値: {e.value}, 理由: {e.message}")

例外にはフィールド名・値・エラーメッセージを保持し、無効な入力内容を詳細に伝えられます。validate_email()関数でメール形式を検証し、不正な場合に例外を発生させ、try-exceptで詳細情報を表示します。

階層化された例外の定義

現実のアプリケーションでは、関連する例外を階層化して定義することが推奨されます。次のコードは、銀行システムにおける独自例外処理の仕組みを示しています。

# ベースとなるカスタム例外
class BankingError(Exception):
    """銀行システムのベース例外"""
    pass

# 特定のエラー種類を表す例外
class InsufficientFundsError(BankingError):
    """残高不足エラー"""

    def __init__(self, current_balance, withdrawal_amount):
        self.current_balance = current_balance
        self.withdrawal_amount = withdrawal_amount
        super().__init__(
            f"残高不足: 現在の残高 {current_balance}円, "
            f"引出額 {withdrawal_amount}円"
        )

class AccountNotFoundError(BankingError):
    """口座不存在エラー"""

    def __init__(self, account_number):
        self.account_number = account_number
        super().__init__(f"口座番号 {account_number} は存在しません")

class InvalidAmountError(BankingError):
    """無効金額エラー"""

    def __init__(self, amount):
        self.amount = amount
        super().__init__(f"無効な金額: {amount}円")

# 使用例
class BankAccount:
    def __init__(self, account_number, balance=0):
        self.account_number = account_number
        self.balance = balance

    def withdraw(self, amount):
        if amount <= 0:
            raise InvalidAmountError(amount)
        if amount > self.balance:
            raise InsufficientFundsError(self.balance, amount)

        self.balance -= amount
        return self.balance

# テスト
account = BankAccount("123456", 10000)

try:
    account.withdraw(15000)
except InsufficientFundsError as e:
    print(f"残高不足エラー: {e}")
except InvalidAmountError as e:
    print(f"無効金額エラー: {e}")
except BankingError as e:
    print(f"その他の銀行エラー: {e}")

BankingErrorを基底クラスとして、InsufficientFundsError（残高不足）、AccountNotFoundError（口座不存在）、InvalidAmountError（無効金額）などの特定エラーを定義しています。BankAccountクラスでは引き出し処理中に条件に応じた例外を発生させ、try-except構文でエラー内容をわかりやすく処理します。

例外階層のメリット

例外を階層化することで、以下のようなメリットがあります。

1. 細かい制御: 特定の例外だけを捕捉できる
2. 一括処理: 親例外で関連する例外をまとめて処理できる
3. 拡張性: 新しい例外を簡単に追加できる
4. 一貫性: エラーハンドリングが統一される

高度な独自例外パターン

データ検証のための例外

データ検証システムでは、複数の検証エラーをまとめて報告する必要がある場合があります。FieldValidationErrorは個々のフィールドの不正を表し、MultipleValidationErrorsはそれらをまとめて保持します。UserValidatorクラスではユーザー情報のname、email、ageを検証し、複数の違反がある場合にまとめて例外を発生させます。これにより、入力データ全体の問題点を一度に把握できます。

class MultipleValidationErrors(Exception):
    """複数の検証エラーをまとめる例外"""

    def __init__(self, errors):
        self.errors = errors
        error_messages = [str(error) for error in errors]
        super().__init__("複数の検証エラー:\n" + "\n".join(error_messages))

class FieldValidationError(Exception):
    """個々のフィールド検証エラー"""

    def __init__(self, field, value, rule):
        self.field = field
        self.value = value
        self.rule = rule
        super().__init__(f"フィールド '{field}' がルール '{rule}' に違反: 値='{value}'")

class UserValidator:
    @staticmethod
    def validate(user_data):
        errors = []

        # 名前の検証
        if "name" not in user_data or not user_data["name"]:
            errors.append(FieldValidationError("name", user_data.get("name"), "必須項目"))

        # メールの検証
        email = user_data.get("email", "")
        if "@" not in email:
            errors.append(FieldValidationError("email", email, "有効なメール形式"))

        # 年齢の検証
        age = user_data.get("age")
        if age is not None and (age < 0 or age > 150):
            errors.append(FieldValidationError("age", age, "0-150の範囲"))

        if errors:
            raise MultipleValidationErrors(errors)

# 使用例
try:
    user_data = {"name": "", "email": "invalid", "age": 200}
    UserValidator.validate(user_data)
except MultipleValidationErrors as e:
    print("検証エラーが発生しました:")
    for error in e.errors:
        print(f"  - {error}")
except FieldValidationError as e:
    print(f"単一の検証エラー: {e}")

コンテキスト情報を含む例外

デバッグを容易にするために、豊富なコンテキスト情報を含む例外を定義できます。次のコードは、発生時の状況（コンテキスト情報）を含む独自例外ContextualErrorを定義した例です。例外にはエラーメッセージのほか、発生時刻・スタックトレース・関連情報（注文IDやユーザーIDなど）を記録できます。process_order()関数では商品がない注文に対してこの例外を発生させ、to_dict()メソッドで辞書形式の詳細情報を取得し、ログ出力やエラーレポートに活用できるようになっています。

import traceback
from datetime import datetime

class ContextualError(Exception):
    """コンテキスト情報を含む例外"""

    def __init__(self, message, context=None):
        self.message = message
        self.context = context or {}
        self.timestamp = datetime.now()
        self.stack_trace = traceback.format_stack()

        # 詳細なメッセージの構築
        detailed_message = f"{message}\n"
        detailed_message += f"発生時刻: {self.timestamp}\n"
        if context:
            detailed_message += "コンテキスト情報:\n"
            for key, value in context.items():
                detailed_message += f"  {key}: {value}\n"

        super().__init__(detailed_message)

    def to_dict(self):
        """例外情報を辞書形式で返す"""
        return {
            "message": self.message,
            "context": self.context,
            "timestamp": self.timestamp.isoformat(),
            "type": self.__class__.__name__
        }

# 使用例
def process_order(order_data):
    try:
        if not order_data.get("items"):
            context = {
                "order_id": order_data.get("id", "不明"),
                "user_id": order_data.get("user_id", "不明"),
                "operation": "order_processing"
            }
            raise ContextualError("注文に商品が含まれていません", context)

        # 処理を続行...
        return "処理成功"

    except ContextualError as e:
        # ロギングなどに利用
        error_info = e.to_dict()
        print("エラー情報:", error_info)
        raise

# テスト
order_data = {"id": "ORD123", "user_id": "USER456", "items": []}
try:
    process_order(order_data)
except ContextualError as e:
    print(f"コンテキスト付きエラー: {e}")

Web APIのエラーレスポンス

REST APIでは、標準化されたエラーレスポンスを返すことが重要です。基底クラスAPIErrorが共通のエラーフォーマット（メッセージ・ステータスコード・エラーコード）を管理し、to_response()でHTTPレスポンス形式の辞書を返します。派生クラスとして、NotFoundError（404）、ValidationError（400）、AuthenticationError（401）などを用意し、用途ごとに適切なレスポ

class APIError(Exception):
    """APIエラーのベースクラス"""

    def __init__(self, message, status_code=500, error_code=None):
        self.message = message
        self.status_code = status_code
        self.error_code = error_code
        super().__init__(self.message)

    def to_response(self):
        """HTTPレスポンス用の辞書を返す"""
        return {
            "error": {
                "code": self.error_code or self.__class__.__name__,
                "message": self.message,
                "status": self.status_code
            }
        }

class NotFoundError(APIError):
    """リソース不存在エラー"""
    def __init__(self, resource_type, resource_id):
        message = f"{resource_type} '{resource_id}' は見つかりません"
        super().__init__(message, status_code=404, error_code="NOT_FOUND")

class ValidationError(APIError):
    """バリデーションエラー"""
    def __init__(self, field_errors):
        self.field_errors = field_errors
        message = "入力データにエラーがあります"
        super().__init__(message, status_code=400, error_code="VALIDATION_ERROR")

    def to_response(self):
        response = super().to_response()
        response["error"]["field_errors"] = self.field_errors
        return response

class AuthenticationError(APIError):
    """認証エラー"""
    def __init__(self, message="認証に失敗しました"):
        super().__init__(message, status_code=401, error_code="UNAUTHORIZED")

# 使用例
def get_user(user_id):
    # ユーザーが存在しない場合
    if user_id not in user_database:
        raise NotFoundError("ユーザー", user_id)

    return user_database[user_id]

def handle_api_request(request):
    try:
        # 何らかの処理
        user = get_user(request.user_id)
        return {"data": user}

    except APIError as e:
        # クライアントに返すエラーレスポンス
        return e.to_response(), e.status_code

ドメイン駆動設計における例外

ドメイン駆動設計（DDD）では、ドメインの概念を例外として表現します。DomainExceptionを基底とし、金額が負の場合のInvalidMoneyAmountError、通貨不一致のCurrencyMismatchError、型不正のInvalidMoneyTypeError、残高不足のInsufficientFundsError（派生的に発生）など、業務上のルールを個別の例外で扱います。これにより、ドメインロジックの意図が明確化され、安全で一貫性のある処理が実現されています。

class DomainException(Exception):
    """ドメイン例外のベースクラス"""
    pass

class Money:
    def __init__(self, amount, currency="JPY"):
        if amount < 0:
            raise InvalidMoneyAmountError(amount)
        self.amount = amount
        self.currency = currency

    def __add__(self, other):
        if self.currency != other.currency:
            raise CurrencyMismatchError(self.currency, other.currency)
        return Money(self.amount + other.amount, self.currency)

class InvalidMoneyAmountError(DomainException):
    def __init__(self, amount):
        self.amount = amount
        super().__init__(f"金額は0以上である必要があります: {amount}")

class CurrencyMismatchError(DomainException):
    def __init__(self, currency1, currency2):
        self.currency1 = currency1
        self.currency2 = currency2
        super().__init__(f"通貨が一致しません: {currency1} vs {currency2}")

class BankAccount:
    def __init__(self, initial_balance=None):
        self.balance = initial_balance or Money(0)

    def deposit(self, amount):
        if not isinstance(amount, Money):
            raise InvalidMoneyTypeError(type(amount))
        self.balance = self.balance + amount

    def withdraw(self, amount):
        if not isinstance(amount, Money):
            raise InvalidMoneyTypeError(type(amount))

        try:
            new_balance = self.balance + Money(-amount.amount)  # 負の値を加算
        except InvalidMoneyAmountError:
            raise InsufficientFundsError(self.balance, amount)

        self.balance = new_balance

class InvalidMoneyTypeError(DomainException):
    def __init__(self, actual_type):
        super().__init__(f"Money型である必要があります: {actual_type}")

# 使用例
try:
    account = BankAccount(Money(1000))
    account.withdraw(Money(1500))  # 残高不足
except DomainException as e:
    print(f"ドメインエラー: {e}")

独自例外のベストプラクティス

意味のある名前の選択

例外クラス名は、その例外が何を表すかを明確に示すべきです。良い例では、InsufficientFundsErrorやUserNotFoundErrorのように、何が原因でどのようなエラーなのかが明確にわかる名前を付けています。一方、悪い例のError1やMyExceptionは、意味が曖昧で内容が推測できず、コードの可読性や保守性を損ないます。例外名には具体的な状況を反映することが重要です。

# 良い例
class InsufficientFundsError(Exception): pass
class UserNotFoundError(Exception): pass

# 悪い例
class Error1(Exception): pass
class MyException(Exception): pass

適切な継承階層の構築

関連する例外は共通の基底クラスから継承し、エラーハンドリングを簡素化します。

豊富なコンテキスト情報の提供

デバッグを容易にするために、関連する情報を例外に含めます。

ドキュメントの整備

独自例外の使用方法と発生条件をドキュメント化します。

まとめ

独自例外の定義は、Pythonで堅牢でメンテナンス性の高いアプリケーションを構築するための重要なスキルです。基本的な例外定義から、階層化、コンテキスト情報の付加、ドメイン固有の例外まで、段階的に理解を深めることで、実際のプロジェクトで効果的に活用できるようになります。

次の章では、デバッグ技法について学び、発生した例外や問題を効果的に調査・解決する方法を探求します。

---

演習問題

初級問題

問題1: 基本的な独自例外
NegativeNumberErrorという独自例外を定義し、数値が負の場合にこの例外を発生させる関数を作成してください。

問題2: 例外の継承
BankAccountErrorを基底クラスとして、InsufficientFundsErrorとInvalidAmountErrorという2つの具体的な例外クラスを定義してください。

問題3: シンプルなバリデーション
ユーザー名のバリデーションを行う関数を作成し、ユーザー名が空の場合にEmptyUsernameErrorを、短すぎる場合にShortUsernameErrorを発生させるようにしてください。

中級問題

問題4: コンテキスト情報付き例外
ファイル処理中に発生するエラーを表すFileProcessingErrorを定義し、ファイル名と操作タイプ（読み込み/書き込み）の情報を含めるようにしてください。

問題5: 複数エラーの収集
フォームバリデーションで複数のフィールドエラーを収集するFormValidationErrorを定義し、個々のフィールドエラーをまとめて報告できるようにしてください。

問題6: APIエラーの階層
REST API用のエラー階層を構築し、APIErrorを基底としてNotFoundError、BadRequestError、AuthenticationErrorを定義してください。

問題7: リトライ可能例外
一時的なエラーを表すRetryableErrorを定義し、最大リトライ回数とリトライ間隔の情報を持たせてください。

問題8: ドメイン例外
ECシステムで在庫切れを表すOutOfStockErrorを定義し、商品IDと在庫数を情報として含めてください。

問題9: 例外チェーン
低レベルの例外をラップして高レベルの例外を発生させる処理を実装し、例外チェーンを確認してください。

上級問題

問題10: マルチレイヤー例外システム
アプリケーション全体で使用する例外階層を設計し、インフラ層、アプリケーション層、ドメイン層それぞれで発生する例外を表現してください。

問題11: 国際化対応例外
多言語対応の例外クラスを定義し、ロケールに応じて適切な言語でエラーメッセージを提供できるようにしてください。

問題12: 監査証跡用例外
セキュリティ監査で使用するSecurityBreachErrorを定義し、ユーザーID、IPアドレス、操作内容などの監査証跡情報を含めて、自動的にログに記録されるようにしてください。

初級問題

問題1: 基本的な独自例外

class NegativeNumberError(Exception):
    """数値が負の場合の例外"""
    pass

def check_positive_number(number):
    """
    数値が正であることをチェックする関数
    """
    if number < 0:
        raise NegativeNumberError(f"数値は正である必要があります: {number}")
    return number

# テスト
print("=== 問題1テスト ===")
try:
    print(check_positive_number(10))  # 正常
    print(check_positive_number(-5))  # 例外発生
except NegativeNumberError as e:
    print(f"エラー: {e}")

問題2: 例外の継承

class BankAccountError(Exception):
    """銀行口座操作のベース例外"""
    pass

class InsufficientFundsError(BankAccountError):
    """残高不足エラー"""

    def __init__(self, balance, amount):
        self.balance = balance
        self.amount = amount
        super().__init__(f"残高不足: {balance}円に対して{amount}円の引出し")

class InvalidAmountError(BankAccountError):
    """無効金額エラー"""

    def __init__(self, amount):
        self.amount = amount
        super().__init__(f"無効な金額: {amount}円")

class BankAccount:
    def __init__(self, balance=0):
        self.balance = balance

    def withdraw(self, amount):
        if amount <= 0:
            raise InvalidAmountError(amount)
        if amount > self.balance:
            raise InsufficientFundsError(self.balance, amount)

        self.balance -= amount
        return self.balance

# テスト
print("\n=== 問題2テスト ===")
account = BankAccount(1000)

try:
    account.withdraw(1500)  # 残高不足
except InsufficientFundsError as e:
    print(f"残高不足エラー: {e}")
except InvalidAmountError as e:
    print(f"無効金額エラー: {e}")

try:
    account.withdraw(-100)  # 無効金額
except BankAccountError as e:
    print(f"銀行口座エラー: {e}")

問題3: シンプルなバリデーション

class UsernameError(Exception):
    """ユーザー名エラーのベースクラス"""
    pass

class EmptyUsernameError(UsernameError):
    """空のユーザー名エラー"""
    def __init__(self):
        super().__init__("ユーザー名は空にできません")

class ShortUsernameError(UsernameError):
    """短すぎるユーザー名エラー"""
    def __init__(self, username, min_length):
        self.username = username
        self.min_length = min_length
        super().__init__(f"ユーザー名は{min_length}文字以上必要です: '{username}' ({len(username)}文字)")

def validate_username(username, min_length=3):
    """
    ユーザー名を検証する関数
    """
    if not username:
        raise EmptyUsernameError()

    if len(username) < min_length:
        raise ShortUsernameError(username, min_length)

    return True

# テスト
print("\n=== 問題3テスト ===")
test_usernames = ["", "ab", "abc", "username"]

for username in test_usernames:
    try:
        validate_username(username)
        print(f"✓ '{username}' は有効です")
    except EmptyUsernameError as e:
        print(f"✗ 空のユーザー名: {e}")
    except ShortUsernameError as e:
        print(f"✗ 短いユーザー名: {e}")

中級問題

問題4: コンテキスト情報付き例外

class FileProcessingError(Exception):
    """ファイル処理エラー"""

    def __init__(self, message, filename, operation):
        self.filename = filename
        self.operation = operation
        self.message = message
        super().__init__(f"{operation}操作中にエラー: {filename} - {message}")

def read_file_safely(filename):
    """
    ファイルを安全に読み込む関数
    """
    try:
        with open(filename, 'r', encoding='utf-8') as file:
            return file.read()
    except FileNotFoundError:
        raise FileProcessingError("ファイルが見つかりません", filename, "読み込み")
    except PermissionError:
        raise FileProcessingError("アクセス権限がありません", filename, "読み込み")
    except UnicodeDecodeError:
        raise FileProcessingError("文字コードのデコードに失敗", filename, "読み込み")

def write_file_safely(filename, content):
    """
    ファイルを安全に書き込む関数
    """
    try:
        with open(filename, 'w', encoding='utf-8') as file:
            file.write(content)
    except PermissionError:
        raise FileProcessingError("アクセス権限がありません", filename, "書き込み")
    except IOError as e:
        raise FileProcessingError(f"入出力エラー: {e}", filename, "書き込み")

# テスト
print("\n=== 問題4テスト ===")
try:
    read_file_safely("nonexistent.txt")
except FileProcessingError as e:
    print(f"ファイル処理エラー: {e}")
    print(f"詳細 - ファイル: {e.filename}, 操作: {e.operation}")

# 正常なファイル操作
try:
    write_file_safely("test_output.txt", "Hello, World!")
    content = read_file_safely("test_output.txt")
    print(f"ファイル内容: {content}")
except FileProcessingError as e:
    print(f"ファイル処理エラー: {e}")

問題5: 複数エラーの収集

class FormValidationError(Exception):
    """フォームバリデーションエラー"""

    def __init__(self, field_errors):
        self.field_errors = field_errors
        error_count = len(field_errors)
        super().__init__(f"フォームに{error_count}個のエラーがあります")

class FieldError:
    """個々のフィールドエラー"""

    def __init__(self, field_name, value, error_message):
        self.field_name = field_name
        self.value = value
        self.error_message = error_message

    def __str__(self):
        return f"{self.field_name}: '{self.value}' - {self.error_message}"

def validate_registration_form(form_data):
    """
    登録フォームを検証する関数
    """
    errors = []

    # ユーザー名の検証
    username = form_data.get("username", "").strip()
    if not username:
        errors.append(FieldError("username", username, "必須項目です"))
    elif len(username) < 3:
        errors.append(FieldError("username", username, "3文字以上必要です"))

    # メールアドレスの検証
    email = form_data.get("email", "").strip()
    if not email:
        errors.append(FieldError("email", email, "必須項目です"))
    elif "@" not in email:
        errors.append(FieldError("email", email, "有効なメールアドレス形式で入力してください"))

    # パスワードの検証
    password = form_data.get("password", "")
    if len(password) < 8:
        errors.append(FieldError("password", "***", "8文字以上必要です"))

    if errors:
        raise FormValidationError(errors)

    return True

# テスト
print("\n=== 問題5テスト ===")
test_forms = [
    {"username": "", "email": "invalid", "password": "short"},
    {"username": "ab", "email": "test@example.com", "password": "password123"},
    {"username": "validuser", "email": "test@example.com", "password": "securepassword"}
]

for i, form_data in enumerate(test_forms, 1):
    print(f"\nフォーム {i}:")
    try:
        validate_registration_form(form_data)
        print("✓ 検証成功")
    except FormValidationError as e:
        print(f"✗ {e}")
        for field_error in e.field_errors:
            print(f"  - {field_error}")

問題6: APIエラーの階層

class APIError(Exception):
    """APIエラーのベースクラス"""

    def __init__(self, message, status_code):
        self.message = message
        self.status_code = status_code
        super().__init__(f"[{status_code}] {message}")

    def to_dict(self):
        """エラー情報を辞書形式で返す"""
        return {
            "error": {
                "type": self.__class__.__name__,
                "message": self.message,
                "status_code": self.status_code
            }
        }

class NotFoundError(APIError):
    """リソース不存在エラー"""

    def __init__(self, resource_type, resource_id):
        self.resource_type = resource_type
        self.resource_id = resource_id
        message = f"{resource_type} '{resource_id}' は見つかりません"
        super().__init__(message, 404)

class BadRequestError(APIError):
    """不正リクエストエラー"""

    def __init__(self, message="リクエストが不正です"):
        super().__init__(message, 400)

class AuthenticationError(APIError):
    """認証エラー"""

    def __init__(self, message="認証に失敗しました"):
        super().__init__(message, 401)

class UserService:
    def __init__(self):
        self.users = {
            "1": {"name": "Alice", "email": "alice@example.com"},
            "2": {"name": "Bob", "email": "bob@example.com"}
        }

    def get_user(self, user_id):
        if user_id not in self.users:
            raise NotFoundError("ユーザー", user_id)
        return self.users[user_id]

    def create_user(self, user_data):
        if not user_data.get("name") or not user_data.get("email"):
            raise BadRequestError("名前とメールアドレスは必須です")
        return user_data

# テスト
print("\n=== 問題6テスト ===")
user_service = UserService()

test_cases = [
    lambda: user_service.get_user("1"),      # 正常
    lambda: user_service.get_user("999"),    # 不存在
    lambda: user_service.create_user({}),    # 不正リクエスト
]

for i, test_case in enumerate(test_cases, 1):
    print(f"\nテストケース {i}:")
    try:
        result = test_case()
        print(f"✓ 成功: {result}")
    except NotFoundError as e:
        print(f"✗ 不存在エラー: {e}")
        print(f"  レスポンス: {e.to_dict()}")
    except BadRequestError as e:
        print(f"✗ 不正リクエスト: {e}")
        print(f"  レスポンス: {e.to_dict()}")
    except APIError as e:
        print(f"✗ APIエラー: {e}")
        print(f"  レスポンス: {e.to_dict()}")

問題7: リトライ可能例外

import time

class RetryableError(Exception):
    """リトライ可能な一時的エラー"""

    def __init__(self, message, max_retries=3, retry_delay=1):
        self.message = message
        self.max_retries = max_retries
        self.retry_delay = retry_delay
        super().__init__(f"{message} (最大リトライ: {max_retries}, 待機: {retry_delay}秒)")

class NetworkError(RetryableError):
    """ネットワークエラー"""
    pass

class DatabaseConnectionError(RetryableError):
    """データベース接続エラー"""
    pass

def retry_on_error(func, *args, **kwargs):
    """
    リトライ可能なエラー発生時にリトライするデコレータ的関数
    """
    max_retries = kwargs.pop('max_retries', 3)
    retry_delay = kwargs.pop('retry_delay', 1)

    for attempt in range(max_retries + 1):
        try:
            return func(*args, **kwargs)
        except RetryableError as e:
            if attempt == max_retries:
                print(f"最大リトライ回数に達しました: {e}")
                raise

            print(f"リトライ可能エラー: {e}. {attempt + 1}/{max_retries}回目をリトライ...")
            time.sleep(retry_delay)

    raise Exception("予期しないエラー")

def unstable_network_operation():
    """
    不安定なネットワーク操作を模擬
    """
    import random
    if random.random() < 0.7:  # 70%の確率で失敗
        raise NetworkError("ネットワーク接続が不安定です", max_retries=3, retry_delay=0.5)
    return "ネットワーク操作成功"

# テスト
print("\n=== 問題7テスト ===")
try:
    result = retry_on_error(unstable_network_operation, max_retries=3, retry_delay=0.5)
    print(f"結果: {result}")
except RetryableError as e:
    print(f"最終的に失敗: {e}")

問題8: ドメイン例外

class InventoryError(Exception):
    """在庫管理エラーのベースクラス"""
    pass

class OutOfStockError(InventoryError):
    """在庫切れエラー"""

    def __init__(self, product_id, product_name, current_stock, requested_quantity):
        self.product_id = product_id
        self.product_name = product_name
        self.current_stock = current_stock
        self.requested_quantity = requested_quantity

        message = (f"商品 '{product_name}' (ID: {product_id}) の在庫が不足しています: "
                  f"現在 {current_stock}個, 要求 {requested_quantity}個")
        super().__init__(message)

class ProductNotFoundError(InventoryError):
    """商品不存在エラー"""

    def __init__(self, product_id):
        self.product_id = product_id
        super().__init__(f"商品ID '{product_id}' は存在しません")

class InventorySystem:
    def __init__(self):
        self.products = {
            "P001": {"name": "ノートパソコン", "stock": 5, "price": 120000},
            "P002": {"name": "マウス", "stock": 20, "price": 2500},
            "P003": {"name": "キーボード", "stock": 0, "price": 5000}
        }

    def check_stock(self, product_id, quantity=1):
        if product_id not in self.products:
            raise ProductNotFoundError(product_id)

        product = self.products[product_id]
        if product["stock"] < quantity:
            raise OutOfStockError(
                product_id, 
                product["name"], 
                product["stock"], 
                quantity
            )

        return True

    def purchase(self, product_id, quantity=1):
        self.check_stock(product_id, quantity)

        # 在庫を減らす
        self.products[product_id]["stock"] -= quantity
        product = self.products[product_id]
        total_price = product["price"] * quantity

        return {
            "product_name": product["name"],
            "quantity": quantity,
            "unit_price": product["price"],
            "total_price": total_price
        }

# テスト
print("\n=== 問題8テスト ===")
inventory = InventorySystem()

test_purchases = [
    ("P001", 2),   # 正常
    ("P002", 25),  # 在庫不足
    ("P003", 1),   # 在庫切れ
    ("P999", 1),   # 商品不存在
]

for product_id, quantity in test_purchases:
    print(f"\n商品 {product_id} を {quantity}個 購入:")
    try:
        result = inventory.purchase(product_id, quantity)
        print(f"✓ 購入成功: {result}")
    except OutOfStockError as e:
        print(f"✗ 在庫切れ: {e}")
    except ProductNotFoundError as e:
        print(f"✗ 商品不存在: {e}")
    except InventoryError as e:
        print(f"✗ 在庫エラー: {e}")

問題9: 例外チェーン

class LowLevelError(Exception):
    """低レベルエラー"""
    pass

class HighLevelError(Exception):
    """高レベルエラー"""

    def __init__(self, message, original_error=None):
        self.original_error = original_error
        if original_error:
            message = f"{message} [原因: {original_error}]"
        super().__init__(message)

def low_level_operation():
    """
    低レベル操作（ファイルIO、データベース接続など）
    """
    # 何らかの低レベルエラーを模擬
    raise LowLevelError("データベース接続に失敗しました")

def high_level_operation():
    """
    高レベル操作（ビジネスロジック）
    """
    try:
        low_level_operation()
    except LowLevelError as e:
        # 低レベルエラーを高レベルエラーでラップ
        raise HighLevelError("ユーザーデータの取得に失敗しました", original_error=e) from e

def business_process():
    """
    ビジネスプロセス（ユーザー向けインターフェース）
    """
    try:
        high_level_operation()
    except HighLevelError as e:
        print(f"ビジネスエラー: {e}")
        print(f"根本原因: {e.original_error}")
        print(f"例外チェーン:")
        current_error = e
        while current_error.__cause__:
            print(f"  - {type(current_error.__cause__).__name__}: {current_error.__cause__}")
            current_error = current_error.__cause__

# テスト
print("\n=== 問題9テスト ===")
business_process()

# 例外チェーンの詳細な確認
print("\n--- 例外チェーンの詳細 ---")
try:
    high_level_operation()
except HighLevelError as e:
    print(f"高レベルエラー: {e}")
    print(f"原因例外: {e.__cause__}")
    print(f"コンテキスト: {e.__context__}")

上級問題

問題10: マルチレイヤー例外システム

# ベース例外
class ApplicationException(Exception):
    """アプリケーション全体のベース例外"""

    def __init__(self, message, layer):
        self.message = message
        self.layer = layer
        super().__init__(f"[{layer}] {message}")

# インフラ層例外
class InfrastructureException(ApplicationException):
    """インフラ層例外のベース"""

    def __init__(self, message):
        super().__init__(message, "Infrastructure")

class DatabaseConnectionException(InfrastructureException):
    """データベース接続例外"""
    pass

class ExternalServiceException(InfrastructureException):
    """外部サービス例外"""
    pass

# アプリケーション層例外
class ApplicationServiceException(ApplicationException):
    """アプリケーションサービス層例外のベース"""

    def __init__(self, message):
        super().__init__(message, "Application")

class UserServiceException(ApplicationServiceException):
    """ユーザーサービス例外"""
    pass

class OrderServiceException(ApplicationServiceException):
    """注文サービス例外"""
    pass

# ドメイン層例外
class DomainException(ApplicationException):
    """ドメイン層例外のベース"""

    def __init__(self, message):
        super().__init__(message, "Domain")

class InvalidEntityStateException(DomainException):
    """不正なエンティティ状態例外"""
    pass

class BusinessRuleViolationException(DomainException):
    """ビジネスルール違反例外"""
    pass

# 使用例
class UserRepository:
    """インフラ層: ユーザーレポジトリ"""

    def find_by_id(self, user_id):
        try:
            # データベース操作を模擬
            if user_id == "error":
                raise DatabaseConnectionException("データベース接続がタイムアウトしました")
            return {"id": user_id, "name": f"User {user_id}"}
        except DatabaseConnectionException as e:
            raise

class UserService:
    """アプリケーション層: ユーザーサービス"""

    def __init__(self):
        self.user_repository = UserRepository()

    def get_user_profile(self, user_id):
        try:
            user = self.user_repository.find_by_id(user_id)
            if not user:
                raise UserServiceException(f"ユーザー {user_id} は存在しません")
            return user
        except InfrastructureException as e:
            # インフラ層例外をアプリケーション層例外でラップ
            raise UserServiceException(f"ユーザー情報の取得に失敗: {e}") from e

class User:
    """ドメイン層: ユーザーエンティティ"""

    def __init__(self, user_id, name, email):
        self.user_id = user_id
        self.name = name
        self.email = email
        self._validate()

    def _validate(self):
        if not self.user_id:
            raise InvalidEntityStateException("ユーザーIDは必須です")
        if not self.name:
            raise InvalidEntityStateException("ユーザー名は必須です")
        if "@" not in self.email:
            raise BusinessRuleViolationException("有効なメールアドレス形式ではありません")

# テスト
print("\n=== 問題10テスト ===")
user_service = UserService()

test_cases = [
    "123",     # 正常
    "error",   # インフラ層エラー
    "",        # ドメイン層エラー
]

for user_id in test_cases:
    print(f"\nユーザーID '{user_id}' の処理:")
    try:
        if user_id:
            result = user_service.get_user_profile(user_id)
            print(f"✓ 成功: {result}")
        else:
            user = User("", "", "invalid")  # ドメイン層エラー
    except ApplicationServiceException as e:
        print(f"✗ アプリケーション層エラー: {e}")
        if e.__cause__:
            print(f"  原因: {e.__cause__}")
    except DomainException as e:
        print(f"✗ ドメイン層エラー: {e}")
    except ApplicationException as e:
        print(f"✗ アプリケーションエラー: {e}")

問題11: 国際化対応例外

class InternationalizedError(Exception):
    """国際化対応例外のベースクラス"""

    # エラーメッセージのテンプレート（多言語対応）
    MESSAGE_TEMPLATES = {
        'en': {
            'required': "Field '{field}' is required",
            'invalid_email': "Invalid email format: {value}",
            'min_length': "Field '{field}' must be at least {min_length} characters"
        },
        'ja': {
            'required': "フィールド '{field}' は必須です",
            'invalid_email': "無効なメール形式: {value}",
            'min_length': "フィールド '{field}' は{min_length}文字以上必要です"
        },
        'es': {
            'required': "El campo '{field}' es obligatorio",
            'invalid_email': "Formato de email inválido: {value}",
            'min_length': "El campo '{field}' debe tener al menos {min_length} caracteres"
        }
    }

    def __init__(self, error_key, locale='en', **format_args):
        self.error_key = error_key
        self.locale = locale
        self.format_args = format_args

        # ロケールに応じたメッセージを取得
        message = self._get_localized_message(error_key, locale, format_args)
        super().__init__(message)

    def _get_localized_message(self, error_key, locale, format_args):
        """ロケールに応じたエラーメッセージを取得"""
        if locale not in self.MESSAGE_TEMPLATES:
            locale = 'en'  # フォールバック

        templates = self.MESSAGE_TEMPLATES[locale]
        if error_key not in templates:
            return f"Unknown error: {error_key}"

        return templates[error_key].format(**format_args)

    def with_locale(self, locale):
        """別のロケールで同じ例外を作成"""
        return self.__class__(self.error_key, locale, **self.format_args)

class ValidationErrorI18n(InternationalizedError):
    """国際化対応バリデーションエラー"""
    pass

def validate_user_data_i18n(user_data, locale='en'):
    """
    多言語対応のユーザーデータ検証
    """
    errors = []

    # 名前の検証
    if not user_data.get('name', '').strip():
        errors.append(ValidationErrorI18n('required', locale, field='name'))

    # メールの検証
    email = user_data.get('email', '')
    if email and '@' not in email:
        errors.append(ValidationErrorI18n('invalid_email', locale, value=email))

    # パスワードの検証
    password = user_data.get('password', '')
    if password and len(password) < 8:
        errors.append(ValidationErrorI18n('min_length', locale, field='password', min_length=8))

    if errors:
        # 複数エラーがある場合は最初のエラーのみ表示（実際はまとめて処理する）
        raise errors[0]

    return True

# テスト
print("\n=== 問題11テスト ===")
test_data = {'name': '', 'email': 'invalid', 'password': 'short'}

locales = ['en', 'ja', 'es']
for locale in locales:
    print(f"\nロケール: {locale}")
    try:
        validate_user_data_i18n(test_data, locale)
        print("✓ 検証成功")
    except ValidationErrorI18n as e:
        print(f"✗ {e}")

        # 他のロケールでのメッセージも表示
        for other_locale in [l for l in locales if l != locale]:
            other_error = e.with_locale(other_locale)
            print(f"  [{other_locale}] {other_error}")

問題12: 監査証跡用例外

import logging
import datetime
from dataclasses import dataclass
from typing import Optional

@dataclass
class AuditContext:
    """監査証跡のコンテキスト情報"""
    user_id: str
    ip_address: str
    action: str
    resource: str
    timestamp: datetime.datetime
    additional_info: Optional[dict] = None

class SecurityBreachError(Exception):
    """セキュリティ侵害エラー"""

    def __init__(self, message, severity, audit_context):
        self.message = message
        self.severity = severity  # 'low', 'medium', 'high', 'critical'
        self.audit_context = audit_context
        self.timestamp = datetime.datetime.now()

        # 自動的に監査ログに記録
        self._log_to_audit_trail()

        super().__init__(self._format_message())

    def _format_message(self):
        """フォーマットされたエラーメッセージを作成"""
        ctx = self.audit_context
        return (f"セキュリティ侵害 detected: {self.message} "
                f"[ユーザー: {ctx.user_id}, アクション: {ctx.action}, "
                f"重要度: {self.severity}]")

    def _log_to_audit_trail(self):
        """監査証跡に記録"""
        logger = logging.getLogger('security_audit')

        log_data = {
            'timestamp': self.timestamp.isoformat(),
            'exception_type': self.__class__.__name__,
            'message': self.message,
            'severity': self.severity,
            'user_id': self.audit_context.user_id,
            'ip_address': self.audit_context.ip_address,
            'action': self.audit_context.action,
            'resource': self.audit_context.resource,
            'additional_info': self.audit_context.additional_info
        }

        # 重要度に応じたログレベル
        log_levels = {
            'low': logging.INFO,
            'medium': logging.WARNING,
            'high': logging.ERROR,
            'critical': logging.CRITICAL
        }

        level = log_levels.get(self.severity, logging.ERROR)
        logger.log(level, f"Security breach: {log_data}")

    def to_audit_dict(self):
        """監査用の辞書形式で返す"""
        return {
            'exception': {
                'type': self.__class__.__name__,
                'message': self.message,
                'severity': self.severity,
                'timestamp': self.timestamp.isoformat()
            },
            'context': {
                'user_id': self.audit_context.user_id,
                'ip_address': self.audit_context.ip_address,
                'action': self.audit_context.action,
                'resource': self.audit_context.resource,
                'timestamp': self.audit_context.timestamp.isoformat(),
                'additional_info': self.audit_context.additional_info
            }
        }

# 監査ログの設定
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('security_audit.log', encoding='utf-8'),
        logging.StreamHandler()
    ]
)

class SecuritySystem:
    def __init__(self):
        self.authorized_roles = {'admin', 'manager', 'user'}
        self.sensitive_resources = {'user_data', 'financial_records', 'system_config'}

    def check_access(self, user_id, user_role, ip_address, resource, action):
        """
        アクセス権限をチェック
        """
        audit_context = AuditContext(
            user_id=user_id,
            ip_address=ip_address,
            action=action,
            resource=resource,
            timestamp=datetime.datetime.now()
        )

        # ロールの検証
        if user_role not in self.authorized_roles:
            context = AuditContext(
                user_id=user_id,
                ip_address=ip_address,
                action=action,
                resource=resource,
                timestamp=datetime.datetime.now(),
                additional_info={'attempted_role': user_role}
            )
            raise SecurityBreachError(
                "未承認のロールでのアクセス試行",
                "high",
                context
            )

        # センシティブリソースへのアクセス制御
        if (resource in self.sensitive_resources and 
            user_role not in {'admin', 'manager'} and
            action == 'write'):
            context = AuditContext(
                user_id=user_id,
                ip_address=ip_address,
                action=action,
                resource=resource,
                timestamp=datetime.datetime.now(),
                additional_info={'user_role': user_role, 'sensitive_resource': resource}
            )
            raise SecurityBreachError(
                "権限のないセンシティブリソースへの書き込み試行",
                "critical",
                context
            )

        # IPアドレスの検証（簡易版）
        if ip_address.startswith('10.0.0.'):
            context = AuditContext(
                user_id=user_id,
                ip_address=ip_address,
                action=action,
                resource=resource,
                timestamp=datetime.datetime.now(),
                additional_info={'internal_network_access': True}
            )
            raise SecurityBreachError(
                "内部ネットワークからの不正アクセス試行",
                "medium",
                context
            )

        return True

# テスト
print("\n=== 問題12テスト ===")
security_system = SecuritySystem()

test_cases = [
    # (user_id, role, ip, resource, action)
    ("user123", "admin", "192.168.1.100", "user_data", "read"),  # 正常
    ("user456", "guest", "192.168.1.101", "public_data", "read"),  # 未承認ロール
    ("user789", "user", "192.168.1.102", "financial_records", "write"),  # 権限不足
    ("user999", "admin", "10.0.0.100", "system_config", "read"),  # 内部ネットワーク
]

for user_id, role, ip, resource, action in test_cases:
    print(f"\nアクセスチェック: {user_id}({role}) from {ip} -> {resource}.{action}")
    try:
        security_system.check_access(user_id, role, ip, resource, action)
        print("✓ アクセス許可")
    except SecurityBreachError as e:
        print(f"✗ セキュリティ侵害: {e}")
        audit_info = e.to_audit_dict()
        print(f"  監査情報: {audit_info}")