with 语句与资源管理的本质

打开文件、申请锁、建立连接、开启事务,这些操作都有一个共同点:用完之后必须释放。漏掉释放,轻则文件句柄耗尽、锁一直占着,重则事务悬空、连接池被打满。Python 用 `with` 语句把「获取资源、使用资源、释放资源」三段逻辑绑定在一起,无论中间是否抛出异常,退出语句块时释放逻辑一定执行。

`with` 的核心是协议:对象只要实现了 `__enter__` 和 `__exit__` 两个方法,就能放进 `with` 里。进入语句块时调用 `__enter__`,它的返回值绑定到 `as` 后面的变量;无论正常退出还是异常退出,都会调用 `__exit__`,由它完成清理并把异常情况交给调用方处理。

with open("data.txt", encoding="utf-8") as f:
    content = f.read()

这里 `open` 返回的文件对象实现了上下文协议,`with` 结束时文件一定被关闭,不需要手动 `f.close()`。把资源生命周期收进语句块,是 Python 里最基础也最重要的资源管理习惯。

好的资源管理不是「记得释放」,而是「想漏都漏不掉」。
本文核心知识点
  • with语句与上下文协议。
  • __enter__/__exit__类实现。
  • contextmanager装饰器。
  • ExitStack动态资源组合。
Python 上下文管理器配图:蓝调插画中对称开合式舱体结构,上下光束进出,周围多层嵌套半透明框架,体现资源开关与生命周期管理
进入与退出之间的代码段被安全包裹,资源在该进时进、该出时出。

基于类的上下文管理器

自己写上下文管理器,最直观的方式是定义一个类,实现 `__enter__` 与 `__exit__`。`__exit__` 接收异常类型、异常对象和回溯对象三个参数;如果它返回真值,异常会被吞掉不再向上传播,返回假值则异常继续抛出。

class ManagedResource:
    def __init__(self, name):
        self.name = name

    def __enter__(self):
        print("进入:获取资源", self.name)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print("退出:释放资源", self.name)
        return False

with ManagedResource("连接池") as res:
    print("使用中:", res.name)

类写法适合状态复杂的资源:可以在 `__enter__` 里做多次初始化,在 `__exit__` 里根据异常类型决定是回滚还是提交。缺点是要写的样板代码多一些,一个资源要拆成五个方法。

contextmanager 装饰器写法

大多数场景下,资源管理就是「进之前做一件事、出之后做一件事」,用 `contextlib.contextmanager` 装饰一个生成器函数就够了。函数里 `yield` 之前是进入逻辑,`yield` 之后是退出逻辑,代码顺序读下来一目了然。

from contextlib import contextmanager

@contextmanager
def timer(name):
    import time
    start = time.perf_counter()
    try:
        yield
    finally:
        print(name, "耗时", round(time.perf_counter() - start, 4), "秒")

with timer("查询接口"):
    sum(range(1_000_000))

注意 `yield` 周围一定要包 `try/finally`,否则函数体中途抛异常时,`yield` 后面的清理代码不会执行,这正好违背了上下文管理器的初衷。`yield` 也可以带返回值,调用方用 `as` 接收,实现「进入后把受管对象交给调用方」。

异步上下文管理器

在 asyncio 程序里,资源本身往往是异步的:数据库连接要 await 建立、也要 await 关闭,HTTP 会话同理。Python 3.7 起支持 `async with` 与 `__aenter__`、`__aexit__`,并且 `contextlib` 提供了对应的 `asynccontextmanager`。

from contextlib import asynccontextmanager

@asynccontextmanager
async def aio_session(url):
    import aiohttp
    async with aiohttp.ClientSession() as sess:
        async with sess.get(url) as resp:
            data = await resp.text()
        yield data
    # 退出时会话由内部 with 自动关闭

async def main():
    async with aio_session("https://example.com") as page:
        print("拿到页面长度", len(page))

异步上下文管理器与同步版本规则一致,只是每个阶段都可能是协程。FastAPI 的依赖注入同样支持 `yield` 形式的依赖,退出逻辑正好复用这套思想,实现「请求进来建立连接、请求结束释放连接」的数据库会话管理。

contextlib 工具箱与 ExitStack

`contextlib` 除了装饰器,还内置了一批即用组件。`closing` 给只有 close 方法、没有上下文协议的旧式对象补上 `with` 支持;`suppress` 指定吞掉哪些异常;`redirect_stdout` 临时把标准输出重定向到别处;`nullcontext` 在需要占位时提供一个什么都不做的上下文。

from contextlib import suppress, ExitStack, redirect_stdout
import io

with suppress(FileNotFoundError):
    open("maybe-missing.txt").read()

buf = io.StringIO()
with redirect_stdout(buf):
    print("这段文字进入 buf")

with ExitStack() as stack:
    f1 = stack.enter_context(open("a.txt", encoding="utf-8"))
    f2 = stack.enter_context(open("b.txt", encoding="utf-8"))
    # 退出时按后进先出逐个关闭

`ExitStack` 是最灵活的组件:资源数量在运行前不确定时,可以在循环里动态把资源压进栈,退出时统一按后进先出清理。它还能作为「延迟注册」的入口,让上层函数决定要不要把某个清理动作挂到栈上。