Python contextlib.ExitStack 怎么管理动态资源：批量打开与异常回收

批量导入文件时，输入目录往往不是固定的：有多少个文件、哪些文件能打开，都是运行到一半才知道。把每个文件都手写一层 try/finally 很快就会变成嵌套地狱。contextlib.ExitStack 适合处理这种动态资源，它会登记已经成功进入的资源，并在退出时按相反顺序清理。

记住两个动作就够了：能作为上下文管理器的资源用 enter_context 登记，需要执行的释放函数用 callback 登记；中途失败时，已经登记的资源仍会被可靠回收。

先做一个可验收的批量读取小项目

假设目录里有一批 UTF-8 文本文件，程序要逐个打开、读取首行，并在任意文件打不开时停止。目标不是吞掉错误，而是保证已经打开的文件不会因为后续失败而遗留。

from contextlib import ExitStack
from pathlib import Path

def read_first_lines(folder: str) -> list[str]:
    paths = sorted(Path(folder).glob("*.txt"))
    lines: list[str] = []
    with ExitStack() as stack:
        files = [stack.enter_context(path.open(encoding="utf-8")) for path in paths]
        for file in files:
            lines.append(file.readline().rstrip("\n"))
    return lines

这里的关键不是把文件对象收集到列表里，而是每次 enter_context 成功后，文件的退出动作已经进入栈。with ExitStack() 结束时，文件会被逐个关闭。

为什么动态资源要交给 ExitStack

固定两个文件时，两个嵌套的 with 很直观；但文件数量来自目录扫描、配置或用户输入时，嵌套层数无法提前写死。ExitStack 把“进入资源”和“离开资源”拆成了可循环执行的操作。

只接管成功进入的资源

如果第三个文件打开失败，可以把已经登记的资源理解为 resource-1、resource-2，而未成功进入栈的 resource-3 不会被登记；退出时前两个资源仍会被关闭，自然也不会凭空执行一次关闭动作。这一点比手写一个“所有文件都关闭”的 finally 更安全。

清理顺序是后进先出

资源之间有依赖时，后打开的资源通常应该先释放。ExitStack 按登记的逆序执行退出回调，符合栈的基本规则。不要把清理顺序理解成目录排序顺序，真正决定顺序的是进入栈的先后。

把临时目录和自定义清理动作放进同一条链

ExitStack 不只管理文件。下面把临时目录的清理动作登记到栈里，再加入一个需要参数的自定义收尾函数。注意 callback 接收的是“退出时调用的函数和参数”，不是现在就执行函数。

from contextlib import ExitStack
from pathlib import Path
import shutil
import tempfile

def remove_tree(path: Path) -> None:
    shutil.rmtree(path, ignore_errors=True)

def prepare_workspace() -> Path:
    stack = ExitStack()
    workspace = Path(tempfile.mkdtemp(prefix="import-"))
    stack.callback(remove_tree, workspace)
    try:
        (workspace / "manifest.txt").write_text("ready\n", encoding="utf-8")
        return workspace
    except Exception:
        stack.close()
        raise

这个函数展示了另一个边界：如果把栈交给调用方继续管理，可以调用 pop_all() 转移清理责任；如果不转移，就必须在确定的退出路径调用 close()。日常代码中，优先把栈放进 with，只有确实要延长资源寿命时才手动转移。

运行检查：分别验证成功和失败路径

准备两个文本文件后调用 read_first_lines，成功路径应该返回按文件名排序的首行列表。再把其中一个文件改成无法按 UTF-8 解码的内容，程序应抛出解码异常，但已经打开的文件仍在离开 ExitStack 时关闭。

from pathlib import Path

root = Path("sample-input")
root.mkdir(exist_ok=True)
(root / "a.txt").write_text("alpha\n", encoding="utf-8")
(root / "b.txt").write_text("beta\n", encoding="utf-8")

print(read_first_lines(str(root)))
# ['alpha', 'beta']

排查时不要只看“有没有异常”。更有价值的检查是：异常前已经成功登记了哪些资源、退出后临时目录是否消失、返回值是否只包含成功读取的数据。若清理动作依赖外部服务，回调内部还要记录失败并决定是否需要补偿。

常见误区与适用边界

把 callback 当成立即执行

stack.callback(remove_tree, workspace) 只是登记动作。若把它误写成 stack.callback(remove_tree(workspace))，清理会提前发生，并把返回值当作回调对象。

资源已经转移后还继续 close

pop_all() 会把回收责任转给新的栈。转移完成后，原栈不再拥有这些回调；代码应明确谁负责最终 close()，不要让两个函数都以为自己拥有清理权。

不需要动态组合时硬套 ExitStack

只有一个文件时，普通 with open(...) 更清晰。ExitStack 的价值在于资源数量、进入时机或清理动作需要动态组合；为了“看起来高级”而使用它，反而会增加阅读成本。

延伸问答：什么时候该选 ExitStack

ExitStack 能替代所有 finally 吗？

不能。它擅长统一管理上下文退出和延迟回调；如果清理逻辑需要根据业务结果分支，仍应在回调或显式 finally 中写清判断。

异常会不会被 ExitStack 吞掉？

默认不会。退出回调完成后，原异常仍会继续向外传播；只有清理回调自己抛出异常时，异常链才需要额外记录和处理。

什么时候应该用 AsyncExitStack？

当资源是异步上下文管理器，或者释放动作需要 await 时，应使用 contextlib.AsyncExitStack，不要在同步栈里强行运行异步清理。

把验收标准写进小项目

这个小项目的验收并不复杂：成功时首行列表正确，失败时异常可见，任意已经进入栈的文件都完成关闭，临时目录在责任结束后被清理。满足这四点，再把 ExitStack 推广到数据库连接、锁和临时文件组合中，代码才会真正变简单。