上下文管理器

上下文管理器把资源获取与必需的清理绑定到with代码块,明确处理异常传播、生成器封装、动态组合与异步释放。

难度 进阶 时长 标准深度约11分钟
版本 Python 3.14
what

上下文管理器(context manager) 为一段代码建立并撤销运行时上下文;with保证已经成功进入的管理器会收到退出通知。

trap

as绑定的是__enter__()的返回值,不一定是管理器本身;__exit__()返回真值还会抑制代码块抛出的异常。

fix

把释放放在无条件执行的退出路径中,默认让异常继续传播,并测试进入失败、代码块失败和退出失败三条路径。

是什么,为什么存在

上下文管理器是实现上下文管理协议的对象。同步协议由__enter__()和__exit__()组成,with语句负责在代码块前后调用它们。文件、锁、事务、临时配置和输出重定向都需要这种成对操作。

它解决的核心问题不是少写一次close(),而是让清理与控制流绑定。代码块可能正常结束,也可能通过异常、return、break或continue离开;只要进入成功,退出方法就会运行。这个保证把资源生命周期放在创建位置附近,审查者不必追踪每条离开路径。

上下文不一定拥有外部资源。它也可以暂时修改进程或对象状态,退出时恢复旧值;还可以在事务成功时提交、失败时回滚。共同点是进入和退出构成一个明确边界,而不是具体调用了哪个API。

这种边界提供一种 异常安全保证(exception-safety guarantee) :发生异常时,管理器仍有机会恢复不变量并释放已取得的资源。它不自动保证操作原子性,也不会自动决定异常是否应被忽略;这些仍是管理器契约的一部分。

当清理只取决于一个词法代码块时,优先使用with。若资源需要跨函数、任务或请求长期存活,就要把所有权放在更外层,并让那个所有者最终进入和退出上下文;不要保存as目标后便丢失管理器生命周期。

工作原理

执行with manager_expression as target:时,Python先求值管理器表达式,再调用进入方法。进入方法的返回值赋给target,随后才执行代码块。代码块离开后,Python调用同一个管理器的退出方法。

正常离开时,__exit__()收到三个None。代码块或as目标绑定抛出异常时,它收到异常类型、异常实例和traceback。退出方法的返回值只在处理异常时决定传播:真值表示抑制原异常,假值表示继续抛出。

下面的流程图刻意把进入失败放在受保护区域之外。若__enter__()自己抛出异常,对应的__exit__()不会被调用;已经完成部分获取的进入方法必须自行撤销那部分工作。

可以把协议近似理解为下列步骤:

  1. 求值上下文表达式并保存管理器。
  2. 查找管理器类型提供的特殊方法,再调用__enter__()。
  3. 把进入结果绑定给as目标并运行代码块。
  4. 正常离开时,用三个None调用__exit__()。
  5. 异常离开时,把异常信息交给__exit__(),再依据返回值传播或抑制异常。

这里的“近似”很重要。解释器对协议方法使用隐式特殊方法查找,不能靠给单个实例临时赋一个__exit__属性来可靠改变行为。协议应实现在类上,组合行为则应通过另一个管理器对象表达。

__enter__()可以返回管理器自身,也可以返回真正供代码块使用的资源。例如文件对象通常返回自身,锁对象的上下文协议则不要求as目标就是锁。API文档必须说明进入结果的类型,调用方不能从管理器类型猜测。

一条with A() as a, B() as b:与两层嵌套with等价。进入顺序从左到右,退出顺序从右到左;若B.__enter__()失败,已经进入的A仍会退出。这种栈式行为是安全组合多个资源的基础。

示例

下面四个示例依次展示内置管理器、类协议、生成器封装和动态清理栈。输出均由本地python3执行对应文件得到。

文件在异常后仍关闭

文件对象已经实现上下文管理协议。代码块主动抛出异常后,文件先被关闭,异常才到达外层except。

read_orders.py
from pathlib import Path

path = Path("orders.txt")
path.write_text("A-17\nB-04\n", encoding="utf-8")

try:
    with path.open(encoding="utf-8") as handle:
        print(f"first order: {handle.readline().strip()}")
        print(f"open inside: {not handle.closed}")
        raise LookupError("customer record missing")
except LookupError as error:
    print(f"caught: {error}")

print(f"closed outside: {handle.closed}")
path.unlink()
first order: A-17
open inside: True
caught: customer record missing
closed outside: True

handle在代码块外仍然是一个绑定,但它指向已关闭的文件。上下文管理器控制资源状态,不会删除变量。后续读写会失败,因此不要把名称仍存在误当成资源仍可用。

示例最后删除自己创建的文件,使重复运行具有相同初始状态。清理样例数据与关闭文件是两个不同责任:文件管理器负责后者,示例所有者负责前者。

用类实现临时状态

TemporaryValue保存键原来是否存在以及旧值,进入时写入临时值,退出时精确恢复。它返回临时值而不是self,说明as目标由协议设计者决定。

temporary_value.py
class TemporaryValue:
    _missing = object()

    def __init__(self, mapping, key, value):
        self.mapping = mapping
        self.key = key
        self.value = value
        self.previous = self._missing

    def __enter__(self):
        self.previous = self.mapping.get(self.key, self._missing)
        self.mapping[self.key] = self.value
        return self.value

    def __exit__(self, exc_type, exc_value, traceback):
        if self.previous is self._missing:
            del self.mapping[self.key]
        else:
            self.mapping[self.key] = self.previous
        name = exc_type.__name__ if exc_type else "None"
        print(f"exit saw: {name}")
        return False


settings = {}
try:
    with TemporaryValue(settings, "mode", "preview") as mode:
        print(mode, settings)
        raise ValueError("invalid draft")
except ValueError as error:
    print(f"propagated: {error}")

print(settings)
preview {'mode': 'preview'}
exit saw: ValueError
propagated: invalid draft
{}

哨兵对象区分“键不存在”和“键存在但值为None”。若用mapping.get(key)记录旧值,这两种状态会混在一起,退出时可能留下原本不存在的键。

退出方法先恢复状态,再返回False。外层因此看见原来的ValueError,同时观察到空字典已经恢复;这是测试异常安全契约时应同时检查的两个结果。

用生成器表达提交与回滚

contextlib.contextmanager把只产生一次值的 生成器(generator) 函数适配为上下文管理器。yield之前对应进入,产生的值成为as目标,恢复生成器后的路径对应退出。

staged_update.py
from contextlib import contextmanager


@contextmanager
def staged_update(store):
    before = store.copy()
    try:
        yield store
    except BaseException:
        store.clear()
        store.update(before)
        print("rollback")
        raise
    else:
        print("commit")


inventory = {"tea": 2}
with staged_update(inventory) as draft:
    draft["tea"] -= 1
print(inventory)

try:
    with staged_update(inventory) as draft:
        draft["coffee"] = 3
        raise KeyError("missing sku")
except KeyError as error:
    print(f"caught: {error}")
print(inventory)
commit
{'tea': 1}
rollback
caught: 'missing sku'
{'tea': 1}

第二个代码块修改了字典后失败,管理器恢复进入前的浅副本并重新抛出异常。浅副本只适合这里的整数值;若值中还有可变对象,契约需要明确复制深度或采用真正的事务机制。

捕获BaseException是因为这段模拟回滚必须覆盖取消和进程级中断,再立即使用裸raise保留原异常。普通业务错误处理通常捕获更窄的Exception子类;资源恢复与业务恢复不是同一个边界。

用ExitStack动态组合

上下文管理器数量由运行时数据决定时,静态写出多项with不够方便。ExitStack.enter_context()每成功进入一个管理器,就立刻把其退出方法登记到栈中。

dynamic_cleanup.py
from contextlib import ExitStack, contextmanager


@contextmanager
def connected(service):
    print(f"connect {service}")
    try:
        yield service.upper()
    finally:
        print(f"disconnect {service}")


services = ["cache", "search"]
with ExitStack() as stack:
    connections = [
        stack.enter_context(connected(service))
        for service in services
    ]
    stack.callback(print, "clear request cache")
    print(" + ".join(connections))
connect cache
connect search
CACHE + SEARCH
clear request cache
disconnect search
disconnect cache

输出显示严格的后进先出顺序。最后登记的普通回调最先运行,然后释放search,最后释放cache;这与嵌套with从内向外退出一致。

若进入第二个服务时失败,第一个服务已经在栈中,因此仍会释放。先用列表推导一次性建立所有连接,再把完成的列表交给栈,会失去这个部分获取失败时的保证。

陷阱

假设进入失败会自动退出

修复方法: 让__enter__()在失败前自行撤销部分工作,或用内部ExitStack边获取边登记清理。专门注入第二步获取失败,验证第一步已释放。

混淆管理器与进入结果

修复方法: 分别标注管理器表达式类型与进入结果类型。实现协议时明确返回self还是代理资源;使用第三方API时查其契约,不靠命名推断。

意外抑制异常

修复方法: 默认显式返回False或None。确实要抑制时,只匹配文档列出的异常类型,并测试其他异常仍会传播;不要用宽泛抑制代替输入验证。

把清理写在yield后但没有finally

修复方法: 无条件释放放在finally中;提交与回滚使用清楚的else、except和raise分支。测试代码块异常和退出代码自身异常,而不只测试成功路径。

一次性获取后才登记清理

修复方法: 在同一个循环中调用stack.enter_context(),每次成功后立即获得清理保证。动态获取的测试要让第二个或更晚的资源失败。

复用一次性管理器或混用同步协议

修复方法: 每次使用时调用管理器工厂,除非文档明确承诺可复用或可重入。根据清理操作是否需要await选择同步或异步协议,并让类型检查器检查边界。

深入 异常路径就是契约

异常路径就是契约

with保护的不只是缩进代码块。as目标绑定也位于退出方法保护范围内,因此解包目标失败时仍会调用__exit__()。相反,管理器表达式求值、特殊方法查找和__enter__()调用发生在保护范围建立之前。

退出方法收到的三个异常参数与sys.exc_info()对应。没有异常时三个值都是None;有异常时,类型用于分类,实例携带数据,traceback记录传播路径。多数管理器只需要判断exc_type is None,不应解析异常消息文本。

返回真值表示管理器已经把异常处理完,调用方会从with后继续执行。这个能力适合范围极窄、语义明确的抑制,例如忽略一个预期不存在的可选文件;事务、锁和文件管理器通常应清理后返回假值,让失败保持可见。

若__exit__()自己抛出新异常,新异常会传播,原异常通常保留在异常上下文中。清理失败不能安全地靠返回值表示,因为调用方可能看不到它;应抛出具有操作语义的异常,并保留原始原因链。

return、break和continue不向退出方法提供异常信息,所以它们走正常退出路径。退出方法仍会执行,但不能仅凭三个None区分是哪一种控制转移;需要这种区别的设计不应把判断隐藏在协议里。

多个管理器按嵌套方式退出,内层退出方法可以改变外层看到的状态。内层若抑制异常,外层会收到三个None;内层若抛出另一个异常,外层会看到新异常。审查组合管理器时要逐层追踪,而不是把所有退出调用视为互不影响。

进入阶段的事务化

复杂的__enter__()可能依次获取多个子资源。可用一个内部ExitStack暂存已经成功的清理动作:每完成一步就登记,全部成功后再用pop_all()转移所有权给对象的长期清理栈。

这种模式把进入阶段本身变成小型事务。失败时临时栈自动展开;成功时,正式__exit__()接管全部回调。不要在成功前调用pop_all(),否则后续获取失败又会失去保护。

获取与登记之间仍应尽量没有可失败操作。例如先打开文件,紧接着调用enter_context();不要在两者之间执行解析、日志格式化或用户回调。窗口越短,所有权越容易证明。

清理异常的优先级

一个退出方法可能同时面对代码块异常与自身清理异常。覆盖原异常有时合理,例如提交失败就是最终操作失败;但若关闭日志句柄失败掩盖了主要业务异常,诊断会更困难。契约应说明优先级,并用异常链保留两者。

多层退出可能连续失败。ExitStack按注册的逆序调用退出函数,并更新当前异常上下文,模拟嵌套with的行为。不要假设第一个清理错误会让后续清理全部停止;应对所用管理器的具体契约编写测试。

生成器管理器与动态栈

@contextmanager调用被装饰函数时,并不会立即运行函数体,而是创建一个包装生成器的管理器。进入时推进到唯一的yield;正常退出时继续推进,异常退出时则把异常注入yield位置。

生成器必须恰好产生一次值。没有产生值就结束会在进入时触发RuntimeError,产生第二次值会在退出时触发RuntimeError。这不是迭代API,不能用多个yield表示多次进入。

最稳妥的基本形状是在获取之后写try: yield resource,并在finally释放。需要区分成功与失败时,再增加except做回滚并重新抛出,或增加else做提交。分支必须覆盖提交自身失败时的资源释放。

生成器在yield处捕获代码块异常后,如果正常结束,适配器会认为异常已经处理并抑制它。因此,“记录后忘记raise”不只是丢失traceback,而是改变调用方控制流。日志语句不能代替传播决定。

每次调用生成器函数都会创建一个新的管理器实例。保存实例并再次进入不是重启生成器;通常会失败。若API需要多次独立使用,应暴露工厂或可调用对象,让每次with都创建新实例。

ExitStack的三类登记

enter_context(cm)先进入管理器,再保存它的退出方法,并返回进入结果。它适合动态数量的完整上下文管理器。若传入对象不支持同步协议,Python 3.11及以后抛出TypeError。

push(exit)直接登记具有退出方法签名的可调用对象,或接管一个管理器的退出方法。它可以接收异常信息并抑制异常,但不会替你调用对应的进入方法,因此只适合已经部分获取或由别处进入的资源。

callback(func, *args, **kwargs)登记普通回调。回调不会收到异常信息,也不能抑制异常;它适合无论结果如何都执行的释放函数。三种登记都会以后进先出顺序展开。

pop_all()把整组回调转移到一个新栈,而不是执行回调。它适合“全部资源成功获取才保留,否则全部释放”的事务式进入。新栈必须由明确所有者调用close()或放入另一个with,否则只是移动了泄漏责任。

异步上下文管理器

异步协议把方法换成__aenter__()与__aexit__(),两者返回可等待对象。async with会等待进入和退出,因此适合需要网络往返、异步锁或其他异步I/O的获取与释放。它只能出现在协程函数体中。

这里的 协程(coroutine) 边界取决于清理动作,而不取决于代码块中是否有其他await。若关闭客户端必须等待,就要使用异步管理器;若只是同步关闭内存对象,普通管理器仍然合适。

取消也是异常路径。任务在代码块中被取消时,__aexit__()必须完成必要清理,并通常让CancelledError继续传播。宽泛捕获BaseException后正常返回会破坏结构化并发,使上层误以为任务成功。

异步退出本身也可能在等待时被再次取消。需要不可中断地完成哪些最小清理取决于资源库契约;不要机械地给整个退出过程套屏蔽,因为过度屏蔽会拖延关闭。应在集成测试中真实取消任务并观察连接、锁与子任务。

AsyncExitStack可以在同一栈中组合同步和异步管理器,也能登记异步回调。动态进入异步管理器使用await stack.enter_async_context(cm);显式释放使用await stack.aclose(),它没有同步close()替代品。

asynccontextmanager与同步装饰器遵循相同的一次yield规则,只是进入和退出代码可以等待。异步生成器若捕获取消或业务异常,仍必须按契约重新抛出;finally中的等待也需要取消测试。

复用、类型与测试

“可复用”表示同一个实例能用于多个不重叠的with;“可重入”还允许同一个实例在尚未退出时再次进入。可重入一定可复用,可复用不一定可重入。两者都不是协议自动提供的性质。

文件和生成器管理器实际上是一次性的。锁、suppress()等某些管理器具有自己的复用或重入语义,但只能以具体API文档为准。自定义管理器若把旧状态只保存在一个实例字段中,嵌套进入很可能覆盖外层状态。

类型注解应描述进入结果,而不只是管理器对象。同步接口可接受contextlib.AbstractContextManager[T],异步接口可接受AbstractAsyncContextManager[T];若函数需要自行控制每次生命周期,接收“返回管理器的工厂”通常比接收一次性实例更准确。

静态类型不会证明资源一定退出,也不会验证异常抑制策略。代码审查仍要检查控制流,测试仍要观察可见状态。特别是__exit__()的返回类型过宽时,类型正确的真值仍可能隐藏错误。

最小故障矩阵

一个自定义管理器至少需要覆盖以下路径:

  1. 进入成功,代码块正常结束,退出成功。
  2. 进入成功,代码块抛出预期异常,清理后继续传播。
  3. 进入过程中失败,已经部分获取的资源得到释放。
  4. 代码块失败且退出也失败,异常链符合契约。
  5. 多个管理器中后一个进入失败,先进入者按逆序退出。

只有抑制异常属于公开行为时,才增加“匹配的异常被抑制,其他异常传播”测试。不要为了覆盖返回True的分支而创造没有领域含义的抑制功能。

资源测试应断言状态,而不只断言打印或模拟方法被调用。例如检查文件确实关闭、锁可被另一执行单元取得、临时键恢复为不存在、事务内容回到旧状态。调用记录可以辅助定位,但不是生命周期契约本身。

并发环境还要测试所有权交错。同一个可复用管理器若在两个任务中共享实例字段,退出顺序可能恢复错误的旧值。除非实现明确支持并发,最安全的默认是假定实例属于一次词法使用。

延伸阅读

检查点

4个问题 · 1道输出预测题 · 1道找错题

前置内容 函数 Classes objects 即将上线
下一篇 Contextlib 即将上线 asyncio 装饰器
复制为Markdown 面试题库 在GitHub上编辑 报告错误 讲清楚了吗?