Python ExitStack 实战：动态文件资源与失败清理怎样安排

要合并的文件来自命令行参数，可能两份，也可能二十份。第一份打开成功，第二份也成功，第三份不存在：这时最容易漏掉的不是那次 open() 报错，而是前面已经拿到的文件句柄归谁关闭。

固定数量的资源用嵌套 with 就够了。数量由输入决定，再写一排 try/finally 或维护“已打开列表”，清理规则很容易与主流程分家。ExitStack 适合处理的正是这个变化：每拿到一份资源，就把退出动作登记进去；后面任何一步失败，已经登记的动作仍按逆序执行。

下面用 Python 3.12 标准库完成一个小型合并工具。它接受多份 UTF-8 文本，先写同目录临时结果，全部读取成功后才替换目标文件。输入打开失败、解码失败时，旧结果保留，临时目录也要清掉。

两层退出栈，各自拥有不同东西

输入句柄与临时目录不能不分先后地释放。文件还开着就删除临时目录，在某些平台会失败；临时输出还没关闭就替换目标文件，同样会把平台差异带进程序。

这里分成两个作用域：

• 外层拥有临时目录，离开整个发布流程时删除它。
• 内层拥有输入和临时输出句柄，写完就全部关闭。

替换操作位于两个作用域之间：文件已经关闭，临时目录仍存在。这比先写满一串布尔标志，再到 finally 里猜到底进行到了哪一步清楚。

完整保存为 merge_texts.py：

import argparse
import os
import sys
import tempfile
from contextlib import ExitStack
from pathlib import Path


def merge_texts(inputs, target):
    target = Path(target).resolve()
    paths = [Path(item).resolve() for item in inputs]
    if not paths:
        raise ValueError("at least one input is required")
    if len(paths) > 64:
        raise ValueError("this tool accepts at most 64 input files")
    if target in paths:
        raise ValueError("output must not also be an input")
    if not target.parent.is_dir():
        raise ValueError("output directory does not exist")

    line_count = 0
    with ExitStack() as cleanup:
        work = Path(cleanup.enter_context(tempfile.TemporaryDirectory(
            prefix=".merge-", dir=target.parent
        )))
        staged = work / "result.txt"

        with ExitStack() as handles:
            streams = [
                handles.enter_context(path.open(
                    "r", encoding="utf-8", errors="strict", newline=""
                ))
                for path in paths
            ]
            output = handles.enter_context(staged.open(
                "w", encoding="utf-8", newline=""
            ))
            for stream in streams:
                for line in stream:
                    # 定义输出协议：每条输入行都以一个 LF 结束。
                    text = line.rstrip("\r\n")
                    output.write(text + "\n")
                    line_count += 1

        # 所有句柄已经关闭；此时才让新结果对调用者可见。
        os.replace(staged, target)

    return line_count


def main(argv=None):
    parser = argparse.ArgumentParser(allow_abbrev=False)
    parser.add_argument("inputs", nargs="+", type=Path)
    parser.add_argument("-o", "--output", required=True, type=Path)
    args = parser.parse_args(argv)
    try:
        count = merge_texts(args.inputs, args.output)
    except (OSError, UnicodeError, ValueError) as error:
        print(f"merge failed: {error}", file=sys.stderr)
        return 1
    print(f"merged lines: {count}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

输出协议刻意限定为文本行合并，而非字节拼接。输入最后一行没有换行，也会补一个 LF；CRLF 与 CR 会被统一成 LF。对要求原字节保真的文件，这个工具并不合适，不能为了演示资源管理把数据语义顺便改了却不说明。

代码还限制一次最多 64 个输入，因为这里会同时打开所有输入。ExitStack 能把关闭动作写正确，却不会提高进程允许打开的文件描述符数量。处理数千份顺序输入时，更合适的实现是逐个文件打开、读完即关闭；动态资源栈不是必须套到所有循环上的写法。

把失败也做成可以运行的样本

再保存 check_merge.py，与前一个文件放在同一目录：

from pathlib import Path
from tempfile import TemporaryDirectory

from merge_texts import merge_texts


with TemporaryDirectory() as directory:
    root = Path(directory)
    first = root / "first.txt"
    second = root / "second.txt"
    broken = root / "broken.txt"
    target = root / "merged.txt"
    first.write_bytes("甲\r\n乙".encode("utf-8"))
    second.write_bytes("丙\n".encode("utf-8"))
    broken.write_bytes(b"\xff\xfe")
    target.write_text("old\n", encoding="utf-8")

    count = merge_texts([first, second], target)
    assert count == 3
    assert target.read_bytes() == "甲\n乙\n丙\n".encode("utf-8")
    print("normal: 3 normalized lines")

    old = target.read_bytes()
    try:
        merge_texts([first, root / "missing.txt"], target)
    except FileNotFoundError:
        pass
    else:
        raise AssertionError("missing input was accepted")
    assert target.read_bytes() == old
    assert not list(root.glob(".merge-*"))
    print("missing input: old output kept, staging removed")

    try:
        merge_texts([first, broken], target)
    except UnicodeDecodeError:
        pass
    else:
        raise AssertionError("invalid UTF-8 was accepted")
    assert target.read_bytes() == old
    assert not list(root.glob(".merge-*"))
    print("bad encoding: old output kept, staging removed")

运行：

python3 check_merge.py
python3 merge_texts.py --help

三个断言组分别覆盖正常合并、打开到一半失败、写入到一半失败。后两种错误发生在不同阶段，却都不应让旧目标变成半份结果。测试检查临时目录已经消失，也能避免只盯着最后返回码。

若要观察句柄关闭顺序，可以给一个最小对象实现 __enter__ 与 __exit__，依次进入 A、B、C 后抛错，退出记录应为 C、B、A。实际程序不需要为了打印这个顺序再封装每个文件；栈的逆序退出规则已经由同一个结构表达。

callback 与 enter_context 不能随便互换

对象已经提供上下文管理协议时，优先 enter_context()。它先调用 __enter__，成功后再登记 __exit__；后来打开其他资源失败，之前成功进入的资源仍会退出。

一个只有 close() 方法、没有上下文协议的对象，可以在成功取得以后用 stack.callback(obj.close) 登记清理。这个回调不会收到异常类型、异常值和 traceback，也不能通过返回真值抑制原异常。若资源退出确实依赖异常信息，应使用完整上下文协议，而不是硬塞一个无参关闭函数。

反过来，某个资源自己的 __enter__ 内部拿到一半资源就失败，外层栈无法替它补上尚未登记的内部清理。获取函数本身仍要保证失败安全；退出栈不是替任意半成品构造函数擦屁股的运行时魔法。

pop_all() 则是在转移整组退出动作的所有权。调用之后原栈不再负责那批动作，新栈必须被关闭。仅为了“成功后别删文件”就把所有回调一起弹掉，很可能把文件句柄的关闭也取消了。上面的工具没有使用它，而是让临时目录一直归外层拥有、最终结果通过移动离开目录。这种资源关系更直观。

清理动作自身也可能失败。例如关闭输出时磁盘报错，或者临时目录没有删除权限，退出过程同样会抛异常。不要把清理回调全部包成静默忽略，否则工具可能对外报告成功，却留下未完成的写入或暂存文件。需要同时保留业务失败与清理失败时，应让异常链继续可见，再由最外层选择日志与告警方式。测试中的临时目录断言，只覆盖当前进程正常获得清理机会的情况；进程被强制终止以后，仍需另设可识别的暂存命名和回收策略。

替换成功和持久化完成仍是两个问题

临时文件建在目标目录中，是为了避免随手跨文件系统移动；os.replace() 成功后，读取者会通过目标路径看到新文件。但这个例子没有 fsync，没有承诺断电后新内容一定持久，也没有处理多个发布者争抢同一个目标的业务次序。

此外，替换目标的路径检查并不是安全沙箱。若目录由不可信用户并发修改，符号链接与路径竞态需要更严格的目录权限和平台能力。这里假定脚本在自己控制的工作目录处理输入，不把一个文本合并示例包装成任意文件访问服务。

ExitStack 真正让程序变简单的地方，是把“刚拿到的东西”与“退出时必须做的动作”紧挨着写出来。之后再看正常路径和失败路径，就不必同时记住十几个开关。资源什么时候关闭、结果什么时候发布，代码的缩进已经给出了答案。