第15章异常处理

学习目标

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

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

15.1 异常机制基础

15.1.1 异常的本质

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

python

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完整语法

python

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

python

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 多异常处理

python

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 异常对象与内省

python

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
    └── ... (更多)

python

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子句匹配异常时，会检查异常是否是指定类的实例或其子类的实例。因此，捕获父类异常会同时捕获所有子类异常：

python

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子句的顺序应从具体到通用，否则更具体的异常处理永远不会被执行：

python

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语法用于在捕获一个异常后抛出另一个异常，同时保留原始异常的上下文：

python

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__属性：

python

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抑制异常链：

python

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 异常链的最佳实践

python

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 设计异常层次结构

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

python

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 异常与错误码

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

python

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 异常聚合

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

python

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__）：

python

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：异常继续传播

python

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装饰器：

python

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：

python

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：

python

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：

python

from contextlib import redirect_stdout
import io

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

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

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

python

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 可重入与线程安全的上下文管理器

python

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 捕获具体异常

python

# 反模式：裸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 不要忽略异常

python

# 反模式：空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子句

python

# 反模式：把所有代码放在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 资源清理保证

python

# 反模式：手动清理
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 异常与断言的选择

python

# 断言：用于检查程序内部不变量，不应被捕获
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 异常处理装饰器

python

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*语法，用于处理多个同时发生的异常：

python

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()方法，可以在不修改异常类的情况下添加上下文信息：

python

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 结构化日志与异常

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

python

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中实现函数式错误处理：

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异常处理机制：

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

15.9 习题与项目练习

基础习题

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

python

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

自定义异常：为一个在线商店设计异常层次结构，包括：库存不足、支付失败、配送错误等，每个异常应包含足够的上下文信息。
上下文管理器：实现一个@timeout(seconds)装饰器，使用上下文管理器限制函数执行时间（提示：使用signal模块或线程）。

进阶习题

异常链实践：实现一个多层API调用链，每层捕获底层异常并转换为更高层的业务异常，使用raise ... from保留异常链。
重试装饰器：实现一个@retry装饰器，支持：
- 最大重试次数
- 指数退避
- 只重试特定异常
- 重试前的回调函数
- 重试耗尽后抛出聚合异常
Result类型：完善Result类型实现，添加and_then（链式操作）、or_else（错误恢复）、unwrap_or（默认值）等方法。

项目练习

统一错误处理框架：为Web API设计一个统一的错误处理框架，包括：
- 异常到HTTP状态码的映射
- 错误响应格式标准化（RFC 7807）
- 请求ID追踪
- 结构化日志记录
- 错误监控集成（Sentry/Prometheus）
- 开发/生产环境的不同行为
事务管理器：实现一个通用的事务管理器，支持：
- 多资源事务（数据库、文件、网络）
- 两阶段提交协议
- 嵌套事务（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章并发编程

第15章 异常处理 ​

学习目标 ​

15.1 异常机制基础 ​

15.1.1 异常的本质 ​

15.1.2 try-except完整语法 ​

15.1.3 多异常处理 ​

15.1.4 异常对象与内省 ​

15.2 异常层次结构 ​

15.2.1 内置异常体系 ​

15.2.2 异常匹配规则 ​

15.3 异常链 ​

15.3.1 显式链（raise ... from） ​

15.3.2 隐式链 ​

15.3.3 抑制链（raise ... from None） ​

15.3.4 异常链的最佳实践 ​

15.4 自定义异常 ​

15.4.1 设计异常层次结构 ​

15.4.2 异常与错误码 ​

15.4.3 异常聚合 ​

15.5 上下文管理器 ​

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

15.5.2 contextlib模块 ​

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

15.6 异常处理最佳实践 ​

15.6.1 捕获具体异常 ​

15.6.2 不要忽略异常 ​

15.6.3 使用else子句 ​

15.6.4 资源清理保证 ​

15.6.5 异常与断言的选择 ​

15.6.6 异常处理装饰器 ​

15.7 前沿技术动态 ​

15.7.1 Python 3.11的异常组与except* ​

15.7.2 Python 3.11的异常注释 ​

15.7.3 结构化日志与异常 ​

15.7.4 函数式异常处理 ​

15.8 本章小结 ​

15.9 习题与项目练习 ​

基础习题 ​

进阶习题 ​

项目练习 ​

15.10 延伸阅读 ​

15.10.1 异常处理理论 ​

15.10.2 上下文管理器 ​

15.10.3 错误处理最佳实践 ​

15.10.4 函数式错误处理 ​

第15章异常处理

学习目标

15.1 异常机制基础

15.1.1 异常的本质

15.1.2 try-except完整语法

15.1.3 多异常处理

15.1.4 异常对象与内省

15.2 异常层次结构

15.2.1 内置异常体系

15.2.2 异常匹配规则

15.3 异常链

15.3.1 显式链（raise ... from）

15.3.2 隐式链

15.3.3 抑制链（raise ... from None）

15.3.4 异常链的最佳实践

15.4 自定义异常

15.4.1 设计异常层次结构

15.4.2 异常与错误码

15.4.3 异常聚合

15.5 上下文管理器

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

15.5.2 contextlib模块

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

15.6 异常处理最佳实践

15.6.1 捕获具体异常

15.6.2 不要忽略异常

15.6.3 使用else子句

15.6.4 资源清理保证

15.6.5 异常与断言的选择

15.6.6 异常处理装饰器

15.7 前沿技术动态

15.7.1 Python 3.11的异常组与except*

15.7.2 Python 3.11的异常注释

15.7.3 结构化日志与异常

15.7.4 函数式异常处理

15.8 本章小结

15.9 习题与项目练习

基础习题

进阶习题

项目练习

15.10 延伸阅读

15.10.1 异常处理理论

15.10.2 上下文管理器

15.10.3 错误处理最佳实践

15.10.4 函数式错误处理