异常、上下文管理与资源安全

异常是函数契约的一部分。可靠程序不会捕获所有错误然后继续，而是只处理自己真正理解的失败，并确保文件、连接和锁在任何路径上都能释放。

一、学习目标

区分语法错误、程序缺陷和可预期失败
捕获具体异常并保留异常链
设计领域异常
使用 with 管理资源
理解事务式“全部成功或全部失败”

二、捕获具体异常

def parse_task_id(raw: str) -> int:
    try:
        task_id = int(raw)
    except ValueError as exc:
        raise ValueError("任务 ID 必须是整数") from exc
    if task_id < 1:
        raise ValueError("任务 ID 必须大于 0")
    return task_id

raise ... from exc 保留原始原因，traceback 会展示清楚的异常链。

不要默认使用：

try:
    run()
except Exception:
    pass

它会吞掉编程错误、数据损坏和关键诊断信息。

三、else 与 finally

try:
    task_id = int(raw)
except ValueError:
    print("格式错误")
else:
    print(f"准备读取任务 {task_id}")
finally:
    print("解析结束")

else 仅在 try 没有异常时执行。
finally 无论是否异常都会执行，适合必须完成的清理。

尽量缩小 try 范围，否则难以判断究竟是哪一行产生了被捕获的异常。

四、领域异常

class TaskError(Exception):
    """任务业务错误基类。"""


class TaskNotFoundError(TaskError):
    def __init__(self, task_id: int) -> None:
        super().__init__(f"任务 {task_id} 不存在")
        self.task_id = task_id

应用入口可以统一捕获 TaskError 并转换为用户消息；意外异常继续暴露或交给顶层日志系统。

异常类应表达调用方能采取的动作。不要为每一句错误文案建立一个类型。

五、上下文管理器

from pathlib import Path

path = Path("tasks.txt")
with path.open("w", encoding="utf-8") as file:
    file.write("阅读 Python 文档\n")

离开 with 块时文件会关闭，即使写入过程中抛出异常。

自定义上下文管理器：

from contextlib import contextmanager
from collections.abc import Iterator


@contextmanager
def operation(name: str) -> Iterator[None]:
    print(f"开始：{name}")
    try:
        yield
    finally:
        print(f"结束：{name}")


with operation("导入任务"):
    print("处理中")

六、资源所有权

谁创建资源，谁通常负责关闭它：

class TaskRepository:
    def __init__(self, connection) -> None:
        self._connection = connection

    def close(self) -> None:
        self._connection.close()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback) -> None:
        self.close()

调用：

with TaskRepository(connect()) as repository:
    repository.list()

如果连接由外部注入，是否由仓储关闭必须在接口中明确，避免重复关闭或泄漏。

七、原子写文件

直接覆盖文件时，程序中断可能留下半份内容。可先写临时文件再替换：

import os
from pathlib import Path
from tempfile import NamedTemporaryFile


def atomic_write(path: Path, content: str) -> None:
    with NamedTemporaryFile(
        "w",
        encoding="utf-8",
        dir=path.parent,
        delete=False,
    ) as file:
        temporary = Path(file.name)
        file.write(content)
        file.flush()
        os.fsync(file.fileno())
    try:
        temporary.replace(path)
    except Exception:
        temporary.unlink(missing_ok=True)
        raise

同一文件系统中的替换通常是原子的，但跨文件系统、权限和平台行为仍需测试。

八、异常组

并发任务可能同时失败，Python 可用 ExceptionGroup 表示多个异常，并用 except* 分类处理：

try:
    raise ExceptionGroup(
        "导入失败",
        [ValueError("标题为空"), OSError("文件不可读")],
    )
except* ValueError as group:
    for error in group.exceptions:
        print(f"数据错误：{error}")
except* OSError as group:
    for error in group.exceptions:
        print(f"系统错误：{error}")

日常顺序代码不必主动使用异常组；它主要出现在结构化并发等“一次等待多个任务”的边界。

九、常见错误

捕获 BaseException：会截获 KeyboardInterrupt、SystemExit 等控制信号。
用 finally 中的 return 覆盖原异常。
记录一次错误后又在每层重复记录，导致日志噪声。
把所有异常都转换成同一个“操作失败”，丢失调用方可用的信息。

十、练习与自测

为任务导入实现 InvalidTaskError，保留行号和原始异常。
写一个上下文管理器统计操作耗时。
模拟原子写入失败，确认临时文件被清理、原文件未破坏。

自测：

为什么应缩小 try 的范围？
raise ... from ... 解决什么问题？
上下文管理器如何保证异常路径也释放资源？

十一、官方资料

上一篇：类、协议与 Python 数据模型｜下一篇：迭代器、生成器与装饰器