Context Managers Beyond Opening Files in Python

Python中超越文件打开的上下文管理器 - MachineLearningMastery.com

超越文件打开的Python上下文管理器

By

Nahla Davies

on

2026年10月27日

in

实用Python

0

分享

发布

在本文中，你将学习Python上下文管理器在底层的工作原理，以及如何将其应用于超出文件处理的真实清理问题。

我们将涵盖的主题包括：

• with语句协议的工作方式，包括__enter__()和__exit__()实际执行的操作，以及__exit__()的返回值为何重要。

• 如何使用类、@contextmanager装饰器以及标准库工具（如TemporaryDirectory和patch.dict()）构建自定义上下文管理器。

• 如何使用ExitStack安全地管理运行时可变数量的资源，以及何时将其封装到自己的类中。

评估脚本在执行任何操作之前会进行大量准备工作。它为提取的测试集设置了临时目录，将环境变量指向更快的磁盘以定位模型缓存，将库日志记录器调整为WARNING级别（否则会淹没控制台），并启动了计时器。文件末尾会按照正确顺序撤销所有操作：删除目录、恢复变量、重置日志记录器并打印耗时。底部的清理操作在脚本成功运行时总会执行。当某天遇到格式错误的记录导致evaluate()中途抛出异常时，这些清理操作都不会执行。下一位使用该机器的人将面临过期的缓存路径、无法解释的静默日志记录器，以及3GB无人记得创建的提取文件。

scratch = Path("scratch"); scratch.mkdir(exist_ok=True) os.environ["MODEL_CACHE"] = "/fast/cache" logging.getLogger("transformers").setLevel(logging.WARNING) start = time.perf_counter() scores = evaluate(model, extract(archive, scratch)) # 在遇到错误记录时抛出异常 print(f"took {time.perf_counter() - start:.1f}s") # 上述代码不会执行 shutil.rmtree(scratch) del os.environ["MODEL_CACHE"] logging.getLogger("transformers").setLevel(logging.INFO)

1

2

3

4

5

6

7

8

9

10

11

12

scratch

=

Path

(

"scratch"

)

;

.

mkdir

exist_ok

True

os

environ

[

"MODEL_CACHE"

]

"/fast/cache"

logging

getLogger

"transformers"

setLevel

WARNING

start

time

perf_counter

scores

evaluate

model

,

extract

archive

raises on a bad record

print

f

"took {time.perf_counter() - start:.1f}s"

none of this runs

shutil

rmtree

del

INFO

大多数Python开发者通过with open(path) as f接触到上下文管理器，学习到文件会被关闭后就止步于此。文件是上下文管理器能处理的最不有趣的事物。with语句真正提供的是一个具有保证退出的代码块。初始化在顶部执行，清理在底部执行，无论中间部分是否成功或抛出异常，清理操作都会执行。如果即使代码块失败也需要执行清理，为何要将其放在块之外？评估脚本设置的这四个事项都是临时约定。每个都可以成为上下文管理器：不会忘记停止的计时器、会自行删除的目录、会自动恢复的狀態变更，以及按正确顺序释放的资源栈。这就是本文的演进过程，从with语句底层的协议，到标准库的快捷方式，再到选择它们的规则。

with语句是边界，而非文件技巧

语言参考将 with 语句描述为一个序列，这个序列值得记住。下面的每个模式都是它的变体。Python 会评估 with 后面的表达式以获取一个上下文管理器并调用其 __enter__() 方法。__enter__() 返回的任何内容都会绑定到 as 后面的名称，然后执行代码块，最后调用 __exit__()。最后一步是整个过程的核心。无论正常退出、从代码块内部返回，还是发生异常，这一过程都会发生，因此代码块中的任何路径都无法跳过 __exit__() 中编写的清理操作。

有两个细节容易让人困惑，且都与返回值有关。首先，as 后面的名称会接收 __enter__() 的返回值，这通常是管理器本身，但不一定是。例如 open() 返回文件对象，TemporaryDirectory() 返回路径字符串。其次，__exit__() 会接收三个参数用于描述异常，或在正常退出时接收三个 None 值，而它的返回值决定了后续操作。数据模型对此有明确说明：返回真值时异常会被抑制，返回 False 或未返回（等同于返回 None）时异常会在清理后继续传播。

class Announce: def __enter__(self): print("enter") return "the value bound after as" def __exit__(self, exc_type, exc_value, traceback): print("exit", exc_type) # 在正常退出时为 None return False # 允许任何异常继续传播

with Announce() as x: print(x)

class

Announce

:

def

__enter__

self

"enter"

return

"the value bound after as"

__exit__

exc_type

exc_value

traceback

"exit"

在正常退出时为 None

False

允许任何异常继续传播

with

as

x

抑制异常的规则是需要首先正确掌握的。如果搞错了，这种错误是静默的。任何以 return True 结尾的 __exit__() 方法都会使代码块内的所有异常消失，同样，返回值恰好为真值的清理操作也会产生相同效果。脚本会继续运行，但分数数据丢失了，本应解释原因的回溯信息在代码块边界被吞噬了。下面的自定义管理器从不从 __exit__() 返回任何真值。代码块内发生的任何问题都会继续作为调用者的责任，这才是它应有的归属。

一个图示说明的保证：代码块的两个退出路径都会经过 __exit__()，在异常路径上，__exit__() 的返回值是异常传递给调用者的唯一屏障。图片来源：基于 Python 语言参考中的 with 语句和数据模型的原始图表。

构建一个不会忘记停止的计时器

开头脚本中的秒表是这个模式的最小实现。两次调用 time.perf_counter() 包围了工作内容，而第二次调用正是异常跳过的典型行。一个包含 __enter__() 和 __exit__() 方法的类可以将第二次调用移动到无法被跳过的位置。

import time

class Timer: def __init__(self, label): self.label = label self.elapsed = None

def __enter__(self): self._start = time.perf_counter() return self

def __exit__(self, exc_type, exc_value, traceback): self.elapsed = time.perf_counter() - self._start status = "failed" if exc_type else "ok" print(f"{self.label}: {self.elapsed:.3f}s ({status})") return False # 从不吞没代码块的异常

with Timer("feature extraction") as t: features = extract(batch) print(t.elapsed) # 代码块结束后仍可读取

13

14

15

16

17

18

19

20

21

import

Timer

__init__

label

elapsed

None

_start

-

status

"failed"

if

else

"ok"

/

"{self.label}: {self.elapsed:.3f}s ({status})"

切勿吞没代码块中的异常

"特征提取"

t

特征

批次

块执行后仍可读取

该类中的两个选择完成了工作。__enter__() 返回计时器本身，这就是为什么在代码块执行完毕后可以读取 t.elapsed 的原因。__exit__() 仅通过检查 exc_type 来决定标签显示 ok 还是 failed。__exit__() 从不保留异常。因此，即使特征提取运行时发生异常，仍然会打印出 "特征提取: 12.418s (failed)" 并继续向堆栈跟踪传递异常。十二秒这个数字正是你真正关心的，当问题核心是批次转换是否卡住而非崩溃时。

perf_counter() 是此处的正确时钟。它在短时间间隔上具有最高分辨率，且在 CPython 中不会倒退。其绝对值没有意义，只有两次调用之间的差值才有意义，而该类通过设计隐藏了这一差值。不过要适度看待该类的野心。它仅测量单个进程中的一个代码块耗时并打印一行信息。这是监控手段，有助于发现预处理任务中真正耗时二十分钟的阶段。Timer 记录不是基准测试。当有人开始比较不同机器上的两个 Timer 记录时，下一步应使用 timeit 或性能分析器，而不是继续增加小数位数。

使用 @contextmanager 实现单一设置-执行-清理流程

当管理器需要保存调用者希望读取的状态时（如计时器），类是最合适的选择。当整个任务只是设置某物、执行代码块、然后恢复该物时，使用类就显得繁琐，此时 @contextmanager 是更简洁的方案。它接受一个生成器函数，并用相同的 __enter__() / __exit__() 保证包裹该函数，由生成器的结构完成工作：yield 之前的所有内容在进入时执行，yield 之后的所有内容在退出时执行。

import logging from contextlib import contextmanager @contextmanager def quiet(logger_name, level=logging.WARNING): logger = logging.getLogger(logger_name) previous = logger.level logger.setLevel(level) try: yield logger finally: logger.setLevel(previous) with quiet("transformers"): scores = evaluate(model, eval_set)

from

contextlib

contextmanager

@

quiet

logger_name

level

logger

previous

try

yield

finally

eval_set

开头的冗长日志器修复方案现在具备了在失败时仍能正常工作的形态。在修改日志级别前先捕获了原始级别，然后进行修改，代码块在 yield 时执行，最后 finally 会将级别恢复原状。文档中的两条规则确保了这一机制的正常运作。生成器必须恰好 yield 一个值，因为该 yield 对应代码块，若存在第二个 yield 会在退出时触发 RuntimeError。如果代码块抛出异常，该异常会在生成器的 yield 处重新抛出。这就是为什么清理操作必须放在 finally 而非 yield 之后的原因。若要捕获异常并记录日志，必须重新抛出异常，否则管理器会静默地成为异常抑制器。

装饰器还免费提供了另一个功能。@contextmanager 基于 ContextDecorator 实现，因此 quiet(...) 也可以作为装饰器置于函数定义上方，整个函数体即为代码块。Machine Learning Mastery 的生产级 Python 装饰器指南详细说明了这种写法适用的场景及可能隐藏过多细节的注意事项。

临时目录应自行清理残留文件

The scratch directory in the opening is the case the standard library already owns, and the fix is to stop owning it yourself. tempfile.TemporaryDirectory() creates a uniquely named directory and hands the path to the as target. When the block ends, on either path out, the directory and everything inside it are removed.

import tempfile from pathlib import Path # before: a fixed path, and a cleanup line that only runs on success scratch = Path("scratch") scratch.mkdir(exist_ok=True) scores = evaluate(extract(archive, scratch)) shutil.rmtree(scratch) # after: the directory's lifetime is the block with tempfile.TemporaryDirectory(prefix="eval-") as tmp: scores = evaluate(extract(archive, Path(tmp))) # tmp is gone here, whether evaluate() raised or not

tempfile

pathlib

before: a fixed path, and a cleanup line that only runs on success

after: the directory's lifetime is the block

TemporaryDirectory

prefix

"eval-"

tmp

tmp is gone here, whether evaluate() raised or not

The before version has three problems beyond the skipped cleanup. Two runs on the same machine share scratch/ . A leftover file from yesterday’s archive can pass as today’s data. A crash leaves the debris in the project tree, where it gets committed. The after version puts the artifacts in a path that is unique per run and outside the repository. It is also easier to inspect while the block is running: print tmp and open it in another shell. When a run needs its artifacts kept for a post-mortem, Python 3.12 added a delete=False parameter for exactly that, which beats commenting out an rmtree .

One platform note earns a sentence. On Windows, a file still open inside the directory makes removal fail with PermissionError at exit. Passing ignore_cleanup_errors=True , available since Python 3.10, downgrades that to a best-effort sweep, which is fine in a script and a bad habit in a library. The proper fix is to close what you opened before the outer block ends, usually by nesting a with for the file inside the one for the directory.

Combine Dynamic Resources With ExitStack

Every example so far manages one thing. The evaluation script’s harder relative is the merge step that opens one file per shard, and the number of shards is whatever the previous job produced. Nesting with statements needs the count at write time. A loop that calls open() and appends to a list needs a try / finally that closes each handle in reverse, and it must not lose an exception raised by the third close() while the second is already running. Closing a runtime-sized set of resources in the right order is the problem ExitStack exists for, and the documentation describes it as a way to programmatically combine other context managers and cleanup functions, especially those driven by input data.

from contextlib import ExitStack def merge_shards(paths, out_path): with ExitStack() as stack: shards = [stack.enter_context(open(p, encoding="utf-8")) for p in paths] out = stack.enter_context(open(out_path, "w", encoding="utf-8")) stack.callback(print, f"merged {len(paths)} shards into {out_path}") for lines in zip(*shards): out.writelines(lines)

ExitStack

merge_shards

paths

out_path

stack

shards

enter_context

open

p

encoding

"utf-8"

for

out

"w"

callback

"merged {len(paths)} shards into {out_path}"

lines

zip

*

writelines

enter_context() 调用管理器的 __enter__() 方法，并将其 __exit__() 注册到栈中。这种方法适用于任何上下文管理器，包括上面提到的 Timer 和 TemporaryDirectory。callback() 用于注册普通函数，适用于完全没有管理器的情况，例如设备释放或通知。普通回调无法抑制异常，因为它们永远无法看到异常细节。当代码块结束时，栈会逆序执行所有注册的内容，因此输出文件会首先关闭，接着是各个分片（最后打开的先处理）——这与手动嵌套的 with 语句产生的顺序完全一致。在生产环境中需要注意两个细节。栈的展开发生在代码块结束或调用 close() 时，而不是当栈对象被垃圾回收时，因此如果 ExitStack 超出其 with 块范围就会造成资源泄漏。此外，内部回调如果抑制了异常，会改变外部回调看到的异常内容，这与嵌套的 with 块行为一致。

一个诚实的警告：如果某段代码之所以能正常工作，仅仅是因为 ExitStack 维持着十几个句柄、锁和回调的关联，那么这说明缺少了某种抽象层。通常，一个拥有自己 __enter__() 和 __exit__() 方法的 ShardSet 类（内部基于 ExitStack 实现）会比循环写法更清晰。调用者此时看到的是一个生命周期统一的资源，而不是需要记住的一系列独立事物。

使状态变更可逆

开头提到的环境变量是一种状态变更而非资源，而状态变更正是最容易产生清理错误的地方，因为没有任何东西会崩溃。缓存路径只是简单地对下一个任务来说是错误的，且不会发出任何提示。对于任何字典结构的数据，标准库已经实现了反转操作，尽管它隐藏在 unittest.mock 的 patch.dict() 中。文档的描述正是这个功能的全部含义：修改字典，然后在代码块结束后将其恢复到原始状态。只要映射对象支持获取、设置和删除操作，任何类型的映射都可以使用。

import os from unittest.mock import patch

with patch.dict(os.environ, {"MODEL_CACHE": "/fast/cache", "TOKENIZERS_PARALLELISM": "false"}): scores = evaluate(model, eval_set)

MODEL_CACHE 恢复为旧值；如果之前未设置，TOKENIZERS_PARALLELISM 也会被重新取消设置

CONFIG = {"batch_size": 64, "device": "cuda"} with patch.dict(CONFIG, {"batch_size": 8}): smoke_test()

使用 batch_size=8 运行，之后 CONFIG 会被恢复

unittest mock patch dict { "TOKENIZERS_PARALLELISM": "false" }

MODEL_CACHE 恢复为旧值；如果之前未设置，TOKENIZERS_PARALLELISM 也会被重新取消设置

CONFIG "batch_size": 64 "device": "cuda" smoke_test

使用 batch_size=8 运行，之后 CONFIG 会被恢复

恢复操作是真正的恢复而非删除。原本存在的键会恢复为旧值，原本不存在的键会被移除。这种机制适用于功能标志、临时路径、测试凭证，以及原本可能需要全局设置的配置覆盖（一个函数设置，另一个函数忘记）。这个名称来源于测试模块，但其本身并不局限于测试场景。与手动编写 old = d.get(k) 的代码块相比，这种方法更不容易出错，因为后者需要自行处理键缺失的情况。

作用域是需要特别注意的地方。os.environ 是对当前进程环境变量的映射，在导入 os 模块时被捕获，每次更改时会通过 putenv() 写回。对它使用 patch.dict() 会改变当前进程以及在代码块执行期间创建的子进程所看到的环境变量。其他正在运行的进程不会察觉到变化。但在当前进程中，所有线程和协程会立即看到这个变化。正因如此，环境变量和当前工作目录是两种在并发环境下行为异常的临时状态。标准库自带的 contextlib.chdir() 也对此发出警告：它会修改全局状态，因此不适合大多数线程或异步代码，也不适合在它执行期间需要 yield 的生成器。当进程只执行单一任务时，应避免修改全局状态。当进程需要处理多个任务时，应将路径或设置作为参数传递。

在关键位置保持清理操作的可见性

选择规则很简单。当管理器需要保留供调用者后续使用的状态（如计时器的已过时间）时，应编写类。当任务是单一设置、单一代码块和单一清理操作（如日志级别设置）时，应使用 @contextmanager 生成器。当标准库已拥有生命周期管理时（如 TemporaryDirectory 管理临时文件，patch.dict 管理可逆映射），应使用标准库提供的管理器。当资源集合仅在运行时确定时，应使用 ExitStack，当它包含超过少量资源时，应将其封装到独立的类中。无论采用哪种形式，除非代码块在失败后需要继续执行，否则 __exit__() 必须返回 False 或 None。contextlib.suppress() 仅用于已知需要继续执行的特定错误。

模式 | 最佳匹配场景 | 避免使用场景 | 主要注意事项 --- | --- | --- | --- class + __enter__/__exit__ | Timer | 管理器包含调用者后续需要读取的状态，或拥有多个自有方法 | 仅需单一设置、单一代码块、单一清理操作，且无其他操作 | __exit__() 必须返回 False 或 None，除非明确意图是抑制异常 @contextmanager | quiet(logger) | 代码块结束后需要读取状态，或清理操作包含分支逻辑 | 仅允许一个 yield，已记录的异常必须重新抛出 内置管理器 | TemporaryDirectory<br>patch.dict | 标准库已拥有生命周期管理 | 生命周期与库的预期行为不一致（保留临时文件、部分恢复） | 修改全局状态（environ、cwd）在多线程或异步环境下是进程级别的危险操作 enter_context() | callback() | 资源数量仅在运行时确定 | 隐藏了本应存在的资源对象 | 在代码块结束或 close() 时释放，从不依赖垃圾回收

四种模式及其适用场景。来源：基于 Python contextlib、tempfile、time 和 unittest.mock 文档的原始表格。

隐藏的收益是初始脚本缺失的部分。性能分析作用域、临时文件、设备句柄、会话和环境变量覆盖等对象都有生命周期。with 代码块使这些生命周期保持局部可见，而不是分散在第 12 行和第 140 行之间。Machine Learning Mastery 对每位 AI 工程师应掌握的 Python 概念的调查中，上下文管理器被列为其中一项，而这项技术在实际应用中的表现正是如此。

当读者打开模块时，应能从一个代码块中清晰看到：模块内获取了什么资源、哪些操作可能失败、以及退出时会恢复哪些状态。位于文件底部的清理逻辑是主流程会执行而错误流程会中断的承诺。错误流程往往在凌晨两点悄然运行。

关于此主题的更多内容

• 《数据分析师手册：CEO与管理者必读》
• 机器学习处理大型数据文件的7种方法
• 使用LLM在Python中构建上下文感知搜索...
• 超越向量搜索：构建确定性系统...
• 5种实用技术检测并缓解LLM问题...
• 超越向量存储：构建完整数据系统...