3.5.2 異常設計架構
3.5.2 異常設計架構
入門系列介紹了異常處理的基本策略。本章深入探討如何為大型專案設計異常架構,包括異常層級、異常鏈、以及 Python 3.11 引入的 ExceptionGroup。
先備知識
- 入門系列 5.1 異常處理策略
異常層級設計
為什麼需要層級設計?
當專案規模增長,你會需要:
- 區分錯誤來源:資料庫錯誤 vs 網路錯誤 vs 驗證錯誤
- 提供不同處理策略:有些錯誤可重試,有些需要通知用戶
- 方便呼叫者選擇處理粒度:捕獲所有錯誤 vs 只捕獲特定錯誤
模組級異常基類
1# myapp/exceptions.py
2
3class AppError(Exception):
4 """應用程式的基礎異常類別
5
6 所有自訂異常都應繼承此類別,讓呼叫者可以:
7 - except AppError: 捕獲所有應用程式錯誤
8 - except SpecificError: 只捕獲特定錯誤
9 """
10
11 def __init__(self, message: str, code: str | None = None) -> None:
12 super().__init__(message)
13 self.message = message
14 self.code = code # 可選的錯誤碼,方便 API 回應
15
16 def __str__(self) -> str:
17 if self.code:
18 return f"[{self.code}] {self.message}"
19 return self.message細粒度異常分類
1# 第一層:按錯誤類型分類
2class ValidationError(AppError):
3 """驗證相關錯誤"""
4 pass
5
6class DataAccessError(AppError):
7 """資料存取相關錯誤"""
8 pass
9
10class NetworkError(AppError):
11 """網路相關錯誤"""
12 pass
13
14class ConfigurationError(AppError):
15 """配置相關錯誤"""
16 pass
17
18# 第二層:更細的分類
19class FieldValidationError(ValidationError):
20 """欄位驗證錯誤"""
21
22 def __init__(self, field: str, message: str) -> None:
23 super().__init__(f"Field '{field}': {message}", code="VALIDATION_FIELD")
24 self.field = field
25
26class SchemaValidationError(ValidationError):
27 """資料結構驗證錯誤"""
28
29 def __init__(self, message: str, errors: list[str]) -> None:
30 super().__init__(message, code="VALIDATION_SCHEMA")
31 self.errors = errors
32
33class EntityNotFoundError(DataAccessError):
34 """實體不存在"""
35
36 def __init__(self, entity_type: str, entity_id: int | str) -> None:
37 super().__init__(
38 f"{entity_type} with id '{entity_id}' not found",
39 code="NOT_FOUND"
40 )
41 self.entity_type = entity_type
42 self.entity_id = entity_id
43
44class DuplicateEntityError(DataAccessError):
45 """實體已存在"""
46
47 def __init__(self, entity_type: str, field: str, value: str) -> None:
48 super().__init__(
49 f"{entity_type} with {field}='{value}' already exists",
50 code="DUPLICATE"
51 )使用範例
1from myapp.exceptions import (
2 AppError,
3 ValidationError,
4 FieldValidationError,
5 EntityNotFoundError
6)
7
8def create_user(data: dict) -> User:
9 # 驗證
10 if not data.get("email"):
11 raise FieldValidationError("email", "Email is required")
12
13 if not is_valid_email(data["email"]):
14 raise FieldValidationError("email", "Invalid email format")
15
16 # 檢查重複
17 if user_exists(data["email"]):
18 raise DuplicateEntityError("User", "email", data["email"])
19
20 return User(**data)
21
22# 呼叫者可以選擇處理粒度
23def handle_user_creation(data: dict) -> dict:
24 try:
25 user = create_user(data)
26 return {"status": "success", "user_id": user.id}
27
28 except FieldValidationError as e:
29 # 處理欄位驗證錯誤
30 return {"status": "error", "field": e.field, "message": e.message}
31
32 except ValidationError as e:
33 # 處理所有驗證錯誤
34 return {"status": "error", "message": str(e)}
35
36 except AppError as e:
37 # 處理所有應用程式錯誤
38 return {"status": "error", "code": e.code, "message": str(e)}異常鏈的進階用法
__cause__ vs __context__
Python 有兩種異常鏈機制:
1# __cause__:明確指定的原因(使用 from)
2try:
3 result = int("not a number")
4except ValueError as e:
5 raise DataAccessError("Failed to parse ID") from e
6 # e 會被設為 __cause__
7
8# __context__:隱式的上下文(在 except 中 raise)
9try:
10 result = int("not a number")
11except ValueError:
12 raise DataAccessError("Failed to parse ID")
13 # 原始的 ValueError 會被設為 __context__輸出差異:
1# 使用 from(__cause__)
2DataAccessError: Failed to parse ID
3
4The above exception was the direct cause of the following exception:
5...
6
7# 不使用 from(__context__)
8DataAccessError: Failed to parse ID
9
10During handling of the above exception, another exception occurred:
11...何時使用 from
1# 情況 1:轉換異常型別
2def load_user(user_id: int) -> User:
3 try:
4 data = database.query(f"SELECT * FROM users WHERE id = {user_id}")
5 return User(**data)
6 except DatabaseError as e:
7 # 將底層資料庫錯誤轉換為應用層錯誤
8 raise EntityNotFoundError("User", user_id) from e
9
10# 情況 2:添加上下文資訊
11def process_config(path: str) -> dict:
12 try:
13 with open(path) as f:
14 return json.load(f)
15 except json.JSONDecodeError as e:
16 # 添加檔案路徑資訊
17 raise ConfigurationError(f"Invalid JSON in {path}") from esuppress context
有時你想切斷異常鏈:
1def get_value(data: dict, key: str) -> str:
2 try:
3 return data[key]
4 except KeyError:
5 # 使用 from None 切斷異常鏈
6 raise ValueError(f"Required key '{key}' not found") from None輸出:
1# 沒有 from None:會顯示原始 KeyError
2# 有 from None:只顯示 ValueError,不顯示 KeyError
3ValueError: Required key 'name' not found使用時機:
- 原始異常不相關或會造成困惑
- 刻意隱藏實作細節
- 簡化錯誤訊息
ExceptionGroup(Python 3.11+)
問題情境
當需要同時處理多個異常時,傳統方式有限制:
1# 傳統方式:只能記錄第一個錯誤,或全部收集後手動處理
2errors = []
3for task in tasks:
4 try:
5 task.run()
6 except Exception as e:
7 errors.append(e)
8
9if errors:
10 # 該拋出哪一個?或如何表示多個錯誤?
11 raise errors[0] # 丟失其他錯誤ExceptionGroup 解決方案
1# Python 3.11+
2def run_all_tasks(tasks: list[Task]) -> None:
3 errors = []
4 for task in tasks:
5 try:
6 task.run()
7 except Exception as e:
8 errors.append(e)
9
10 if errors:
11 raise ExceptionGroup("Multiple tasks failed", errors)except* 語法
1try:
2 run_all_tasks(tasks)
3except* ValueError as eg:
4 # eg 是包含所有 ValueError 的 ExceptionGroup
5 print(f"Value errors: {eg.exceptions}")
6except* TypeError as eg:
7 # eg 是包含所有 TypeError 的 ExceptionGroup
8 print(f"Type errors: {eg.exceptions}")實際範例:並行任務處理
1import asyncio
2from typing import Callable, Any
3
4async def run_parallel(
5 tasks: list[Callable[[], Any]]
6) -> list[Any]:
7 """並行執行多個任務,收集所有錯誤"""
8
9 async def run_task(task: Callable[[], Any]) -> Any:
10 return task()
11
12 # 使用 TaskGroup(Python 3.11+)
13 results = []
14 async with asyncio.TaskGroup() as tg:
15 futures = [tg.create_task(run_task(task)) for task in tasks]
16
17 # 如果有異常,TaskGroup 會拋出 ExceptionGroup
18 return [f.result() for f in futures]
19
20# 處理
21async def main():
22 tasks = [task1, task2, task3]
23
24 try:
25 results = await run_parallel(tasks)
26 except* ValueError as eg:
27 for e in eg.exceptions:
28 print(f"Validation failed: {e}")
29 except* ConnectionError as eg:
30 for e in eg.exceptions:
31 print(f"Connection failed: {e}")巢狀 ExceptionGroup
1# ExceptionGroup 可以巢狀
2outer = ExceptionGroup("outer", [
3 ValueError("value error"),
4 ExceptionGroup("inner", [
5 TypeError("type error 1"),
6 TypeError("type error 2"),
7 ])
8])
9
10# 使用 .subgroup() 過濾
11type_errors = outer.subgroup(lambda e: isinstance(e, TypeError))
12# 返回只包含 TypeError 的新 ExceptionGroup異常 vs 返回值
何時使用異常
1# 適合使用異常的情況:
2
3# 1. 無法繼續執行的錯誤
4def connect_database(url: str) -> Connection:
5 if not url:
6 raise ConfigurationError("Database URL is required")
7 # ...
8
9# 2. 呼叫者通常不處理的錯誤
10def validate_schema(data: dict) -> None:
11 errors = find_schema_errors(data)
12 if errors:
13 raise SchemaValidationError("Invalid data", errors)
14
15# 3. 深層呼叫需要跨多層傳遞錯誤
16def process_order(order_id: int) -> Order:
17 order = load_order(order_id) # 可能 raise EntityNotFoundError
18 validate_order(order) # 可能 raise ValidationError
19 return execute_order(order) # 可能 raise PaymentError何時使用 Result 模式
1from dataclasses import dataclass
2from typing import Generic, TypeVar
3
4T = TypeVar("T")
5E = TypeVar("E")
6
7@dataclass
8class Ok(Generic[T]):
9 value: T
10
11@dataclass
12class Err(Generic[E]):
13 error: E
14
15Result = Ok[T] | Err[E]
16
17# 適合使用 Result 的情況:
18
19# 1. 錯誤是預期的、常見的
20def find_user(user_id: int) -> Result[User, str]:
21 user = db.get_user(user_id)
22 if user is None:
23 return Err(f"User {user_id} not found")
24 return Ok(user)
25
26# 2. 需要強制呼叫者處理錯誤
27def parse_config(text: str) -> Result[Config, list[str]]:
28 errors = []
29 # ... 解析邏輯
30 if errors:
31 return Err(errors)
32 return Ok(config)
33
34# 使用
35match parse_config(text):
36 case Ok(config):
37 use_config(config)
38 case Err(errors):
39 show_errors(errors)
40
41# 3. 效能敏感的熱點路徑
42# 異常有效能成本,頻繁拋出會影響效能混合使用
1class UserService:
2 """服務層:使用異常表示錯誤"""
3
4 def create_user(self, data: dict) -> User:
5 if not data.get("email"):
6 raise FieldValidationError("email", "required")
7 # ...
8 return user
9
10class UserAPI:
11 """API 層:將異常轉換為 Result"""
12
13 def __init__(self, service: UserService) -> None:
14 self.service = service
15
16 def create_user(self, data: dict) -> Result[dict, dict]:
17 try:
18 user = self.service.create_user(data)
19 return Ok({"id": user.id, "email": user.email})
20 except FieldValidationError as e:
21 return Err({"field": e.field, "message": e.message})
22 except AppError as e:
23 return Err({"code": e.code, "message": str(e)})補充:contextlib.suppress
簡潔地忽略特定異常:
1from contextlib import suppress
2
3# 傳統方式
4try:
5 os.remove(temp_file)
6except FileNotFoundError:
7 pass
8
9# 使用 suppress
10with suppress(FileNotFoundError):
11 os.remove(temp_file)
12
13# 可以同時忽略多種異常
14with suppress(FileNotFoundError, PermissionError):
15 os.remove(temp_file)注意:只應該用於「真正可以安全忽略」的異常。
設計檢查表
設計異常架構時,考慮以下問題:
| 問題 | 建議 |
|---|---|
| 錯誤會被如何處理? | 決定用異常還是返回值 |
| 需要區分多少種錯誤? | 決定異常層級深度 |
| 呼叫者需要什麼資訊? | 決定異常屬性 |
| 原始錯誤重要嗎? | 決定是否使用 from |
| 會有並行錯誤嗎? | 考慮 ExceptionGroup |
小結
| 概念 | 用途 |
|---|---|
| 異常層級 | 區分錯誤來源,提供不同處理粒度 |
raise ... from e | 明確指定異常原因,保留追蹤資訊 |
raise ... from None | 切斷異常鏈,隱藏實作細節 |
| ExceptionGroup | 同時處理多個異常 |
except* | 按型別過濾 ExceptionGroup |
| Result 模式 | 強制錯誤處理,適合預期的錯誤 |
思考題
- 為什麼要使用異常層級而不是一個通用的
AppError? from e和from None分別在什麼情況下使用?- ExceptionGroup 解決了什麼問題?在什麼場景下最有用?
上一章:3.5.1 泛型進階 下一章:3.5.3 進階上下文管理