Python 与数据实践 · 第 13 章

异常体系与处理

程序只要在跑，就免不了出岔子：文件找不到、网络断了、用户输入了奇怪的东西……出错本身并不可怕，可怕的是程序直接躺平——屏幕上甩出一大坨红色 traceback，然后整个进程退场。Python 为此准备了一整套"异常处理"…

约 16 分钟6,254 字16 个代码示例

程序只要在跑，就免不了出岔子：文件找不到、网络断了、用户输入了奇怪的东西……出错本身并不可怕，可怕的是程序直接躺平——屏幕上甩出一大坨红色 traceback，然后整个进程退场。Python 为此准备了一整套"异常处理"机制：既能优雅地接住错误、就地恢复，也能在必要的时候主动制造错误，把错误的原因一层层传递下去。

上一章拆解 with 语句时你应该已经注意到了，它底层翻译出来赫然写着 try / except / raise。本章就把这套语法正式讲透——它是后面写健壮的数据处理脚本的地基。

1. 接住异常：try / except 全家桶

1.1 最基本的接法

把"可能出事"的代码圈进 try，把"出了事怎么办"写进 except：

try:
    count = int("十二")  # 这行必然出事
except ValueError:
    print("这不是一个能转成整数的字符串")

如果没有 try，int("十二") 会当场把程序打死；现在它只是被拦下来，打印一句话之后，程序还活着。

1.2 多路拦截，顺便拿到异常对象

不同的问题对应不同的异常类型，可以写多个 except 分支分别对待；再加个 as，还能把异常对象本身抓在手里：

try:
    ratio = float("95%")
except ValueError as err:
    print(f"内容转不成数字: {err}")
except TypeError as err:
    print(f"类型根本不对: {err}")
else:
    print("一路平安，才轮到我执行")
finally:
    print("不管上面闹成什么样，我都会执行")

四个关键字各司其职：

try：圈出"危险区"，区内的代码一抛异常就立刻跳走；
except：按类型"认领"异常，命中哪个分支就走哪个；
else：try 里从头到尾没出过任何事，才会执行；
finally：收尾工作，出不出事都执行（关文件、释放连接这类事放这里最合适）。

1.3 异常对象身上有什么

try:
    top = [10, 20, 30][5]
except IndexError as err:
    print(err.args)           # ('list index out of range',) —— 构造异常时传入的参数
    print(err.__traceback__)  # 堆栈跟踪对象，记录异常是在哪一行冒出来的

2. 内置异常是一棵家族树

Python 的内置异常不是一盘散沙，而是一棵继承树。捕获父类，等于把它的全部子孙一网打尽：

BaseException
 ├── SystemExit          # sys.exit() 触发
 ├── KeyboardInterrupt   # Ctrl+C 触发
 └── Exception           # 日常异常的总祖先
      ├── ArithmeticError
      │    └── ZeroDivisionError
      ├── LookupError
      │    ├── IndexError
      │    └── KeyError
      ├── TypeError
      ├── ValueError
      │    └── UnicodeError
      └── ...

这张图透露出两条实用信息。

其一，父类可以代收子类。 当你并不在乎"到底是哪种查不到"时，拦父类最省事：

profile = {"name": "Alex"}
try:
    print(profile["age"])
except LookupError:
    print("没查到这个键")

KeyError 是 LookupError 的子类，所以这一个分支就能把它接住。

其二，拦 Exception 等于拦下几乎所有常规异常，写起来最省心：

try:
    run_migration()  # 什么异常都可能冒出来
except Exception as err:
    print(f"迁移失败: {err}")

但省心的代价是"无差别捕获"：连本该暴露出来的 bug 也被一并吞掉了。原则上，能拦多窄就拦多窄，只处理你确实有能力处理的异常；其余的让它炸出来，交给上层调用者或日志去管。

另外注意 SystemExit 和 KeyboardInterrupt 挂在 Exception 之外——所以 except Exception 拦不住它们，你按 Ctrl+C 依然能终止程序，这是刻意的设计。

3. 主动抛出：raise

异常不光能被动接，还能主动制造。当函数发现调用者传来的参数不讲道理时，最好的做法不是默默返回一个错误值，而是 raise 一个异常，把问题大声嚷出来：

def deduct_stock(stock: int, ordered: int) -> int:
    """扣减库存，返回剩余数量"""
    if ordered <= 0:
        raise ValueError("下单数量必须是正数")
    if ordered > stock:
        raise ValueError("库存不足，无法下单")
    return stock - ordered


try:
    deduct_stock(3, 5)
except ValueError as err:
    print(err)  # 库存不足，无法下单

跟返回 None、返回 -1 之类的"暗号"相比，raise 的好处一目了然：错误带类型、带说明，调用方想装看不见都难。

3.1 转手再抛：光杆 raise

有时接住异常只是为了记一笔日志、做点善后，处理完还打算让它继续往上走。这时用不带参数的 raise，把手里正处理着的异常原样抛回去：

try:
    fetch_remote_stats()
except Exception:
    print("抓取远端数据失败，先记一笔日志")
    raise  # 原封不动地重新抛出

注意光杆 raise 只在 except 块里有效，它抛的就是当前正在处理的那个异常，堆栈信息一点不丢。

4. 打造你自己的异常

内置异常终究是通用词汇。真实项目里，我们更希望异常能"自报家门"——一看到异常名就知道是哪块业务出了问题。做法很简单：继承 Exception。

class InvalidCodeError(Exception):
    """编号本身不合法"""
    pass


class CourseMissingError(Exception):
    """编号在系统里查无此课"""

    def __init__(self, course_name: str, code: int):
        self.course_name = course_name
        self.code = code
        super().__init__(f"{course_name}（编号 {code}）不在课程表中")

CourseMissingError 的写法值得细看：重写 __init__，把业务数据（课程名、编号）存进实例属性，再调用 super().__init__ 把拼好的消息交给父类。这样一来，print(err) 有可读的输出；except 住之后，还能直接取 err.code 继续编程。

course_table = {
    101: "Python 与数据实践",
    102: "统计思维入门",
}


def lookup_course(code: int) -> str:
    if code <= 0:
        raise InvalidCodeError("课程编号必须是正整数")
    if code not in course_table:
        raise CourseMissingError("未命名课程", code)
    return course_table[code]

5. 异常链：把"前因"挂在"后果"上

在 except 块里再抛一个新异常，是极其常见的操作——底层错误往往要"翻译"成上层语言再往上报告。可新异常抛出去之后，最初那个错误去哪了？Python 的答案是：把它们串成链，并且提供了两种串法。

def relay_implicitly():
    """串法一：直接抛新异常（隐式关联）"""
    try:
        raise KeyError("原始故障：缓存键失效")
    except KeyError:
        # 这里可以先做一点补救动作。
        raise RuntimeError("上层故障：页面渲染中断")  # __context__ 自动记录


def relay_explicitly():
    """串法二：用 from 手动指定因果（显式关联）"""
    try:
        raise KeyError("原始故障：缓存键失效")
    except KeyError as err:
        # 做了一点补救动作。
        raise RuntimeError("上层故障：页面渲染中断") from err  # __cause__ 明确记录

两种串法的差别，藏在异常对象的两个属性里：

try:
    relay_implicitly()
except RuntimeError as err:
    print("隐式关联:")
    print(f"  __cause__:   {err.__cause__}")   # None —— 没声明过因果关系
    print(f"  __context__: {err.__context__}") # KeyError —— 上下文自动记录

try:
    relay_explicitly()
except RuntimeError as err:
    print("显式关联:")
    print(f"  __cause__:   {err.__cause__}")   # KeyError —— from 指认的"元凶"
    print(f"  __context__: {err.__context__}") # KeyError —— 上下文同样有记录

规律可以这样记：

在 except 块里抛异常，Python 会自动把"当时正在处理的异常"记到 __context__ 上，这叫隐式关联；
写 raise 新异常 from 旧异常，旧异常会被记到 __cause__ 上，明确宣告"它是因、我是果"，这叫显式关联。

终端里那句 During handling of the above exception, another exception occurred: 就是隐式链在说话；而 The above exception was the direct cause of the following exception: 则是显式链。排查问题时顺着链条往下挖，往往一挖一个准。

习题

一、容错的比值计算

def success_rate(hits, trials):
    """
    计算命中率，要求：
    1. trials 为 0 时捕获 ZeroDivisionError，返回 0
    2. 参数类型不对时捕获 TypeError，打印"统计数字类型不对"并返回 None
    """
    # 你的代码
    pass


print(success_rate(8, 10))    # 0.8
print(success_rate(3, 0))     # 0
print(success_rate(8, "10"))  # 统计数字类型不对，None

二、自动重试装饰器 retry

import time


def retry(times, wait=1):
    """
    自动重试装饰器
    被装饰的函数若抛出异常，先等待 wait 秒再重试；最多尝试 times 次，
    全部失败则把最后一次的异常抛出去
    """
    # 你的代码
    pass


@retry(times=3, wait=1)
def pull_sensor_data():
    """模拟一个时好时坏的数据采集接口"""
    import random
    if random.random() < 0.7:  # 70% 概率失败
        raise ConnectionError("采集器无响应")
    return "采集成功"


# 应当能扛住失败并自动重试，最终返回"采集成功"，或在三次全败后抛出异常

提示： 带参数的装饰器怎么写，可以回看第 10 章；本题的关键在于"最后一次失败就别再吞了，用光杆 raise 抛出去"。

三、网盘配额管理

class QuotaExceededError(Exception):
    """存储配额已用尽"""
    pass


class AccountLockedError(Exception):
    """账号处于锁定状态"""
    pass


class CloudDrive:
    def __init__(self, quota=100, used=0, locked=False):
        self.quota = quota
        self.used = used
        self.locked = locked

    def upload(self, size):
        """
        上传文件，要求：
        1. locked=True 时，抛出 AccountLockedError
        2. size <= 0 时，抛出 ValueError
        3. used + size > quota 时，抛出 QuotaExceededError
        """
        # 你的代码
        pass

    def remove(self, size):
        """
        删除文件以释放空间，要求：
        1. locked=True 时，抛出 AccountLockedError
        2. size <= 0 时，抛出 ValueError
        """
        # 你的代码
        pass


# 测试
drive = CloudDrive(quota=100)
drive.upload(30)  # used = 30
drive.remove(10)  # used = 20

# drive.upload(200)       # QuotaExceededError
# drive.upload(-5)        # ValueError

locked_drive = CloudDrive(quota=100, locked=True)
# locked_drive.upload(10)  # AccountLockedError

四、把杂音翻译成同一种异常

实现一个函数，把数据采集环节可能冒出来的各种异常，统一转换成 PipelineError：

class PipelineError(Exception):
    def __init__(self, code, message):
        self.code = code
        self.message = message
        super().__init__(message)


def read_sensor():
    """模拟读取传感器，可能抛出各种异常"""
    import random
    failures = [
        ValueError("字段格式不合法"),
        ConnectionError("采集器掉线"),
        TimeoutError("读取超时"),
        RuntimeError("固件内部崩溃"),
    ]
    raise random.choice(failures)


def guarded_read():
    """
    调用 read_sensor()，把异常统一翻译成 PipelineError：
    - ValueError → PipelineError(400, "输入数据不合法")
    - ConnectionError / TimeoutError → PipelineError(503, "采集服务暂不可用")
    - 其他异常 → PipelineError(500, "采集服务内部错误")
    """
    # 你的代码
    pass

提示： 转换时用 from 把原始异常挂到链上，报错时能看到完整的来龙去脉。

下一章：14 · 类型标注