Python try exceptで業務エラー処理を完全マスター｜実務パターン解説

Pythonを使った業務開発では、エラーハンドリングが非常に重要です。APIの呼び出し失敗、データベース接続エラー、ファイル読み込み失敗など、本番環境ではあらゆる予期しない状況が発生します。単に「エラーをキャッチする」だけではなく、適切にログを記録し、ユーザーに通知し、必要に応じてリトライするなど、業務レベルのエラーハンドリングが求められます。

この記事では、Pythonの try-except-finally 構文を使った実務的なエラー処理パターンを、具体的なコード例を交えて解説します。

簡易的な解説：try-except-finally の基本
業務でのユースケース
実装コード：実務で使えるパターン集
よくある応用パターン
1. パターン5：カスタム例外クラス
2. パターン6：複数の操作を順序実行し、一つのエラーで全体を失敗させる
注意点：業務で避けるべきパターン
まとめ

簡易的な解説：try-except-finally の基本

Pythonのエラーハンドリングの基本は try-except-finally です。

try：エラーが発生する可能性のあるコードを記述
except：特定の例外を捕捉して処理
finally：エラーの有無に関わらず必ず実行される処理
else：エラーが発生しなかった場合だけ実行

基本的な書き方は以下の通りです。

try:
    # エラーが発生する可能性のあるコード
    result = 10 / 0
except ZeroDivisionError as e:
    # 特定の例外をキャッチして処理
    print(f\"エラーが発生しました: {e}\")
except Exception as e:
    # その他の例外をキャッチ
    print(f\"予期しないエラー: {e}\")
finally:
    # 必ず実行される処理
    print(\"処理完了\")\n

業務でのユースケース

実務開発では、以下のようなシーンで try-except が活躍します。

1. 外部API呼び出しのエラーハンドリング

APIサーバーがダウンしている、タイムアウト、レート制限など、様々なエラーが考えられます。

2. データベース操作のエラーハンドリング

接続エラー、トランザクション失敗、制約違反など、DBMSから返されるエラーに対応する必要があります。

3. ファイルI/Oのエラーハンドリング

ファイルが見つからない、パーミッション不足、ディスク容量不足など、ファイル操作固有のエラーが発生します。

4. データ処理のバリデーション

CSVやJSONの形式が不正、必須項目が欠落しているなど、データの異常を検出する必要があります。

実装コード：実務で使えるパターン集

パターン1：ログ出力を含むAPI呼び出し

業務では、エラーが発生した時点をログに記録することが必須です。以下は、外部APIを呼び出す時の実装例です。

import requests\nimport logging\nfrom typing import Optional, Dict, Any\n\nlogger = logging.getLogger(__name__)\n\ndef fetch_user_data(user_id: str) -> Optional[Dict[str, Any]]:\n    \"\"\"外部APIからユーザーデータを取得\"\"\"\n    url = f\"https://api.example.com/users/{user_id}\"\n    \n    try:\n        logger.info(f\"ユーザーデータ取得開始: user_id={user_id}\")\n        response = requests.get(url, timeout=5)\n        response.raise_for_status()  # ステータスコード4xx, 5xxでエラー発生\n        \n        data = response.json()\n        logger.info(f\"ユーザーデータ取得成功: user_id={user_id}\")\n        return data\n        \n    except requests.exceptions.Timeout:\n        logger.error(f\"タイムアウト: user_id={user_id}\")\n        return None\n        \n    except requests.exceptions.ConnectionError:\n        logger.error(f\"接続エラー: APIサーバーに接続できません\")\n        return None\n        \n    except requests.exceptions.HTTPError as e:\n        logger.error(f\"HTTPエラー: status_code={e.response.status_code}, user_id={user_id}\")\n        return None\n        \n    except ValueError as e:\n        logger.error(f\"JSON解析エラー: {e}\")\n        return None\n        \n    except Exception as e:\n        logger.exception(f\"予期しないエラーが発生: {e}\")\n        return None\n\n

パターン2：リトライロジック付きのAPI呼び出し

ネットワークエラーやサーバー一時的なダウンに対応するため、リトライは業務では必須パターンです。

import time\nfrom functools import wraps\n\ndef retry_on_exception(max_retries: int = 3, wait_seconds: int = 2):\n    \"\"\"指定回数までリトライするデコレータ\"\"\"\n    def decorator(func):\n        @wraps(func)\n        def wrapper(*args, **kwargs):\n            last_exception = None\n            \n            for attempt in range(1, max_retries + 1):\n                try:\n                    logger.info(f\"{func.__name__} 試行 {attempt}/{max_retries}\")\n                    return func(*args, **kwargs)\n                    \n                except (requests.exceptions.Timeout, \n                        requests.exceptions.ConnectionError) as e:\n                    last_exception = e\n                    \n                    if attempt < max_retries:\n                        logger.warning(\n                            f\"{func.__name__} 失敗（試行{attempt}）、\"\n                            f\"{wait_seconds}秒後にリトライします: {e}\"\n                        )\n                        time.sleep(wait_seconds)\n                    else:\n                        logger.error(\n                            f\"{func.__name__} {max_retries}回の試行後も失敗\"\n                        )\n                        \n                except requests.exceptions.HTTPError as e:\n                    # HTTPエラー（4xx, 5xx）はリトライしない\n                    logger.error(f\"HTTPエラー（リトライしない）: {e}\")\n                    raise\n            \n            raise last_exception\n        \n        return wrapper\n    return decorator\n\n@retry_on_exception(max_retries=3, wait_seconds=2)\ndef fetch_user_data_with_retry(user_id: str) -> Dict[str, Any]:\n    \"\"\"リトライ機能付きのユーザーデータ取得\"\"\"\n    response = requests.get(\n        f\"https://api.example.com/users/{user_id}\",\n        timeout=5\n    )\n    response.raise_for_status()\n    return response.json()\n\n

パターン3：データベース操作のトランザクション処理

DBの更新操作では、複数のクエリを一つのトランザクション内で実行し、エラーが発生したら全ロールバックするパターンが一般的です。

import psycopg2\nfrom contextlib import contextmanager\nfrom typing import Generator\n\nclass DatabaseConnection:\n    def __init__(self, connection_string: str):\n        self.connection_string = connection_string\n        self.conn = None\n    \n    @contextmanager\n    def transaction(self) -> Generator:\n        \"\"\"トランザクション処理を安全に実行するコンテキストマネージャー\"\"\"\n        try:\n            self.conn = psycopg2.connect(self.connection_string)\n            logger.info(\"データベース接続成功\")\n            yield self.conn\n            self.conn.commit()\n            logger.info(\"トランザクションコミット\")\n            \n        except psycopg2.DatabaseError as e:\n            if self.conn:\n                self.conn.rollback()\n            logger.error(f\"データベースエラーが発生、ロールバック: {e}\")\n            raise\n            \n        except psycopg2.IntegrityError as e:\n            if self.conn:\n                self.conn.rollback()\n            logger.error(f\"制約違反エラー、ロールバック: {e}\")\n            raise\n            \n        except Exception as e:\n            if self.conn:\n                self.conn.rollback()\n            logger.exception(f\"予期しないエラーが発生、ロールバック: {e}\")\n            raise\n            \n        finally:\n            if self.conn:\n                self.conn.close()\n                logger.info(\"データベース接続クローズ\")\n\n# 使用例\ndb = DatabaseConnection(\"postgresql://user:password@localhost/dbname\")\n\ntry:\n    with db.transaction() as conn:\n        cursor = conn.cursor()\n        # 複数のINSERT/UPDATE操作\n        cursor.execute(\n            \"INSERT INTO users (name, email) VALUES (%s, %s)\",\n            (\"山田太郎\", \"yamada@example.com\")\n        )\n        cursor.execute(\n            \"UPDATE user_stats SET total_users = total_users + 1\"\n        )\n        logger.info(\"データベース更新完了\")\n        \nexcept psycopg2.Error as e:\n    logger.error(f\"データベース操作失敗: {e}\")\n    # 上位で適切に処理（ユーザーへの通知など）\n\n

パターン4：ファイル操作とバリデーション

CSVファイルを読み込む場合、ファイルが存在しない、形式が不正、必須列が欠落しているなど、複数の層でエラーハンドリングが必要です。

import csv\nfrom pathlib import Path\nfrom dataclasses import dataclass\nfrom typing import List\n\n@dataclass\nclass UserRecord:\n    name: str\n    email: str\n    phone: str\n\ndef load_users_from_csv(file_path: str) -> List[UserRecord]:\n    \"\"\"CSVファイルからユーザーデータを読み込む\"\"\"\n    users = []\n    file_p = Path(file_path)\n    \n    try:\n        # ファイルの存在確認\n        if not file_p.exists():\n            raise FileNotFoundError(f\"ファイルが見つかりません: {file_path}\")\n        \n        logger.info(f\"ファイル読み込み開始: {file_path}\")\n        \n        with open(file_p, 'r', encoding='utf-8') as f:\n            reader = csv.DictReader(f)\n            \n            # ヘッダー検証\n            required_columns = {'name', 'email', 'phone'}\n            if not required_columns.issubset(set(reader.fieldnames or [])):\n                missing = required_columns - set(reader.fieldnames or [])\n                raise ValueError(f\"必須列が欠落しています: {missing}\")\n            \n            # 各行を処理\n            for row_num, row in enumerate(reader, start=2):  # start=2はヘッダー行をスキップ\n                try:\n                    # データの検証\n                    if not row.get('name') or not row.get('name').strip():\n                        raise ValueError(f\"行{row_num}: nameが空です\")\n                    \n                    if not row.get('email') or '@' not in row.get('email', ''):\n                        raise ValueError(f\"行{row_num}: emailの形式が不正です\")\n                    \n                    if not row.get('phone') or not row.get('phone').isdigit():\n                        raise ValueError(f\"行{row_num}: phoneは数字である必要があります\")\n                    \n                    user = UserRecord(\n                        name=row['name'].strip(),\n                        email=row['email'].strip(),\n                        phone=row['phone'].strip()\n                    )\n                    users.append(user)\n                    \n                except ValueError as e:\n                    logger.warning(f\"行のスキップ: {e}\")\n                    # 1行のエラーで全体を失敗させず、スキップして続行\n                    continue\n        \n        logger.info(f\"ファイル読み込み完了: {len(users)}件のユーザーを読み込み\")\n        return users\n        \n    except FileNotFoundError as e:\n        logger.error(f\"ファイルエラー: {e}\")\n        raise\n        \n    except ValueError as e:\n        logger.error(f\"ヘッダーエラー: {e}\")\n        raise\n        \n    except UnicodeDecodeError:\n        logger.error(\"ファイルのエンコーディングがUTF-8ではありません\")\n        raise\n        \n    except Exception as e:\n        logger.exception(f\"予期しないエラーが発生: {e}\")\n        raise\n\n

よくある応用パターン

パターン5：カスタム例外クラス

業務では、エラーの種類を細かく分類し、呼び出し側で適切に処理したいケースが多いです。カスタム例外クラスを定義することで、エラーハンドリングがより明確になります。

class ApplicationError(Exception):\n    \"\"\"アプリケーション固有の基本例外\"\"\"\n    pass\n\nclass ValidationError(ApplicationError):\n    \"\"\"バリデーションエラー\"\"\"\n    def __init__(self, field: str, message: str):\n        self.field = field\n        self.message = message\n        super().__init__(f\"Validation Error [{field}]: {message}\")\n\nclass ExternalAPIError(ApplicationError):\n    \"\"\"外部API呼び出し時のエラー\"\"\"\n    def __init__(self, api_name: str, status_code: int, message: str):\n        self.api_name = api_name\n        self.status_code = status_code\n        super().__init__(f\"API Error [{api_name}] {status_code}: {message}\")\n\nclass DatabaseError(ApplicationError):\n    \"\"\"データベース操作時のエラー\"\"\"\n    pass\n\ndef validate_email(email: str) -> str:\n    \"\"\"メールアドレスのバリデーション\"\"\"\n    if not email or '@' not in email:\n        raise ValidationError('email', 'メールアドレスの形式が不正です')\n    return email\n\ndef create_user(name: str, email: str) -> dict:\n    \"\"\"ユーザー作成処理\"\"\"\n    try:\n        # バリデーション\n        if not name or len(name) < 2:\n            raise ValidationError('name', '名前は2文字以上である必要があります')\n        \n        validated_email = validate_email(email)\n        \n        # DB保存（省略）\n        logger.info(f\"ユーザー作成成功: {name}\")\n        return {\"id\": 1, \"name\": name, \"email\": validated_email}\n        \n    except ValidationError as e:\n        # バリデーションエラーはログレベルはWARNING\n        logger.warning(f\"バリデーション失敗: {e}\")\n        raise\n        \n    except DatabaseError as e:\n        logger.error(f\"データベース保存失敗: {e}\")\n        raise\n        \n    except Exception as e:\n        logger.exception(f\"予期しないエラー: {e}\")\n        raise ApplicationError(f\"ユーザー作成処理失敗: {e}\")\n\n

パターン6：複数の操作を順序実行し、一つのエラーで全体を失敗させる

複数のAPI呼び出しやDB操作を順序実行する場合、どれか一つが失敗したら後続の処理をスキップするパターンです。

def process_order(order_id: str) -> bool:\n    \"\"\"注文処理（複数ステップ）\"\"\"\n    try:\n        # ステップ1: 在庫確認\n        logger.info(f\"ステップ1: 在庫確認開始 (order_id={order_id})\")\n        if not check_inventory(order_id):\n            raise ApplicationError(\"在庫が不足しています\")\n        logger.info(\"ステップ1: 在庫確認完了\")\n        \n        # ステップ2: 決済処理\n        logger.info(\"ステップ2: 決済処理開始\")\n        payment_result = process_payment(order_id)\n        if not payment_result:\n            raise ExternalAPIError(\"PaymentAPI\", 500, \"決済処理に失敗しました\")\n        logger.info(\"ステップ2: 決済処理完了\")\n        \n        # ステップ3: 配送手配\n        logger.info(\"ステップ3: 配送手配開始\")\n        shipping_result = arrange_shipping(order_id)\n        if not shipping_result:\n            # 決済は完了しているため、ロールバック処理が必要\n            try:\n                refund_payment(order_id)\n                logger.warning(\"決済をキャンセルしました\")\n            except Exception as e:\n                logger.error(f\"返金処理失敗: {e}\")\n                raise\n            raise ApplicationError(\"配送手配に失敗しました\")\n        logger.info(\"ステップ3: 配送手配完了\")\n        \n        logger.info(f\"注文処理完了: {order_id}\")\n        return True\n        \n    except ApplicationError as e:\n        logger.error(f\"注文処理失敗: {e}\")\n        return False\n        \n    except Exception as e:\n        logger.exception(f\"予期しないエラー: {e}\")\n        return False\n\n

注意点：業務で避けるべきパターン

1. 単純に例外を無視する

以下のコードは絶対に避けるべきです。

# ❌ 悪い例\ntry:\n fetch_user_data(user_id)\nexcept:\n pass # エラーを完全に無視\n\n\nこれでは、何が起きているのか全く分からず、デバッグが困難になります。

2. 汎用的すぎる例外処理

以下のように Exception でキャッチするのは、最後の砦として使用すべきです。

# ⚠️ 避けるべき\ntry:\n    response = requests.get(url)\nexcept Exception as e:\n    logger.error(f\"エラー: {e}\")\n\n# ✅ 良い例\ntry:\n    response = requests.get(url)\nexcept requests.exceptions.Timeout:\n    logger.error(\"タイムアウト\")\nexcept requests.exceptions.HTTPError as e:\n    logger.error(f\"HTTPエラー: {e.response.status_code}\")\nexcept Exception as e:\n    logger.exception(f\"予期しないエラー: {e}\")\n\n

3. ログ記録なしでエラーをキャッチ

業務では必ずログを記録してください。

# ❌ 悪い例\ntry:\n    result = some_operation()\nexcept Exception as e:\n    return None  # エラーログなし\n\n# ✅ 良い例\ntry:\n    result = some_operation()\nexcept SpecificError as e:\n    logger.error(f\"エラーが発生: {e}\")\n    return None\n\n

4. リソースのクリーンアップを忘れる

ファイルやDB接続は、必ずクローズする必要があります。

# ❌ 悪い例\ntry:\n    f = open('file.txt')\n    data = f.read()\nexcept Exception as e:\n    logger.error(e)\n# ファイルが閉じられない可能性\n\n# ✅ 良い例\ntry:\n    with open('file.txt') as f:\n        data = f.read()\nexcept Exception as e:\n    logger.error(e)\n# with文で自動的にクローズされる\n\n