Python 基础体系 · 第 19/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。

Python 上下文管理器：with、协议、ExitStack 与资源安全

资源管理的核心问题不是“最后调用一个 close()”，而是：

资源一旦成功获得，程序是否能在所有退出路径上，以正确顺序、正确次数完成清理？

这里的“所有退出路径”包括正常返回、异常、循环中的 break、函数中的 return，以及多资源初始化过程中后续资源获取失败。上下文管理器正是 Python 用来表达这类生命周期约束的协议。

本文基于 Python 3.14 的语言参考和标准库文档，重点解释：

• with 语句实际执行了什么；
• 上下文管理器协议如何处理进入、退出和异常；
• __exit__() 返回值为什么会决定异常是否传播；
• @contextmanager 如何把生成器转换为上下文管理器；
• 多个资源为什么必须按后进先出顺序释放；
• ExitStack 如何解决动态资源组合和部分初始化失败；
• AsyncExitStack 与异步资源清理有什么不同；
• 可复用、单次使用、可重入和不可重入之间的边界；
• 如何诊断资源泄漏、错误覆盖和错误抑制。

一、上下文管理器解决的是什么问题

设资源获取操作为 $A$，资源释放操作为 $R$，业务代码为 $B$。最基本的资源安全要求是：

$$A \text{ 成功} \Rightarrow R \text{ 最终执行}$$

更完整地说，如果资源获取成功，程序随后无论经历哪种控制流离开业务区域，都必须执行释放：

$$\forall e \in \{\text{正常结束}, \text{return}, \text{break}, \text{异常}\},\quad A \text{ 成功} \Rightarrow R(e)$$

手写代码通常是：

resource = acquire()
try:
    use(resource)
finally:
    release(resource)

这段代码的关键不在 try，而在于 finally 与资源获取之间形成了明确的生命周期边界：

1. acquire() 成功；
2. 从成功点开始，资源进入“必须释放”状态；
3. use(resource) 可以正常返回或抛出异常；
4. finally 无论如何都会执行 release(resource)。

上下文管理器把这段结构封装为一个对象协议，使调用方可以写成：

with managed_resource() as resource:
    use(resource)

因此，with 不是“自动关闭对象”的语法糖。它是一种围绕代码块建立进入和退出协议的控制流结构。文件、锁、临时状态、数据库事务、网络连接、异步生成器等，都可以使用这一协议。

Python 语言参考将上下文管理器定义为：能够建立并处理代码块运行时上下文的对象；典型用途包括保存和恢复全局状态、加锁与解锁、关闭打开的文件等。(docs.python.org)

二、with 语句的协议

2.1 最小协议

同步上下文管理器至少需要两个方法：

class ContextManager:
    def __enter__(self):
        ...

    def __exit__(self, exc_type, exc_value, traceback):
        ...

它们的职责不同：

• __enter__()：建立上下文，通常获取资源，并返回供 as target 接收的对象；
• __exit__()：离开上下文，通常执行清理，并决定异常是否被抑制。

一个完整的上下文管理器示例：

class ManagedFile:
    def __init__(self, path, mode="r", encoding="utf-8"):
        self.path = path
        self.mode = mode
        self.encoding = encoding
        self.file = None

    def __enter__(self):
        self.file = open(
            self.path,
            self.mode,
            encoding=self.encoding,
        )
        return self.file

    def __exit__(self, exc_type, exc_value, traceback):
        self.file.close()
        return False

使用：

with ManagedFile("data.txt") as file:
    content = file.read()

这里 as file 接收的是 __enter__() 的返回值，而不是上下文管理器本身。若 __enter__() 返回 self，才会得到管理器对象：

class LockContext:
    def __enter__(self):
        self.lock.acquire()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.lock.release()
        return False

2.2 with 的执行步骤

对于：

with expression as target:
    suite

可以将其理解为以下步骤：

1. 求值 expression，得到上下文管理器对象；
2. 获取它的 __enter__() 和 __exit__() 方法；
3. 调用 __enter__()；
4. 将 __enter__() 的返回值绑定到 target；
5. 执行 suite；
6. 离开 suite 时调用 __exit__()；
7. 如果 suite 抛出异常，将异常类型、异常对象和 traceback 传给 __exit__()；
8. 如果 __exit__() 返回真值，异常被抑制；否则异常继续传播。

语言参考明确规定：只要 __enter__() 正常返回，__exit__() 就一定会被调用；即使随后给 as 目标赋值失败，也属于需要调用 __exit__() 的退出路径。(docs.python.org)

可以用近似伪代码表示：

manager = expression
enter = type(manager).__enter__
exit = type(manager).__exit__

value = enter(manager)

try:
    target = value
    suite
except BaseException as exc:
    suppress = exit(
        manager,
        type(exc),
        exc,
        exc.__traceback__,
    )
    if not suppress:
        raise
else:
    exit(manager, None, None, None)

这段伪代码用于理解控制流，不应视为 Python 的逐字实现。重要的是三个事实：

• __enter__() 成功后，退出阶段具有确定性；
• __exit__() 会收到异常信息；
• __exit__() 的真值返回结果影响异常传播。

2.3 __enter__() 失败时不会调用 __exit__()

考虑：

class Broken:
    def __enter__(self):
        print("enter")
        raise RuntimeError("cannot enter")

    def __exit__(self, exc_type, exc_value, traceback):
        print("exit")
        return False

with Broken():
    print("body")

输出：

enter
Traceback (most recent call last):
    ...
RuntimeError: cannot enter

不会输出 exit。

原因是：上下文尚未成功建立，__exit__() 没有被注册为对应的清理阶段。上下文管理器如果在 __enter__() 中先获取资源，再执行可能失败的验证或配置，就必须自行保证“已获取资源、但进入失败”时的回滚。

错误设计：

class BadResource:
    def __enter__(self):
        self.resource = acquire()
        validate(self.resource)  # 这里失败时没有释放
        return self.resource

    def __exit__(self, exc_type, exc_value, traceback):
        release(self.resource)

更安全的设计：

class SafeResource:
    def __enter__(self):
        resource = acquire()
        try:
            validate(resource)
        except BaseException:
            release(resource)
            raise

        self.resource = resource
        return resource

    def __exit__(self, exc_type, exc_value, traceback):
        release(self.resource)
        return False

这里的边界非常重要：

• __enter__() 成功之前：由 __enter__() 自己负责清理已获取但尚未完成初始化的资源；
• __enter__() 成功之后：由 __exit__() 负责清理。

后文的 ExitStack.push() 专门用于简化这种“进入过程中逐步建立回滚保护”的问题。

三、__exit__() 的异常语义

3.1 三个异常参数

当代码块正常结束时：

__exit__(None, None, None)

当代码块抛出异常时：

__exit__(exception_type, exception_value, traceback)

例如：

class Reporter:
    def __enter__(self):
        print("enter")
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        print("exc_type:", exc_type)
        print("exc_value:", exc_value)
        print("traceback is None:", traceback is None)
        return False

with Reporter():
    raise ValueError("bad input")

输出类似：

enter
exc_type: <class 'ValueError'>
exc_value: bad input
traceback is None: False

如果业务代码没有异常，三个参数都是 None：

with Reporter():
    print("ok")

输出类似：

enter
ok
exc_type: None
exc_value: None
traceback is None: True

__exit__() 不需要重新抛出传入的异常。按照协议，如果它返回假值，调用方会继续处理原异常；如果它返回真值，异常会被抑制。Python 数据模型文档特别指出，__exit__() 通常不应重新抛出收到的异常，因为异常是否重新传播由调用方根据返回值处理。(docs.python.org)

3.2 返回值决定是否抑制异常

class SuppressValueError:
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        return exc_type is ValueError

使用：

with SuppressValueError():
    raise ValueError("ignored")

print("继续执行")

输出：

继续执行

但以下异常不会被抑制：

with SuppressValueError():
    raise TypeError("not ignored")

此时 __exit__() 返回 False，TypeError 继续传播。

抑制异常本质上是一个控制流决定：

$$\text{传播结果} = \begin{cases} \text{继续抛出异常}, & \text{若 } \operatorname{bool}(\_\_exit\_\_) = \text{False}\\ \text{正常继续}, & \text{若 } \operatorname{bool}(\_\_exit\_\_) = \text{True} \end{cases}$$

因此，以下写法风险很高：

def __exit__(self, exc_type, exc_value, traceback):
    cleanup()
    return True

它会把代码块中的所有异常都吞掉，包括程序错误、数据损坏、协议错误和取消信号。除非上下文管理器明确承担“某类异常可以安全忽略”的契约，否则 __exit__() 应返回 False 或 None。

3.3 清理异常会覆盖原异常

class BadCleanup:
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        raise OSError("cleanup failed")

with BadCleanup():
    raise ValueError("body failed")

最终看到的异常通常是清理阶段抛出的 OSError，而原来的 ValueError 会作为上下文保留在异常链中。

这反映了一个不可回避的事实：如果“业务失败”和“清理失败”同时发生，单个异常位置无法同时直接呈现两个异常。此时应该根据资源类型设计错误契约：

• 清理失败是否比原始业务失败更重要；
• 是否记录原始异常；
• 是否将清理异常作为 __context__ 或显式异常链的一部分；
• 是否使用 Python 的异常分组机制表达多个独立故障。

清理函数不应随意吞掉异常，也不应盲目覆盖更有诊断价值的原始异常。上下文管理器只是提供了异常传递入口，并不能自动替你决定故障优先级。

四、多个 with 项的真实结构

代码：

with open("a.txt") as a, open("b.txt") as b:
    process(a, b)

语义上等价于嵌套：

with open("a.txt") as a:
    with open("b.txt") as b:
        process(a, b)

因此退出顺序是：

1. 退出 b.txt；
2. 退出 a.txt。

也就是后进入的上下文先退出：

$$A_1 \rightarrow A_2 \rightarrow B_2 \rightarrow B_1$$

其中 $A_i$ 表示获取资源，$B_i$ 表示释放资源。

这种顺序不是形式问题，而是依赖关系的要求。例如：

with open("output.txt", "w") as file:
    with lock:
        file.write("data")

锁先获得、文件后获得时，退出顺序是先释放文件，再释放锁。若资源之间有依赖，必须让被依赖资源更早退出。

Python 3.14 还支持使用括号将多个上下文管理器拆成多行：

with (
    open("input.txt", encoding="utf-8") as source,
    open("output.txt", "w", encoding="utf-8") as target,
):
    target.write(source.read())

多项 with 的括号换行语法与多个上下文表达式本身都是语言规范支持的行为。(docs.python.org)

五、类形式上下文管理器：如何正确设计生命周期

一个上下文管理器通常有以下状态：

未进入
  │
  ├── __enter__ 失败 ──> 失败，__exit__ 不调用
  │
  └── __enter__ 成功 ──> 已进入
                              │
                              ├── body 正常结束
                              ├── body return
                              ├── body break
                              └── body 抛出异常
                                      │
                                  __exit__
                                      │
                              释放并决定异常传播

一个锁上下文管理器可以这样写：

import threading

class Locked:
    def __init__(self, lock):
        self.lock = lock

    def __enter__(self):
        self.lock.acquire()
        return self.lock

    def __exit__(self, exc_type, exc_value, traceback):
        self.lock.release()
        return False

使用：

lock = threading.Lock()

with Locked(lock):
    update_shared_state()

这里有三个隐含契约：

1. acquire() 成功后必须执行 release()；
2. release() 应该执行一次；
3. 业务代码的异常不能因为释放锁而被无意抑制。

5.1 __exit__() 中不要依赖业务变量存在

错误示例：

class BadConnection:
    def __enter__(self):
        self.connection = connect()
        self.session_id = authenticate(self.connection)
        return self.connection

    def __exit__(self, exc_type, exc_value, traceback):
        logout(self.connection, self.session_id)
        self.connection.close()

如果 authenticate() 失败，__exit__() 不会调用，所以这个例子不会因为 session_id 不存在而触发退出清理；但如果设计改成了在进入后又引入新的中间状态，就很容易出现“部分初始化对象被退出逻辑错误处理”的问题。

更稳妥的方式是让 __enter__() 内部使用局部回滚：

class Connection:
    def __init__(self):
        self.connection = None
        self.session_id = None

    def __enter__(self):
        connection = connect()
        try:
            session_id = authenticate(connection)
        except BaseException:
            connection.close()
            raise

        self.connection = connection
        self.session_id = session_id
        return connection

    def __exit__(self, exc_type, exc_value, traceback):
        try:
            if self.session_id is not None:
                logout(self.connection, self.session_id)
        finally:
            if self.connection is not None:
                self.connection.close()

        return False

清理操作也可能失败，因此这里使用嵌套的 try/finally，确保即使 logout() 失败，底层连接仍有机会关闭。

5.2 资源清理的幂等性

幂等清理是指重复执行清理不会产生新的错误或破坏额外状态。例如：

class FileLike:
    def __init__(self, file):
        self.file = file
        self.closed = False

    def close(self):
        if not self.closed:
            self.file.close()
            self.closed = True

但不能假设所有外部资源的关闭操作都天然幂等。某些连接、事务或锁在错误状态下重复释放可能抛出异常。因此，上下文管理器应明确：

• 是否允许手动关闭后再离开 with；
• 是否允许重复进入；
• 是否允许嵌套使用同一个实例；
• 关闭失败时如何处理。

六、@contextmanager：用生成器表达进入与退出

对于一次性、线性的资源管理逻辑，contextlib.contextmanager 可以把生成器函数转换为上下文管理器：

from contextlib import contextmanager

@contextmanager
def managed_file(path, mode="r", encoding="utf-8"):
    file = open(path, mode, encoding=encoding)
    try:
        yield file
    finally:
        file.close()

使用：

with managed_file("data.txt") as file:
    print(file.read())

其结构是：

yield 之前：进入阶段
yield 的值：as 目标接收的对象
yield 之后：退出阶段
finally：确保异常和正常退出都执行清理

可以把它理解为：

generator = managed_file("data.txt")
value = next(generator)       # 执行到 yield
try:
    body(value)
except BaseException as exc:
    generator.throw(...)
else:
    next(generator)            # 继续执行 yield 后的清理

这仍然是语义模型，而不是实现细节。@contextmanager 的标准模式是：获取资源后进入 try，通过 yield 暴露资源，再在 finally 中释放资源。(docs.python.org)

6.1 yield 必须恰好形成一个上下文边界

错误示例：

from contextlib import contextmanager

@contextmanager
def invalid():
    yield "first"
    yield "second"

一个 with 只能对应一个进入值和一个退出阶段。第二次 yield 会导致生成器协议错误。

同样，完全不执行 yield 也不成立：

@contextmanager
def invalid():
    raise RuntimeError("never yields")

@contextmanager 适合表达单一的“获取—使用—释放”流程，不适合把多个互相独立、动态数量的资源全部硬编码到一个生成器中。后者通常更适合 ExitStack。

6.2 不要把资源获取写在 try 之外再假设一定清理

下面这种结构是正确的：

@contextmanager
def resource():
    value = acquire()
    try:
        yield value
    finally:
        release(value)

如果 acquire() 本身失败，没有资源需要释放。

下面这种结构也常见，但要注意获取后的验证：

@contextmanager
def validated_resource():
    value = acquire()
    try:
        validate(value)
        yield value
    except BaseException:
        release(value)
        raise
    else:
        release(value)

不过实际代码中，try/finally 通常更清晰：

@contextmanager
def validated_resource():
    value = acquire()
    try:
        validate(value)
        yield value
    finally:
        release(value)

因为 validate() 和 with 主体中的异常都应触发相同清理。

6.3 上下文管理器的单次使用问题

@contextmanager 生成的具体上下文管理器实例通常是单次使用的：

cm = managed_file("data.txt")

with cm:
    pass

with cm:  # 不应依赖这种用法
    pass

正确方式是每次创建新的实例：

with managed_file("data.txt"):
    pass

with managed_file("data.txt"):
    pass

标准库文档也特别区分了单次使用、可复用和可重入：

• 文件对象通常是单次使用上下文管理器，因为第一次离开后文件已关闭；
• @contextmanager 创建的实例通常是单次使用；
• threading.Lock 可以多次使用，但不是可重入的；
• threading.RLock 用于需要同一线程重复获取的场景；
• 可复用不等于可重入。(docs.python.org)

七、contextlib 中的几个基础工具

7.1 closing()：只有 close()，没有上下文协议

某些第三方对象提供 close()，但没有实现 __enter__() 和 __exit__()。可以使用：

from contextlib import closing
from urllib.request import urlopen

with closing(urlopen("https://example.com")) as response:
    data = response.read()

closing(obj) 的逻辑近似为：

@contextmanager
def closing(obj):
    try:
        yield obj
    finally:
        obj.close()

它只适合清理方法明确是 close() 的对象。若对象需要 shutdown()、rollback() 或 aclose()，应该使用对应协议或自定义上下文管理器。

7.2 nullcontext()：统一“调用方提供”和“函数创建”的资源

一个函数可能接受文件路径，也可能接受已打开的文件：

from contextlib import nullcontext

def read_text(file_or_path):
    if isinstance(file_or_path, (str, bytes)):
        cm = open(file_or_path, encoding="utf-8")
    else:
        cm = nullcontext(file_or_path)

    with cm as file:
        return file.read()

这里的资源所有权不同：

• 函数自己 open() 的文件：函数负责关闭；
• 调用方传入的文件：函数只使用，不负责关闭。

nullcontext(value) 会在进入时返回 value，退出时不做任何事情，也可用于异步上下文管理器的可选分支。(docs.python.org)

7.3 suppress()：只抑制明确允许忽略的异常

from contextlib import suppress
import os

with suppress(FileNotFoundError):
    os.remove("temporary.txt")

它等价于：

try:
    os.remove("temporary.txt")
except FileNotFoundError:
    pass

关键不是“少写几行”，而是把抑制范围限制在一个非常小的代码块中：

with suppress(FileNotFoundError):
    os.remove("temporary.txt")

不要写成：

with suppress(Exception):
    whole_business_operation()

这会让真正的程序错误看起来像成功。Python 3.12 起，suppress() 还支持从 BaseExceptionGroup 中移除匹配的异常分支；未被抑制的分支会以新的异常组重新抛出。(docs.python.org)

八、ExitStack：动态构建退出栈

普通 with 适合资源数量和结构在代码中静态可见的情况：

with open("a.txt") as a:
    with open("b.txt") as b:
        process(a, b)

但如果资源数量由输入决定，就不能简单地写出固定嵌套：

paths = ["a.txt", "b.txt", "c.txt"]

此时使用：

from contextlib import ExitStack

def read_all(paths):
    with ExitStack() as stack:
        files = [
            stack.enter_context(open(path, encoding="utf-8"))
            for path in paths
        ]
        return [file.read() for file in files]

ExitStack 内部维护一个退出回调栈：

注册 open(a) 的退出动作
注册 open(b) 的退出动作
注册 open(c) 的退出动作

离开 with：
执行 close(c)
执行 close(b)
执行 close(a)

它的退出顺序是后进先出，与嵌套 with 相同。标准库文档规定，注册的回调会在显式调用 close() 或离开 with 时逆序执行；垃圾回收并不会隐式关闭 ExitStack。(docs.python.org)

8.1 部分初始化失败

考虑以下代码：

def open_many(paths):
    files = []
    try:
        for path in paths:
            files.append(open(path, encoding="utf-8"))
    except BaseException:
        for file in reversed(files):
            file.close()
        raise

    return files

它需要手动维护：

• 已成功打开哪些文件；
• 打开失败后按什么顺序关闭；
• 清理过程中出现异常如何处理。

使用 ExitStack 后：

def open_many(paths):
    with ExitStack() as stack:
        files = [
            stack.enter_context(open(path, encoding="utf-8"))
            for path in paths
        ]

        # 只有这里显式转移所有权后，文件才会在 with 外继续保持打开。
        close_all = stack.pop_all().close

    return files, close_all

执行过程如下：

1. 打开第一个文件成功，注册它的 __exit__()；
2. 打开第二个文件成功，再注册第二个退出动作；
3. 如果第三个文件打开失败，ExitStack.__exit__() 逆序关闭前两个文件；
4. 如果全部成功，调用 pop_all()；
5. pop_all() 将回调转移到新的栈，当前 with 不再关闭文件；
6. 调用方之后执行 close_all()，完成最终释放。

pop_all() 是资源所有权转移，而不是“取消清理”。它不会执行回调，只是把回调移动到新栈中。(docs.python.org)

8.2 enter_context()、push() 和 callback() 的区别

enter_context(cm)

with ExitStack() as stack:
    file = stack.enter_context(open("data.txt"))

它会：

1. 调用上下文管理器的 __enter__()；
2. 返回 __enter__() 的结果；
3. 注册对应的 __exit__()。

因此，它等价于把一个动态上下文管理器加入当前 with 生命周期。

push(exit)

with ExitStack() as stack:
    resource = acquire()
    stack.push(resource_manager)

push() 注册退出动作，但不会调用 resource_manager.__enter__()。这使它特别适合保护 __enter__() 中已经完成的部分初始化：

from contextlib import AbstractContextManager, ExitStack

class ValidatedResource(AbstractContextManager):
    def __init__(self, acquire, release, validate):
        self.acquire = acquire
        self.release = release
        self.validate = validate
        self.resource = None

    def __enter__(self):
        self.resource = self.acquire()

        with ExitStack() as stack:
            stack.callback(self.release, self.resource)

            self.validate(self.resource)

            # 验证成功，转移清理责任
            stack.pop_all()

        return self.resource

    def __exit__(self, exc_type, exc_value, traceback):
        self.release(self.resource)
        return False

这里的含义是：

• 资源刚获取后，立即注册一个“失败时释放”的回调；
• 验证成功后，通过 pop_all() 移除临时回滚责任；
• 后续责任转移给外层上下文管理器的 __exit__()。

callback(function, *args, **kwargs)

with ExitStack() as stack:
    stack.callback(remove_temp_file, "tmp.dat")

它注册普通函数调用。与 push() 不同，普通 callback 不接收异常类型、异常对象和 traceback，因此不能通过返回值抑制异常。(docs.python.org)

8.3 ExitStack 中的异常状态会沿栈更新

多个退出回调并不是各自独立地面对原异常。若内层退出回调抑制或替换异常，外层退出回调收到的是更新后的异常状态。

这与嵌套结构一致：

with outer:
    with inner:
        raise ValueError("body")

退出顺序：

inner.__exit__(ValueError, ...)
outer.__exit__(更新后的异常状态)

因此，退出回调的顺序不仅影响资源释放，也影响异常传播和错误链。若清理动作之间存在依赖，应按照依赖关系注册。

九、ExitStack 的事务式资源获取模式

有时我们希望实现“全有或全无”：

• 所有文件都成功打开，调用方获得它们；
• 任意一个文件打开失败，已经打开的全部关闭；
• 中间不能出现部分成功但资源泄漏。

完整示例：

from contextlib import ExitStack
from pathlib import Path

def open_inputs(paths):
    with ExitStack() as stack:
        files = []

        for path in paths:
            file = stack.enter_context(
                Path(path).open("r", encoding="utf-8")
            )
            files.append(file)

        # 成功后把清理责任整体移交给调用方
        ownership = stack.pop_all()

    return files, ownership

调用：

files, ownership = open_inputs(["a.txt", "b.txt"])

try:
    for file in files:
        print(file.read())
finally:
    ownership.close()

状态变化：

开始：
stack = []

打开 a：
stack = [close(a)]

打开 b：
stack = [close(a), close(b)]

全部成功：
ownership = 新 ExitStack([close(a), close(b)])
stack = []

调用方使用 files

ownership.close():
close(b)
close(a)

这里需要注意一个生产边界：ExitStack 不会替你管理返回对象的生命周期。若函数返回了资源，却没有返回关闭句柄、没有把资源转移给另一个管理器，仍然可能泄漏。

十、异步上下文管理器与 async with

异步上下文管理器使用：

async def __aenter__(self):
    ...

async def __aexit__(self, exc_type, exc_value, traceback):
    ...

对应语句：

async with resource() as value:
    await use(value)

与同步协议相比，主要差异是：

• __aenter__() 必须返回可等待对象；
• __aexit__() 也必须返回可等待对象；
• 退出时需要 await 清理动作。

Python 数据模型将异步上下文管理器定义为能够在 __aenter__() 和 __aexit__() 中挂起执行的上下文管理器。(docs.python.org)

示例：

class AsyncConnection:
    async def __aenter__(self):
        self.connection = await connect_async()
        return self.connection

    async def __aexit__(self, exc_type, exc_value, traceback):
        await self.connection.close()
        return False

使用：

async with AsyncConnection() as connection:
    await connection.execute("SELECT 1")

10.1 @asynccontextmanager

from contextlib import asynccontextmanager

@asynccontextmanager
async def connection():
    conn = await connect_async()
    try:
        yield conn
    finally:
        await conn.close()

使用：

async def query():
    async with connection() as conn:
        return await conn.fetch("SELECT 1")

它与同步的 @contextmanager 结构相同，只是获取和清理过程可以 await。标准库文档将其定义为应用于异步生成器函数的装饰器，用于创建 async with 上下文管理器。(docs.python.org)

10.2 AsyncExitStack

当异步资源数量动态变化，使用 AsyncExitStack：

from contextlib import AsyncExitStack

async def open_connections(count):
    async with AsyncExitStack() as stack:
        connections = []

        for _ in range(count):
            conn = await stack.enter_async_context(connection())
            connections.append(conn)

        return connections

如果第 $k$ 个连接获取失败，前 $k-1$ 个连接会按逆序异步关闭。

AsyncExitStack 还可以组合：

• 同步上下文管理器；
• 异步上下文管理器；
• 异步清理协程。

异步栈没有 close()，必须使用 aclose()；push_async_exit() 注册异步退出逻辑，push_async_callback() 注册异步回调。(docs.python.org)

例如：

from contextlib import AsyncExitStack

async def remove_temp(path):
    await async_remove(path)

async def workflow():
    async with AsyncExitStack() as stack:
        session = await stack.enter_async_context(connection())

        temp_path = await create_temp_file()
        stack.push_async_callback(remove_temp, temp_path)

        await session.execute("PROCESS")

10.3 异步生成器提前结束

异步生成器可能因为 break 或异常提前结束：

async def stream():
    try:
        yield 1
        yield 2
    finally:
        await release_async_state()

为了保证异步生成器的退出代码在正确的异步上下文中执行，可以使用 aclosing()：

from contextlib import aclosing

async with aclosing(stream()) as values:
    async for value in values:
        if value == 1:
            break

这比依赖垃圾回收触发异步生成器清理更确定，也能避免退出代码在依赖的任务或上下文变量已经失效后才运行。(docs.python.org)

十一、同步与异步资源不能混用

以下代码不成立：

async with open("data.txt") as file:
    ...

普通文件对象是同步上下文管理器，不能直接放入 async with。

反过来：

with async_connection() as connection:
    ...

也不成立，因为异步上下文管理器的进入和退出都需要等待。

如果异步函数中使用同步文件，可以直接使用同步 with，但要认识到文件操作本身可能阻塞事件循环：

async def read_file():
    with open("data.txt", encoding="utf-8") as file:
        return file.read()

这段代码在协议上是正确的，但在文件很大或存储很慢时可能阻塞事件循环。是否需要线程池、异步文件库或其他 I/O 方案，取决于应用的并发模型，而不是由 with 自动决定。

十二、上下文管理器的复用、重入与线程安全

这几个概念容易混淆。

12.1 单次使用

单次使用表示一个实例只能安全地进入一次：

cm = open("data.txt")

with cm:
    pass

# cm 已关闭

文件对象就是典型例子。

12.2 可复用

可复用表示退出后可以再次进入：

lock = threading.Lock()

with lock:
    pass

with lock:
    pass

同一个锁实例可以被多次使用。

12.3 可重入

可重入表示同一个实例可以嵌套进入：

with cm:
    with cm:
        ...

一个可复用对象不一定可重入。普通 threading.Lock 可以重复使用，但同一线程持有它时再次获取会阻塞；threading.RLock 才支持同一线程递归获取。

ExitStack 也不能随意嵌套复用：

from contextlib import ExitStack

stack = ExitStack()

with stack:
    stack.callback(print, "outer")

    with stack:
        stack.callback(print, "inner")

内层退出时可能清空整个栈，使外层不再拥有预期的回调。应使用不同实例：

with ExitStack() as outer:
    outer.callback(print, "outer")

    with ExitStack() as inner:
        inner.callback(print, "inner")

标准库文档明确指出，ExitStack 可复用但不可按这种方式重入；嵌套使用同一个实例会导致回调在内层退出时被清空。(docs.python.org)

十三、常见错误与诊断方法

13.1 把 with 当作“对象有 close()”

以下代码只有在对象实现上下文管理器协议时才成立：

with resource:
    ...

仅仅存在 close() 方法并不够。否则会得到类似：

TypeError: 'SomeObject' object does not support the context manager protocol

解决方式：

• 使用对象原生提供的上下文管理器；
• 使用 closing(resource)；
• 编写包装器；
• 在库设计中直接实现 __enter__() 和 __exit__()。

13.2 在 __exit__() 中无条件返回真值

错误：

def __exit__(self, exc_type, exc_value, traceback):
    release()
    return True

这会抑制所有异常。通常应该：

def __exit__(self, exc_type, exc_value, traceback):
    release()
    return False

如果确实需要抑制异常，应精确匹配：

def __exit__(self, exc_type, exc_value, traceback):
    release()

    if exc_type is FileNotFoundError:
        return True

    return False

13.3 清理回调注册得太晚

错误模式：

resource = acquire()
do_more_initialization(resource)
register_cleanup(resource)

如果 do_more_initialization() 抛出异常，资源已经获取，但清理动作还没有注册。

正确原则是：

资源成功获取后，尽快注册对应的清理动作。

使用 ExitStack：

with ExitStack() as stack:
    resource = acquire()
    stack.callback(release, resource)

    do_more_initialization(resource)

13.4 误以为垃圾回收等于资源清理

对象没有引用后，垃圾回收何时发生取决于实现、引用关系和运行环境。文件描述符、锁、连接和临时文件等外部资源不应依赖对象析构来释放。

特别是 ExitStack 的回调不会因为栈对象被垃圾回收而自动执行；必须显式调用 close()，或使用 with ExitStack()。(docs.python.org)

13.5 用过大的 with 范围包住无关代码

with database_transaction() as transaction:
    read_configuration()
    render_template()
    call_remote_service()
    write_database(transaction)

上下文范围越大，资源持有时间越长，锁和事务的竞争窗口也越大。with 的边界应该覆盖资源真正需要存在的代码，而不是简单地包住整个函数。

例如：

configuration = read_configuration()
rendered = render_template(configuration)
remote_result = call_remote_service(rendered)

with database_transaction() as transaction:
    write_database(transaction, remote_result)

这不是形式上的优化，而是缩短事务和锁的生命周期，降低并发冲突与故障影响范围。

十四、一个完整的资源组合示例

下面的例子模拟一个需要同时管理以下资源的任务：

• 输入文件；
• 输出文件；
• 可选的临时目录；
• 任务结束时的日志回调；
• 任意一步初始化失败时的自动回滚。

from contextlib import ExitStack
from pathlib import Path
import tempfile


def process(input_paths, output_path, keep_temp=False):
    with ExitStack() as stack:
        input_files = [
            stack.enter_context(
                Path(path).open("r", encoding="utf-8")
            )
            for path in input_paths
        ]

        output_file = stack.enter_context(
            Path(output_path).open("w", encoding="utf-8")
        )

        temp_dir = stack.enter_context(
            tempfile.TemporaryDirectory()
        )

        if not keep_temp:
            # 这里仅作示例：TemporaryDirectory 已经自带清理逻辑。
            # 真实代码不应重复注册同一资源的清理动作。
            pass

        stack.callback(
            print,
            f"cleanup completed for {output_path}",
        )

        for input_file in input_files:
            for line in input_file:
                output_file.write(line.upper())

        print(f"temporary directory: {temp_dir}")

其生命周期是：

进入 ExitStack
  ↓
打开第一个输入文件，注册 close
  ↓
打开后续输入文件，继续注册 close
  ↓
打开输出文件，注册 close
  ↓
创建临时目录，注册 cleanup
  ↓
注册日志 callback
  ↓
执行处理
  ↓
退出时：
  日志 callback
  临时目录 cleanup
  输出文件 close
  输入文件逆序 close

注意，普通 callback() 的执行顺序取决于注册顺序。因此如果日志需要在所有资源都清理完后输出，就不能在最开始注册，而应在资源清理动作之后安排，或者使用单独的外层上下文：

def process(input_paths, output_path):
    with ExitStack() as resources:
        input_files = [
            resources.enter_context(
                Path(path).open("r", encoding="utf-8")
            )
            for path in input_paths
        ]
        output_file = resources.enter_context(
            Path(output_path).open("w", encoding="utf-8")
        )

        for input_file in input_files:
            for line in input_file:
                output_file.write(line.upper())

    print(f"cleanup completed for {output_path}")

此时 print() 位于 ExitStack 外部，因此它一定发生在资源清理之后。

十五、如何选择实现方式

可以按资源结构选择：

直接使用 with

资源数量固定，结构清晰：

with open("input.txt") as source:
    data = source.read()

自定义类

需要：

• 保存状态；
• 支持复杂的进入和退出逻辑；
• 明确控制复用、重入或错误契约；
• 将上下文管理器作为公共类型暴露。

class Transaction:
    ...

@contextmanager

逻辑是单一线性的“获取—使用—释放”：

@contextmanager
def transaction():
    tx = begin()
    try:
        yield tx
    finally:
        rollback_or_commit(tx)

ExitStack

需要：

• 动态数量的上下文管理器；
• 可选资源；
• 普通清理回调；
• 部分初始化失败时的回滚；
• 将一组资源的所有权整体转移。

AsyncExitStack

需要同时处理异步上下文管理器、异步清理协程和动态资源集合。

这些工具不是层层替代关系。@contextmanager 适合封装单个协议；ExitStack 适合在运行时组织多个协议。

十六、资源安全的最终判断标准

判断一个上下文管理器是否正确，不能只看它能否运行一次，而应逐条验证：

1. __enter__() 获取资源后，如果后续初始化失败，是否回滚；
2. __enter__() 成功后，__exit__() 是否必然释放资源；
3. 正常结束、异常、return 和 break 是否都能触发清理；
4. 多资源是否按逆序释放；
5. __exit__() 是否错误地抑制了异常；
6. 清理异常是否会覆盖原始异常；
7. 清理动作是否可能重复执行；
8. 对象是否允许复用或重入；
9. 异步资源是否使用 async with 和 await；
10. ExitStack 是否显式关闭，或始终位于 with 中；
11. pop_all() 后的资源所有权是否清楚；
12. 返回给调用方的资源是否同时返回了明确的关闭责任。

上下文管理器的本质可以概括为一个状态转换：

$$\text{未持有资源} \xrightarrow{\_\_enter\_\_} \text{持有资源} \xrightarrow{\_\_exit\_\_} \text{资源已释放}$$

资源安全要求的是：从“持有资源”状态出发，不存在绕过“资源已释放”的退出路径。with 提供静态结构，协议提供异常语义，ExitStack 提供动态组合；三者共同构成 Python 中可验证、可组合的资源生命周期模型。

---

系列导航与关联阅读

• 系列入口：Python 完整学习路线：从语言模型、并发到 Web、数据、AI 与生产交付
• 上一篇：Python 异常体系：传播、链、分组、清理和错误契约
• 下一篇：Python 模块与包：命名空间、init、入口和项目边界
• 延伸：Python I/O 模型：文本、二进制、缓冲、编码和流式处理

官方资料

• Python 3.14 Standard Library
• Python HOWTOs

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。