入門系列介紹了異常處理的基本策略。本章深入探討如何為大型專案設計異常架構,包括異常層級、異常鏈、以及 Python 3.11 引入的 ExceptionGroup。

先備知識

異常層級設計

為什麼需要層級設計?

當專案規模增長,你會需要:

  • 區分錯誤來源:資料庫錯誤 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 e

suppress 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 模式強制錯誤處理,適合預期的錯誤

思考題

  1. 為什麼要使用異常層級而不是一個通用的 AppError
  2. from efrom None 分別在什麼情況下使用?
  3. ExceptionGroup 解決了什麼問題?在什麼場景下最有用?

上一章:3.5.1 泛型進階 下一章:3.5.3 進階上下文管理