Python例外処理のパターン実装ガイド：業務で使えるtry-except活用法
\n\n
1. 例外処理の基本概念
\n
Pythonで堅牢なコードを書く上で、例外処理（Exception Handling）は必須の技術です。try-except文を使うことで、エラーが発生した際にプログラムの強制終了を防ぎ、適切な対応を取ることができます。
\n\n
実務では、APIのタイムアウト、データベース接続エラー、ファイルの読み込み失敗など、予測不可能なエラーが頻繁に発生します。こうした状況で単にエラーメッセージを表示するだけでなく、ログを記録し、リトライを実行し、ユーザーに分かりやすいメッセージを返すといった対応が求められます。
\n\n
基本的なtry-except文の構文は次の通りです：
\n\n
try:\n # エラーが発生する可能性のあるコード\n risky_operation()\nexcept SpecificError:\n # 特定のエラーに対する処理\n handle_error()\nexcept Exception as e:\n # その他のエラーをキャッチ\n log_error(e)\nfinally:\n # エラーの有無に関わらず実行される処理\n cleanup()\n\n\n
2. 業務で頻出するユースケース
\n\n
2.1 外部APIとの通信
\n
実務では外部APIとの通信がエラーの温床になります。タイムアウト、ステータスコードエラー、不正なレスポンスフォーマットなど、様々な問題が発生する可能性があります。
\n\n
2.2 データベース操作
\n
データベースのコネクション切断、クエリのシンタックスエラー、デッドロックなど、DB関連のエラーは本番環境で避けられません。
\n\n
2.3 ファイル入出力
\n
ファイルが存在しない、読み取り権限がない、ディスクが満杯といったシステムレベルのエラーに対応する必要があります。
\n\n
3. 実装コード：業務パターン別解説
\n\n
3.1 外部API通信の例外処理
\n
リトライロジックを含めたAPI通信の例外処理は、実務では必須の実装パターンです。タイムアウトや一時的なエラーに対してリトライを行い、回数制限を設けることで無限ループを防ぎます。
\n\n
import requests\nfrom requests.exceptions import RequestException, Timeout, ConnectionError\nimport logging\nfrom time import sleep\nfrom typing import Optional, Dict, Any\n\nlogger = logging.getLogger(__name__)\n\ndef call_external_api(\n url: str,\n params: Optional[Dict[str, Any]] = None,\n max_retries: int = 3,\n timeout: int = 10\n) -> Optional[Dict[str, Any]]:\n \"\"\"\n 外部APIを呼び出し、リトライロジックを含む\n \n Args:\n url: API エンドポイント\n params: クエリパラメータ\n max_retries: 最大リトライ回数\n timeout: タイムアウト秒数\n \n Returns:\n JSONレスポンスまたはNone\n \"\"\"\n \n for attempt in range(max_retries):\n try:\n response = requests.get(\n url,\n params=params,\n timeout=timeout\n )\n \n # ステータスコードの確認\n if response.status_code == 200:\n logger.info(f\"API call successful: {url}\")\n return response.json()\n \n elif response.status_code in [429, 503, 504]:\n # リトライ可能なエラー\n logger.warning(\n f\"Retryable error {response.status_code}. \"\n f\"Attempt {attempt + 1}/{max_retries}\"\n )\n if attempt < max_retries - 1:\n wait_time = 2 ** attempt # 指数バックオフ\n sleep(wait_time)\n continue\n else:\n raise Exception(\n f\"Max retries exceeded. Status: {response.status_code}\"\n )\n \n else:\n # リトライ不可のエラー\n logger.error(\n f\"Non-retryable error {response.status_code}: \"\n f\"{response.text}\"\n )\n raise Exception(\n f\"API returned {response.status_code}: {response.text}\"\n )\n \n except Timeout:\n logger.warning(\n f\"Timeout occurred. Attempt {attempt + 1}/{max_retries}\"\n )\n if attempt < max_retries - 1:\n sleep(2 ** attempt)\n continue\n else:\n logger.error(\"Max retries exceeded for timeout\")\n return None\n \n except ConnectionError as e:\n logger.warning(\n f\"Connection error: {str(e)}. \"\n f\"Attempt {attempt + 1}/{max_retries}\"\n )\n if attempt < max_retries - 1:\n sleep(2 ** attempt)\n continue\n else:\n logger.error(\"Failed to connect after max retries\")\n return None\n \n except ValueError as e:\n # JSONパースエラー\n logger.error(f\"Invalid JSON response: {str(e)}\")\n return None\n \n except RequestException as e:\n # その他のrequestsライブラリのエラー\n logger.error(f\"Request exception: {str(e)}\")\n return None\n \n except Exception as e:\n # 予期しないエラー\n logger.error(f\"Unexpected error: {str(e)}\", exc_info=True)\n return None\n \n logger.error(f\"All retries exhausted for {url}\")\n return None\n\n# 使用例\nif __name__ == \"__main__\":\n logging.basicConfig(\n level=logging.INFO,\n format='%(asctime)s - %(levelname)s - %(message)s'\n )\n \n result = call_external_api(\n \"https://api.example.com/data\",\n params={\"key\": \"value\"},\n max_retries=3,\n timeout=10\n )\n \n if result:\n print(f\"Success: {result}\")\n else:\n print(\"Failed to retrieve data\")\n\n\n
3.2 データベース操作の例外処理
\n
データベース操作では、接続エラー、クエリエラー、トランザクション管理が重要です。実務ではコネクションプーリングとエラーハンドリングを組み合わせることが多いです。
\n\n
import sqlite3\nfrom contextlib import contextmanager\nimport logging\nfrom typing import Optional, List, Dict, Any\n\nlogger = logging.getLogger(__name__)\n\nclass DatabaseManager:\n def __init__(self, db_path: str):\n self.db_path = db_path\n \n @contextmanager\n def get_connection(self):\n \"\"\"コネクション取得とクローズを自動化\"\"\"\n conn = None\n try:\n conn = sqlite3.connect(self.db_path, timeout=5.0)\n conn.row_factory = sqlite3.Row\n yield conn\n except sqlite3.OperationalError as e:\n logger.error(f\"Database connection failed: {str(e)}\")\n raise\n finally:\n if conn:\n conn.close()\n \n def execute_query(\n self,\n query: str,\n params: tuple = (),\n is_transaction: bool = False\n ) -> Optional[List[Dict[str, Any]]]:\n \"\"\"\n SELECTクエリを実行\n \n Args:\n query: SQL文\n params: バインドパラメータ\n is_transaction: トランザクション用か\n \n Returns:\n クエリ結果またはNone\n \"\"\"\n try:\n with self.get_connection() as conn:\n cursor = conn.cursor()\n cursor.execute(query, params)\n \n results = cursor.fetchall()\n return [dict(row) for row in results]\n \n except sqlite3.ProgrammingError as e:\n logger.error(f\"SQL syntax error: {str(e)}\")\n logger.error(f\"Query: {query}\")\n return None\n \n except sqlite3.IntegrityError as e:\n logger.error(f\"Data integrity error: {str(e)}\")\n return None\n \n except sqlite3.DatabaseError as e:\n logger.error(f\"Database error: {str(e)}\")\n return None\n \n except Exception as e:\n logger.error(f\"Unexpected error in query: {str(e)}\", exc_info=True)\n return None\n \n def execute_insert(\n self,\n table: str,\n data: Dict[str, Any]\n ) -> bool:\n \"\"\"\n INSERTを実行\n \n Args:\n table: テーブル名\n data: 挿入データ\n \n Returns:\n 成功時True\n \"\"\"\n try:\n columns = ', '.join(data.keys())\n placeholders = ', '.join(['?' for _ in data])\n query = f\"INSERT INTO {table} ({columns}) VALUES ({placeholders})\"\n \n with self.get_connection() as conn:\n cursor = conn.cursor()\n cursor.execute(query, tuple(data.values()))\n conn.commit()\n logger.info(f\"Inserted into {table}: {data}\")\n return True\n \n except sqlite3.IntegrityError as e:\n logger.error(f\"Integrity constraint failed: {str(e)}\")\n return False\n \n except sqlite3.ProgrammingError as e:\n logger.error(f\"Invalid insert query: {str(e)}\")\n return False\n \n except Exception as e:\n logger.error(f\"Insert operation failed: {str(e)}\", exc_info=True)\n return False\n \n def execute_bulk_insert(\n self,\n table: str,\n data_list: List[Dict[str, Any]]\n ) -> bool:\n \"\"\"\n 複数行のINSERTをトランザクションで実行\n \n Args:\n table: テーブル名\n data_list: 挿入データのリスト\n \n Returns:\n 成功時True\n \"\"\"\n if not data_list:\n return True\n \n try:\n columns = ', '.join(data_list[0].keys())\n placeholders = ', '.join(['?' for _ in data_list[0]])\n query = f\"INSERT INTO {table} ({columns}) VALUES ({placeholders})\"\n \n with self.get_connection() as conn:\n cursor = conn.cursor()\n \n try:\n cursor.execute(\"BEGIN TRANSACTION\")\n \n for data in data_list:\n cursor.execute(query, tuple(data.values()))\n \n conn.commit()\n logger.info(f\"Bulk inserted {len(data_list)} rows into {table}\")\n return True\n \n except Exception as e:\n conn.rollback()\n logger.error(f\"Bulk insert failed, rolled back: {str(e)}\")\n raise\n \n except Exception as e:\n logger.error(f\"Bulk insert operation failed: {str(e)}\", exc_info=True)\n return False\n\n# 使用例\nif __name__ == \"__main__\":\n logging.basicConfig(\n level=logging.INFO,\n format='%(asctime)s - %(levelname)s - %(message)s'\n )\n \n db = DatabaseManager(\":memory:\")\n \n # テーブル作成\n try:\n with db.get_connection() as conn:\n conn.execute(\"\"\"\n CREATE TABLE users (\n id INTEGER PRIMARY KEY,\n name TEXT NOT NULL,\n email TEXT UNIQUE NOT NULL\n )\n \"\"\")\n conn.commit()\n except Exception as e:\n logger.error(f\"Failed to create table: {str(e)}\")\n \n # 単一行挿入\n success = db.execute_insert(\n \"users\",\n {\"name\": \"John Doe\", \"email\": \"john@example.com\"}\n )\n print(f\"Insert result: {success}\")\n \n # 複数行挿入\n bulk_data = [\n {\"name\": \"Jane Smith\", \"email\": \"jane@example.com\"},\n {\"name\": \"Bob Johnson\", \"email\": \"bob@example.com\"},\n ]\n success = db.execute_bulk_insert(\"users\", bulk_data)\n print(f\"Bulk insert result: {success}\")\n \n # クエリ実行\n results = db.execute_query(\"SELECT * FROM users\")\n if results:\n for row in results:\n print(row)\n\n\n
3.3 ファイル処理の例外処理
\n
ファイル操作は、エンコーディングエラーや権限問題、ディスク容量不足など、様々なエラーが発生します。with文を使った自動クローズと適切なエラーハンドリングが重要です。
\n\n
import os\nimport json\nfrom pathlib import Path\nimport logging\nfrom typing import Optional, Dict, Any, List\nimport csv\n\nlogger = logging.getLogger(__name__)\n\nclass FileProcessor:\n def __init__(self, base_directory: str):\n self.base_directory = Path(base_directory)\n \n def read_text_file(\n self,\n filename: str,\n encoding: str = 'utf-8'\n ) -> Optional[str]:\n \"\"\"\n テキストファイルを読み込む\n \n Args:\n filename: ファイル名\n encoding: 文字コード\n \n Returns:\n ファイル内容またはNone\n \"\"\"\n filepath = self.base_directory / filename\n \n try:\n with open(filepath, 'r', encoding=encoding) as f:\n content = f.read()\n logger.info(f\"Read file: {filepath}\")\n return content\n \n except FileNotFoundError:\n logger.error(f\"File not found: {filepath}\")\n return None\n \n except PermissionError:\n logger.error(f\"Permission denied: {filepath}\")\n return None\n \n except UnicodeDecodeError as e:\n logger.error(\n f\"Encoding error in {filepath} (tried {encoding}): {str(e)}\"\n )\n # フォールバック: 別のエンコーディングを試す\n try:\n with open(filepath, 'r', encoding='utf-8-sig') as f:\n content = f.read()\n logger.info(f\"Successfully read with utf-8-sig: {filepath}\")\n return content\n except Exception:\n return None\n \n except IOError as e:\n logger.error(f\"IO error reading {filepath}: {str(e)}\")\n return None\n \n except Exception as e:\n logger.error(\n f\"Unexpected error reading {filepath}: {str(e)}\",\n exc_info=True\n )\n return None\n \n def write_text_file(\n self,\n filename: str,\n content: str,\n encoding: str = 'utf-8',\n create_dir: bool = True\n ) -> bool:\n \"\"\"\n テキストファイルに書き込む\n \n Args:\n filename: ファイル名\n content: 書き込み内容\n encoding: 文字コード\n create_dir: ディレクトリが無い場合は作成\n \n Returns:\n 成功時True\n \"\"\"\n filepath = self.base_directory / filename\n \n try:\n # ディレクトリの作成\n if create_dir and not filepath.parent.exists():\n filepath.parent.mkdir(parents=True, exist_ok=True)\n logger.info(f\"Created directory: {filepath.parent}\")\n \n with open(filepath, 'w', encoding=encoding) as f:\n f.write(content)\n logger.info(f\"Wrote file: {filepath}\")\n return True\n \n except PermissionError:\n logger.error(f\"Permission denied writing to {filepath}\")\n return False\n \n except OSError as e:\n if e.errno == 28: # ディスク満杯\n logger.error(f\"No space left on device: {filepath}\")\n else:\n logger.error(f\"OS error writing {filepath}: {str(e)}\")\n return False\n \n except Exception as e:\n logger.error(\n f\"Unexpected error writing {filepath}: {str(e)}\",\n exc_info=True\n )\n return False\n \n def read_json_file(\n self,\n filename: str\n ) -> Optional[Dict[str, Any]]:\n \"\"\"\n JSONファイルを読み込む\n \n Args:\n filename: ファイル名\n \n Returns:\n パースされたJSONまたはNone\n \"\"\"\n filepath = self.base_directory / filename\n \n try:\n with open(filepath, 'r', encoding='utf-8') as f:\n data = json.load(f)\n logger.info(f\"Loaded JSON from {filepath}\")\n return data\n \n except FileNotFoundError:\n logger.error(f\"JSON file not found: {filepath}\")\n return None\n \n except json.JSONDecodeError as e:\n logger.error(\n f\"Invalid JSON in {filepath} at line {e.lineno}, \"\n f\"column {e.colno}: {e.msg}\"\n )\n return None\n \n except Exception as e:\n logger.error(\n f\"Error reading JSON {filepath}: {str(e)}\",\n exc_info=True\n )\n return None\n \n def write_json_file(\n self,\n filename: str,\n data: Dict[str, Any],\n indent: int = 2\n ) -> bool:\n \"\"\"\n JSONファイルに書き込む\n \n Args:\n filename: ファイル名\n data: 書き込みデータ\n indent: インデント数\n \n Returns:\n 成功時True\n \"\"\"\n filepath = self.base_directory / filename\n \n try:\n filepath.parent.mkdir(parents=True, exist_ok=True)\n \n with open(filepath, 'w', encoding='utf-8') as f:\n json.dump(data, f, indent=indent, ensure_ascii=False)\n logger.info(f\"Wrote JSON to {filepath}\")\n return True\n \n except TypeError as e:\n logger.error(\n f\"Data not JSON serializable: {str(e)}\"\n )\n return False\n \n except Exception as e:\n logger.error(\n f\"Error writing JSON {filepath}: {str(e)}\",\n exc_info=True\n )\n return False\n \n def read_csv_file(\n self,\n filename: str\n ) -> Optional[List[Dict[str, Any]]]:\n \"\"\"\n CSVファイルを読み込む\n \n Args:\n filename: ファイル名\n \n Returns:\n CSVデータのリストまたはNone\n \"\"\"\n filepath = self.base_directory / filename\n \n try:\n with open(filepath, 'r', encoding='utf-8') as f:\n reader = csv.DictReader(f)\n data = list(reader)\n logger.info(f\"Read {len(data)} rows from {filepath}\")\n return data\n \n except FileNotFoundError:\n logger.error(f\"CSV file not found: {filepath}\")\n return None\n \n except csv.Error as e:\n logger.error(f\"CSV parsing error in {filepath}: {str(e)}\")\n return None\n \n except Exception as e:\n logger.error(\n f\"Error reading CSV {filepath}: {str(e)}\",\n exc_info=True\n )\n return None\n\n# 使用例\nif __name__ == \"__main__\":\n logging.basicConfig(\n level=logging.INFO,\n format='%(asctime)s - %(levelname)s - %(message)s'\n )\n \n processor = FileProcessor(\"/tmp/data\")\n \n # テキストファイル書き込み\n success = processor.write_text_file(\n \"test.txt\",\n \"Hello, World!\"\n )\n print(f\"Write text: {success}\")\n \n # テキストファイル読み込み\n content = processor.read_text_file(\"test.txt\")\n if content:\n print(f\"Read content: {content}\")\n \n # JSONファイル書き込み\n data = {\n \"name\": \"Alice\",\n \"age\": 30,\n \"email\": \"alice@example.com\"\n }\n success = processor.write_json_file(\"user.json\", data)\n print(f\"Write JSON: {success}\")\n \n # JSONファイル読み込み\n loaded_data = processor.read_json_file(\"user.json\")\n if loaded_data:\n print(f\"Loaded JSON: {loaded_data}\")\n\n\n
4. よくある応用パターン
\n\n
4.1 カスタム例外クラスの作成
\n
業務アプリケーションでは、汎用的な例外クラスではなく、ビジネスロジックに特化したカスタム例外を定義することが重要です。これにより、エラーハンドリングが明確になり、デバッグも容易になります。
\n\n
class APIError(Exception):\n \"\"\"API関連のエラー基底クラス\"\"\"\n pass\n\nclass RetryableAPIError(APIError):\n \"\"\"リトライ可能なAPIエラー\"\"\"\n def __init__(self, status_code: int, message: str):\n self.status_code = status_code\n self.message = message\n super().__init__(f\"[{status_code}] {message}\")\n\nclass NonRetryableAPIError(APIError):\n \"\"\"リトライ不可のAPIエラー\"\"\"\n pass\n\nclass ValidationError(Exception):\n \"\"\"バリデーションエラー\"\"\"\n def __init__(self, field: str, value: Any, reason: str):\n self.field = field\n self.value = value\n self.reason = reason\n super().__init__(\n f\"Validation failed for '{field}': {reason} (value={value})\"\n )\n\n# 使用例\ndef validate_email(email: str):\n if '@' not in email:\n raise ValidationError(\n field='email',\n value=email,\n reason='Invalid format'\n )\n\ntry:\n validate_email('invalid-email')\nexcept ValidationError as e:\n print(f\"Field: {e.field}\")\n print(f\"Reason: {e.reason}\")\n logger.error(str(e))\n\n\n
4.2 コンテキストマネージャーを使った自動リソース管理
\n
with文とコンテキストマネージャーを組み合わせることで、リソースの確実なクローズを保証しながら、例外処理を適切に行えます。
\n\n
from contextlib import contextmanager\nfrom typing import Generator\n\n@contextmanager\ndef database_transaction(db_manager: DatabaseManager) -> Generator:\n \"\"\"トランザクションを自動的に開始・コミット・ロールバック\"\"\"\n conn = None\n try:\n conn = db_manager.get_connection()\n conn.execute(\"BEGIN TRANSACTION\")\n yield conn\n conn.commit()\n logger.info(\"Transaction committed\")\n except Exception as e:\n if conn:\n conn.rollback()\n logger.error(f\"Transaction rolled back: {str(e)}\")\n raise\n finally:\n if conn:\n conn.close()\n\n# 使用例\ndb = DatabaseManager(\":memory:\")\n\ntry:\n with database_transaction(db) as conn:\n cursor = conn.cursor()\n cursor.execute(\"INSERT INTO users (name, email) VALUES (?, ?)\",\n (\"John\", \"john@example.com\"))\n # エラーが発生するとロールバックされる\n cursor.execute(\"INSERT INTO invalid_table VALUES (1)\")\nexcept Exception as e:\n logger.error(f\"Transaction failed: {str(e)}\")\n\n\n
4.3 デコレーターを使った例外処理の共通化
\n
同じ例外処理パターンが複数の関数で繰り返される場合、デコレーターで共通化することでコードの保守性が向上します。
\n\n
from functools import wraps\nfrom typing import Callable, Any\n\ndef handle_api_errors(max_retries: int = 3):\n \"\"\"API呼び出しの例外処理とリトライをデコレート\"\"\"\n def decorator(func: Callable) -> Callable:\n @wraps(func)\n def wrapper(*args, **kwargs) -> Any:\n for attempt in range(max_retries):\n try:\n return func(*args, **kwargs)\n except RetryableAPIError as e:\n if attempt < max_retries - 1:\n wait_time = 2 ** attempt\n logger.warning(\n f\"Retrying {func.__name__} \"\n f\"(attempt {attempt + 1}/{max_retries}) \"\n f\"after {wait_time}s\"\n )\n sleep(wait_time)\n else:\n logger.error(\n f\"Max retries exceeded for {func.__name__}: {str(e)}\"\n )\n raise\n except NonRetryableAPIError as e:\n logger.error(f\"Non-retryable error in {func.__name__}: {str(e)}\")\n raise\n except Exception as e:\n logger.error(\n f\"Unexpected error in {func.__name__}: {str(e)}\",\n exc_info=True\n )\n raise\n return wrapper\n return decorator\n\n@handle_api_errors(max_retries=3)\ndef fetch_user_data(user_id: int) -> Dict[str, Any]:\n \"\"\"ユーザーデータを取得\"\"\"\n response = requests.get(\n f\"https://api.example.com/users/{user_id}\",\n timeout=10\n )\n \n if response.status_code == 429:\n raise RetryableAPIError(429, \"Rate limited\")\n elif response.status_code != 200:\n raise NonRetryableAPIError(\n f\
