第15章 异常处理

学习目标

完成本章学习后，读者应能够：

1. 理解异常机制：掌握Python异常的底层实现原理、异常传播机制与调用栈展开过程
2. 精通try-except语法：熟练使用try/except/else/finally的完整语法，理解各子句的执行语义
3. 设计异常层次结构：能够为项目设计合理的自定义异常层次结构，遵循Liskov替换原则
4. 掌握异常链：理解raise ... from的语义，正确使用显式链与隐式链
5. 运用上下文管理器：深入理解with语句与上下文管理器协议，设计资源管理模式
6. 遵循最佳实践：掌握异常处理的行业最佳实践，避免常见反模式

15.1 异常机制基础

15.1.1 异常的本质

异常（Exception）是Python处理错误和特殊情况的机制。从底层看，异常是一种控制流转移机制——当解释器检测到错误条件时，它会创建一个异常对象并”抛出”（raise），然后沿着调用栈向上”传播”（propagate），直到被”捕获”（catch）或导致程序终止。

def demonstrate_exception_propagation():
    def level_3():
        raise ValueError("来自level_3的错误")

    def level_2():
        level_3()

    def level_1():
        level_2()

    try:
        level_1()
    except ValueError as e:
        print(f"捕获异常: {e}")
        import traceback
        traceback.print_exc()

demonstrate_exception_propagation()

异常与错误的区别：

• 错误（Error）：程序无法恢复的严重问题，如SyntaxError、MemoryError、SystemExit
• 异常（Exception）：程序可以捕获和处理的异常情况，如ValueError、TypeError
• 警告（Warning）：潜在问题的提示，默认不中断程序执行

15.1.2 try-except完整语法

def safe_divide(a, b):
    try:
        result = a / b
    except ZeroDivisionError as e:
        print(f"除零错误: {e}")
        result = float("inf")
    except TypeError as e:
        print(f"类型错误: {e}")
        result = None
    else:
        print(f"成功计算: {result}")
    finally:
        print("清理完成")
    return result

print(safe_divide(10, 3))
print("---")
print(safe_divide(10, 0))
print("---")
print(safe_divide(10, "a"))

各子句的执行语义：

子句	执行条件	典型用途
try	始终执行	包含可能抛出异常的代码
except	try块抛出匹配异常时	处理特定异常
else	try块未抛出异常时	仅在无异常时执行的代码
finally	始终执行	资源清理、状态恢复

关键细节：

• else子句中的代码不会被except捕获，这有助于避免意外屏蔽异常
• finally子句在return、break、continue之前执行
• finally中的return会覆盖try或except中的return

def finally_return_demo():
    try:
        return "from try"
    finally:
        return "from finally"

print(finally_return_demo())

def finally_with_exception():
    try:
        raise ValueError("try中的异常")
    finally:
        print("finally执行了，异常继续传播")

try:
    finally_with_exception()
except ValueError as e:
    print(f"捕获: {e}")

15.1.3 多异常处理

def parse_config(value, key):
    try:
        value = int(value)
        result = 100 / value
        data = {"ratio": result}
        return data[key]
    except (ValueError, TypeError) as e:
        print(f"值/类型错误: {e}")
        return None
    except ZeroDivisionError:
        print("除零错误，使用默认值")
        return {"ratio": 0}
    except KeyError as e:
        print(f"键错误: {e}")
        return None
    except Exception as e:
        print(f"未预期的异常: {type(e).__name__}: {e}")
        raise

parse_config("10", "ratio")
parse_config("0", "ratio")
parse_config("abc", "ratio")
parse_config("10", "missing")

15.1.4 异常对象与内省

def inspect_exception():
    try:
        1 / 0
    except ZeroDivisionError as e:
        print(f"异常类型: {type(e)}")
        print(f"异常消息: {e}")
        print(f"异常参数: {e.args}")
        print(f"字符串表示: {str(e)}")
        print(f"repr表示: {repr(e)}")

        import sys
        exc_type, exc_value, exc_tb = sys.exc_info()
        print(f"sys.exc_info类型: {exc_type}")
        print(f"sys.exc_info值: {exc_value}")

inspect_exception()

15.2 异常层次结构

15.2.1 内置异常体系

Python的内置异常形成了一个层次结构，所有异常都继承自BaseException：

BaseException
├── SystemExit
├── KeyboardInterrupt
├── GeneratorExit
└── Exception
    ├── StopIteration
    ├── ArithmeticError
    │   ├── ZeroDivisionError
    │   ├── OverflowError
    │   └── FloatingPointError
    ├── LookupError
    │   ├── IndexError
    │   └── KeyError
    ├── OSError
    │   ├── FileNotFoundError
    │   ├── PermissionError
    │   ├── IsADirectoryError
    │   └── TimeoutError
    ├── TypeError
    ├── ValueError
    │   └── UnicodeError
    ├── AttributeError
    ├── RuntimeError
    │   └── RecursionError
    ├── NameError
    │   └── UnboundLocalError
    ├── ImportError
    │   └── ModuleNotFoundError
    └── ... (更多)

def explore_exception_hierarchy():
    print(f"Exception的基类: {Exception.__bases__}")
    print(f"ArithmeticError的子类: {ArithmeticError.__subclasses__()}")
    print(f"OSError的子类: {OSError.__subclasses__()}")

    print("\nLookupError子类:")
    for cls in LookupError.__subclasses__():
        print(f"  {cls.__name__}")

    def print_hierarchy(cls, indent=0):
        print("  " * indent + cls.__name__)
        for sub in cls.__subclasses__():
            print_hierarchy(sub, indent + 1)

    print("\nArithmeticError层次:")
    print_hierarchy(ArithmeticError)

explore_exception_hierarchy()

15.2.2 异常匹配规则

except子句匹配异常时，会检查异常是否是指定类的实例或其子类的实例。因此，捕获父类异常会同时捕获所有子类异常：

def exception_matching():
    try:
        raise FileNotFoundError("文件不存在")
    except OSError as e:
        print(f"捕获OSError: {type(e).__name__}: {e}")

    try:
        raise ZeroDivisionError("除零")
    except ArithmeticError as e:
        print(f"捕获ArithmeticError: {type(e).__name__}: {e}")

exception_matching()

重要原则：except子句的顺序应从具体到通用，否则更具体的异常处理永远不会被执行：

def wrong_order():
    try:
        raise FileNotFoundError("文件不存在")
    except OSError:
        print("OSError处理")
    except FileNotFoundError:
        print("这永远不会执行！")

def correct_order():
    try:
        raise FileNotFoundError("文件不存在")
    except FileNotFoundError:
        print("FileNotFoundError处理")
    except OSError:
        print("OSError处理")

correct_order()

15.3 异常链

15.3.1 显式链（raise … from）

raise ... from语法用于在捕获一个异常后抛出另一个异常，同时保留原始异常的上下文：

def load_config(filepath):
    try:
        with open(filepath) as f:
            return f.read()
    except FileNotFoundError as e:
        raise ConfigError(f"配置文件缺失: {filepath}") from e

class ConfigError(Exception):
    pass

try:
    load_config("missing_config.ini")
except ConfigError as e:
    print(f"异常: {e}")
    print(f"原因: {e.__cause__}")
    print(f"原因类型: {type(e.__cause__).__name__}")

15.3.2 隐式链

在异常处理过程中抛出新异常时，Python自动设置__context__属性：

def implicit_chain():
    try:
        raise ValueError("原始错误")
    except ValueError:
        raise TypeError("处理过程中出错")

try:
    implicit_chain()
except TypeError as e:
    print(f"异常: {e}")
    print(f"上下文: {e.__context__}")
    print(f"上下文类型: {type(e.__context__).__name__}")

15.3.3 抑制链（raise … from None）

当不希望暴露原始异常时，使用raise ... from None抑制异常链：

def suppressed_chain():
    try:
        int("abc")
    except ValueError:
        raise ValueError("无效输入") from None

try:
    suppressed_chain()
except ValueError as e:
    print(f"异常: {e}")
    print(f"__cause__: {e.__cause__}")
    print(f"__context__: {e.__context__}")
    print(f"__suppress_context__: {e.__suppress_context__}")

15.3.4 异常链的最佳实践

class DatabaseError(Exception):
    pass

class ConnectionError(DatabaseError):
    pass

class QueryError(DatabaseError):
    pass

def execute_query(sql):
    import random
    if random.random() < 0.3:
        raise ConnectionError("数据库连接失败")
    if random.random() < 0.3:
        raise QueryError(f"查询语法错误: {sql}")
    return [{"id": 1, "name": "Alice"}]

def get_user(user_id):
    try:
        result = execute_query(f"SELECT * FROM users WHERE id = {user_id}")
    except ConnectionError as e:
        raise DatabaseError("无法获取用户数据: 连接失败") from e
    except QueryError as e:
        raise DatabaseError("无法获取用户数据: 查询错误") from e
    return result[0] if result else None

try:
    user = get_user(1)
except DatabaseError as e:
    print(f"数据库错误: {e}")
    if e.__cause__:
        print(f"原始原因: {type(e.__cause__).__name__}: {e.__cause__}")

15.4 自定义异常

15.4.1 设计异常层次结构

为项目设计自定义异常层次结构是重要的架构决策。好的异常层次应反映业务领域概念：

class ApplicationError(Exception):
    """应用程序基础异常"""
    def __init__(self, message, code=None):
        self.message = message
        self.code = code
        super().__init__(message)

    def to_dict(self):
        result = {"error": self.message, "type": type(self).__name__}
        if self.code:
            result["code"] = self.code
        return result

class ValidationError(ApplicationError):
    """数据验证错误"""
    def __init__(self, field, message, value=None, code="VALIDATION_ERROR"):
        self.field = field
        self.value = value
        super().__init__(f"{field}: {message}", code=code)

class BusinessError(ApplicationError):
    """业务逻辑错误"""
    pass

class InsufficientFundsError(BusinessError):
    """余额不足"""
    def __init__(self, balance, amount):
        self.balance = balance
        self.amount = amount
        self.deficit = amount - balance
        super().__init__(
            f"余额不足: 当前{balance}, 需要{amount}, 缺口{self.deficit}",
            code="INSUFFICIENT_FUNDS"
        )

class AccountLockedError(BusinessError):
    """账户锁定"""
    def __init__(self, account_id, reason):
        self.account_id = account_id
        self.reason = reason
        super().__init__(
            f"账户 {account_id} 已锁定: {reason}",
            code="ACCOUNT_LOCKED"
        )

class InfrastructureError(ApplicationError):
    """基础设施错误"""
    pass

class DatabaseError(InfrastructureError):
    """数据库错误"""
    pass

class CacheError(InfrastructureError):
    """缓存错误"""
    pass

def withdraw(balance, amount, account_locked=False):
    if account_locked:
        raise AccountLockedError("ACC-001", "多次密码错误")
    if amount > balance:
        raise InsufficientFundsError(balance, amount)
    return balance - amount

try:
    withdraw(100, 150)
except InsufficientFundsError as e:
    print(f"错误: {e.message}")
    print(f"代码: {e.code}")
    print(f"缺口: {e.deficit}")
    print(f"字典: {e.to_dict()}")

15.4.2 异常与错误码

在企业级应用中，异常通常与错误码关联，便于日志分析和客户端处理：

class ErrorCode:
    VALIDATION_ERROR = "VAL_001"
    NOT_FOUND = "RES_001"
    UNAUTHORIZED = "AUTH_001"
    FORBIDDEN = "AUTH_002"
    CONFLICT = "RES_002"
    RATE_LIMITED = "RATE_001"
    INTERNAL_ERROR = "SYS_001"

class APIError(ApplicationError):
    """API错误基类"""
    status_code = 500

    def __init__(self, message, code=None, details=None):
        self.details = details or {}
        super().__init__(message, code=code)

    def to_response(self):
        return {
            "error": {
                "code": self.code,
                "message": self.message,
                "details": self.details,
            }
        }

class NotFoundError(APIError):
    status_code = 404

    def __init__(self, resource, resource_id=None):
        details = {"resource": resource}
        if resource_id:
            details["id"] = resource_id
        super().__init__(
            f"{resource}不存在" + (f" (id={resource_id})" if resource_id else ""),
            code=ErrorCode.NOT_FOUND,
            details=details
        )

class UnauthorizedError(APIError):
    status_code = 401

    def __init__(self, reason="认证失败"):
        super().__init__(reason, code=ErrorCode.UNAUTHORIZED)

def get_user(user_id):
    if user_id < 0:
        raise NotFoundError("User", user_id)
    return {"id": user_id, "name": "Alice"}

try:
    get_user(-1)
except APIError as e:
    print(f"HTTP {e.status_code}: {e.to_response()}")

15.4.3 异常聚合

当需要同时报告多个错误时（如表单验证），可以使用异常聚合：

class MultiError(Exception):
    """聚合多个异常"""
    def __init__(self, errors, message="多个错误发生"):
        self.errors = list(errors)
        super().__init__(message)

    def __str__(self):
        lines = [super().__str__()]
        for i, err in enumerate(self.errors, 1):
            lines.append(f"  {i}. {err}")
        return "\n".join(lines)

    def to_dict(self):
        return {
            "error": super().__str__(),
            "errors": [
                {"type": type(e).__name__, "message": str(e)}
                for e in self.errors
            ]
        }

def validate_user(name, age, email):
    errors = []

    if not name or len(name) < 2:
        errors.append(ValidationError("name", "名称至少2个字符"))
    if age < 0 or age > 150:
        errors.append(ValidationError("age", "年龄必须在0-150之间", value=age))
    if "@" not in email:
        errors.append(ValidationError("email", "无效的邮箱格式", value=email))

    if errors:
        raise MultiError(errors, "用户数据验证失败")

try:
    validate_user("A", 200, "invalid")
except MultiError as e:
    print(e)
    print(f"\n字典: {e.to_dict()}")

15.5 上下文管理器

15.5.1 with语句与上下文管理协议

with语句是Python资源管理的核心机制，它基于上下文管理协议（__enter__/__exit__）：

class Timer:
    def __init__(self, name=""):
        self.name = name
        self.elapsed = 0

    def __enter__(self):
        import time
        self._start = time.perf_counter()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        import time
        self.elapsed = time.perf_counter() - self._start
        print(f"[{self.name}] 耗时: {self.elapsed:.6f}秒")
        return False

with Timer("计算") as t:
    total = sum(range(1_000_000))

print(f"外部访问: {t.elapsed:.6f}秒")

__exit__方法的返回值决定是否抑制异常：

• True：抑制异常，继续执行with之后的代码
• False：异常继续传播

class SuppressErrors:
    def __init__(self, *exceptions):
        self.exceptions = exceptions

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type and issubclass(exc_type, self.exceptions):
            print(f"抑制异常: {exc_type.__name__}: {exc_val}")
            return True
        return False

with SuppressErrors(ValueError, TypeError):
    int("abc")

print("程序继续执行")

15.5.2 contextlib模块

contextlib模块提供了创建上下文管理器的便捷工具：

@contextmanager装饰器：

from contextlib import contextmanager

@contextmanager
def timer(name=""):
    import time
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        print(f"[{name}] 耗时: {elapsed:.6f}秒")

with timer("计算"):
    total = sum(range(1_000_000))

@contextmanager
def database_transaction(conn):
    try:
        yield conn
        conn.commit()
        print("事务提交")
    except Exception:
        conn.rollback()
        print("事务回滚")
        raise

class FakeConnection:
    def __init__(self):
        self.committed = False
        self.rolled_back = False

    def commit(self):
        self.committed = True

    def rollback(self):
        self.rolled_back = True

conn = FakeConnection()
with database_transaction(conn):
    pass
print(f"已提交: {conn.committed}")

conn2 = FakeConnection()
try:
    with database_transaction(conn2):
        raise ValueError("模拟错误")
except ValueError:
    pass
print(f"已回滚: {conn2.rolled_back}")

closing：

from contextlib import closing

class Resource:
    def __init__(self):
        self.closed = False

    def close(self):
        self.closed = True
        print("资源已关闭")

with closing(Resource()) as r:
    print(f"使用资源: closed={r.closed}")

print(f"关闭后: closed={r.closed}")

suppress：

from contextlib import suppress

with suppress(FileNotFoundError):
    with open("nonexistent.txt") as f:
        content = f.read()

print("文件不存在但程序继续")

with suppress(ValueError, TypeError):
    int("abc")

print("类型错误被抑制")

redirect_stdout / redirect_stderr：

from contextlib import redirect_stdout
import io

output = io.StringIO()
with redirect_stdout(output):
    print("这行被重定向")
    print("不会显示在控制台")

captured = output.getvalue()
print(f"捕获的输出: {repr(captured)}")

ExitStack：动态管理多个上下文：

from contextlib import ExitStack

def process_files(filepaths):
    with ExitStack() as stack:
        files = [
            stack.enter_context(open(fp, "r"))
            for fp in filepaths
        ]
        for f in files:
            print(f"处理: {f.name}")

process_files([])
print("ExitStack示例完成")

class ManagedResource:
    def __init__(self, name):
        self.name = name

    def __enter__(self):
        print(f"  获取: {self.name}")
        return self

    def __exit__(self, *args):
        print(f"  释放: {self.name}")
        return False

with ExitStack() as stack:
    resources = []
    for name in ["DB", "Cache", "Lock"]:
        resources.append(stack.enter_context(ManagedResource(name)))
    print("使用所有资源")

15.5.3 可重入与线程安全的上下文管理器

import threading

class LockManager:
    def __init__(self, lock):
        self.lock = lock

    def __enter__(self):
        self.lock.acquire()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.lock.release()
        return False

class ReadWriteLock:
    def __init__(self):
        self._read_ready = threading.Condition(threading.Lock())
        self._readers = 0

    @contextmanager
    def read_lock(self):
        with self._read_ready:
            self._readers += 1
        try:
            yield
        finally:
            with self._read_ready:
                self._readers -= 1
                if self._readers == 0:
                    self._read_ready.notify_all()

    @contextmanager
    def write_lock(self):
        self._read_ready.acquire()
        try:
            while self._readers > 0:
                self._read_ready.wait()
            yield
        finally:
            self._read_ready.release()

rw_lock = ReadWriteLock()

with rw_lock.read_lock():
    print("读取数据")

with rw_lock.write_lock():
    print("写入数据")

15.6 异常处理最佳实践

15.6.1 捕获具体异常

# 反模式：裸except
def bad_practice():
    try:
        result = 10 / 0
    except:
        print("出错了")

# 正确做法：捕获具体异常
def good_practice():
    try:
        result = 10 / 0
    except ZeroDivisionError as e:
        print(f"除零错误: {e}")
        result = float("inf")
    return result

# 反模式：过宽的Exception捕获
def too_broad():
    try:
        value = int("abc")
        result = 10 / value
    except Exception:
        pass

# 正确做法：分别处理
def specific_handling():
    try:
        value = int("abc")
        result = 10 / value
    except ValueError as e:
        print(f"输入无效: {e}")
    except ZeroDivisionError as e:
        print(f"除零错误: {e}")

15.6.2 不要忽略异常

# 反模式：空except
def silent_failure():
    try:
        important_operation()
    except Exception:
        pass

# 正确做法：至少记录日志
import logging
logger = logging.getLogger(__name__)

def logged_failure():
    try:
        important_operation()
    except Exception as e:
        logger.error(f"操作失败: {e}", exc_info=True)
        raise

def important_operation():
    raise RuntimeError("关键操作失败")

15.6.3 使用else子句

# 反模式：把所有代码放在try中
def bad_try_scope(filepath):
    try:
        with open(filepath) as f:
            data = f.read()
        processed = process(data)
        save(processed)
    except FileNotFoundError:
        print("文件不存在")

# 正确做法：最小化try范围
def good_try_scope(filepath):
    try:
        f = open(filepath)
    except FileNotFoundError:
        print("文件不存在")
        return
    else:
        data = f.read()
        f.close()
        processed = process(data)
        save(processed)

def process(data):
    return data.upper()

def save(data):
    print(f"保存: {data[:20]}...")

15.6.4 资源清理保证

# 反模式：手动清理
def manual_cleanup():
    f = None
    try:
        f = open("example.txt")
        data = f.read()
    except FileNotFoundError:
        print("文件不存在")
    finally:
        if f:
            f.close()

# 正确做法：使用with
def with_cleanup():
    try:
        with open("example.txt") as f:
            data = f.read()
    except FileNotFoundError:
        print("文件不存在")

15.6.5 异常与断言的选择

# 断言：用于检查程序内部不变量，不应被捕获
def calculate_average(numbers):
    assert len(numbers) > 0, "列表不能为空"
    return sum(numbers) / len(numbers)

# 异常：用于检查外部输入和可恢复错误
def safe_average(numbers):
    if not numbers:
        raise ValueError("列表不能为空")
    return sum(numbers) / len(numbers)

原则：

• 断言用于开发期间检测编程错误，可被-O标志禁用
• 异常用于运行时处理可预期的错误情况
• 永远不要用断言验证外部输入

15.6.6 异常处理装饰器

from functools import wraps
import logging

logger = logging.getLogger(__name__)

def handle_errors(*exceptions, default=None, log_level=logging.ERROR):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except exceptions as e:
                logger.log(log_level, f"{func.__name__} 失败: {e}")
                return default
        return wrapper
    return decorator

@handle_errors(ValueError, TypeError, default=0)
def parse_number(value):
    return int(value)

print(parse_number("42"))
print(parse_number("abc"))

@handle_errors(ConnectionError, default={}, log_level=logging.WARNING)
def fetch_config():
    raise ConnectionError("无法连接配置服务器")

print(fetch_config())

15.7 前沿技术动态

15.7.1 Python 3.11的异常组与except*

Python 3.11引入了ExceptionGroup和except*语法，用于处理多个同时发生的异常：

def demonstrate_exception_group():
    errors = []
    
    for task in ["task_a", "task_b", "task_c"]:
        try:
            if task == "task_a":
                raise ValueError("任务A值错误")
            if task == "task_b":
                raise TypeError("任务B类型错误")
            if task == "task_c":
                raise ConnectionError("任务C连接失败")
        except Exception as e:
            errors.append(e)
    
    if errors:
        raise ExceptionGroup("多个任务失败", errors)

try:
    demonstrate_exception_group()
except ExceptionGroup as eg:
    print(f"异常组: {eg.message}")
    for i, e in enumerate(eg.exceptions):
        print(f"  {i+1}. {type(e).__name__}: {e}")

15.7.2 Python 3.11的异常注释

Python 3.11为异常添加了add_note()方法，可以在不修改异常类的情况下添加上下文信息：

def process_data(data):
    try:
        result = int(data["value"])
    except (KeyError, ValueError) as e:
        e.add_note(f"处理数据时出错: {data}")
        e.add_note(f"可用键: {list(data.keys())}")
        raise

try:
    process_data({"name": "test"})
except Exception as e:
    print(f"异常: {e}")
    print(f"注释: {e.__notes__}")

15.7.3 结构化日志与异常

现代应用使用结构化日志记录异常，便于日志分析系统处理：

import logging
import json
from datetime import datetime

class StructuredLogger:
    def __init__(self, name):
        self.logger = logging.getLogger(name)

    def error(self, message, **kwargs):
        record = {
            "timestamp": datetime.now().isoformat(),
            "level": "ERROR",
            "message": message,
            **kwargs
        }
        self.logger.error(json.dumps(record, ensure_ascii=False))

    def exception(self, message, **kwargs):
        import traceback
        record = {
            "timestamp": datetime.now().isoformat(),
            "level": "ERROR",
            "message": message,
            "traceback": traceback.format_exc(),
            **kwargs
        }
        self.logger.error(json.dumps(record, ensure_ascii=False, default=str))

struct_logger = StructuredLogger("app")

try:
    result = 1 / 0
except ZeroDivisionError as e:
    struct_logger.exception(
        "计算错误",
        operation="division",
        error_type=type(e).__name__
    )

15.7.4 函数式异常处理

借鉴Rust/Scala的Result类型，在Python中实现函数式错误处理：

from dataclasses import dataclass
from typing import TypeVar, Generic, Callable

T = TypeVar("T")
E = TypeVar("E")

@dataclass
class Ok(Generic[T]):
    value: T

    def is_ok(self):
        return True

    def is_err(self):
        return False

    def unwrap(self):
        return self.value

    def map(self, func):
        return Ok(func(self.value))

@dataclass
class Err(Generic[E]):
    error: E

    def is_ok(self):
        return False

    def is_err(self):
        return True

    def unwrap(self):
        raise ValueError(f"Called unwrap on Err: {self.error}")

    def map(self, func):
        return self

Result = Ok[T] | Err[E]

def safe_divide(a, b) -> Result:
    if b == 0:
        return Err(ZeroDivisionError("除零错误"))
    return Ok(a / b)

result = safe_divide(10, 3)
print(f"成功: {result.is_ok()}, 值: {result.unwrap() if result.is_ok() else None}")

result2 = safe_divide(10, 0)
print(f"成功: {result2.is_ok()}, 错误: {result2.error if result2.is_err() else None}")

def safe_parse(value: str) -> Result:
    try:
        return Ok(int(value))
    except ValueError as e:
        return Err(e)

results = [safe_parse(x) for x in ["42", "abc", "100", "xyz"]]
oks = [r.unwrap() for r in results if r.is_ok()]
errs = [r.error for r in results if r.is_err()]
print(f"成功: {oks}")
print(f"失败: {errs}")

15.8 本章小结

本章深入探讨了Python异常处理机制：

1. 异常基础：异常传播机制、try/except/else/finally的完整语义、异常对象内省
2. 异常层次：内置异常体系、匹配规则、从具体到通用的捕获顺序
3. 异常链：显式链（raise ... from）、隐式链（__context__）、抑制链（from None）
4. 自定义异常：异常层次设计、错误码关联、异常聚合
5. 上下文管理器：__enter__/__exit__协议、contextlib工具、ExitStack
6. 最佳实践：捕获具体异常、不忽略异常、最小化try范围、资源清理保证
7. 前沿动态：ExceptionGroup、异常注释、结构化日志、函数式异常处理

15.9 习题与项目练习

基础习题

1. 异常分析：分析以下代码的输出顺序，并解释原因：

   def func():
       try:
           return "try"
       finally:
           return "finally"
   print(func())

2. 自定义异常：为一个在线商店设计异常层次结构，包括：库存不足、支付失败、配送错误等，每个异常应包含足够的上下文信息。

3. 上下文管理器：实现一个@timeout(seconds)装饰器，使用上下文管理器限制函数执行时间（提示：使用signal模块或线程）。

进阶习题

4. 异常链实践：实现一个多层API调用链，每层捕获底层异常并转换为更高层的业务异常，使用raise ... from保留异常链。

5. 重试装饰器：实现一个@retry装饰器，支持：

   • 最大重试次数
   • 指数退避
   • 只重试特定异常
   • 重试前的回调函数
   • 重试耗尽后抛出聚合异常

6. Result类型：完善Result类型实现，添加and_then（链式操作）、or_else（错误恢复）、unwrap_or（默认值）等方法。

项目练习

7. 统一错误处理框架：为Web API设计一个统一的错误处理框架，包括：

   • 异常到HTTP状态码的映射
   • 错误响应格式标准化（RFC 7807）
   • 请求ID追踪
   • 结构化日志记录
   • 错误监控集成（Sentry/Prometheus）
   • 开发/生产环境的不同行为

8. 事务管理器：实现一个通用的事务管理器，支持：

   • 多资源事务（数据库、文件、网络）
   • 两阶段提交协议
   • 嵌套事务（Savepoint）
   • 超时与死锁检测
   • 事务日志与恢复

15.10 延伸阅读

15.10.1 异常处理理论

• 《Exception Handling: Issues and a Proposed Notation》 (Goodenough, 1975) — 异常处理理论基础
• PEP 3134 — Exception Chaining and Embedded Tracebacks (https://peps.python.org/pep-3134/) — 异常链设计
• PEP 654 — Exception Groups and except* (https://peps.python.org/pep-0654/) — 异常组设计

15.10.2 上下文管理器

• PEP 343 — The “with” Statement (https://peps.python.org/pep-0343/) — with语句设计
• contextlib官方文档 (https://docs.python.org/3/library/contextlib.html) — 上下文管理工具
• 《Fluent Python》第18章 — 上下文管理器深度解析

15.10.3 错误处理最佳实践

• 《Effective Python》第3版 — 异常处理最佳实践
• 《Clean Code》第7章 — 错误处理原则
• Python异常处理指南 (https://docs.python.org/3/tutorial/errors.html) — 官方教程

15.10.4 函数式错误处理

• Rust Result类型 (https://doc.rust-lang.org/std/result/) — 函数式错误处理范式
• Scala Either类型 (https://www.scala-lang.org/api/current/scala/util/Either.html) — 函数式错误处理
• returns库 (https://github.com/dry-python/returns) — Python函数式错误处理库

---

下一章：第16章 并发编程