第 5 章上下文管理器：Python 的 defer

这一章解决一个你天天都会碰的痛点：资源（文件、连接、锁、事务）用完必须关，Go 用 defer 兜底，Python 用 with。但 with 远不止「自动关文件」这么简单——它是一套异常安全的资源管理协议。

墨叔：老哥，你回想一下 Go 里最让你安心的一句——defer f.Close()。管你函数中间 return 还是 panic，这行都给你兜底执行，资源绝不泄漏。你现在写 Python，是不是经常 f = open(...) 然后忘了关，或者 try 里 return 漏了 f.close()？

老哥：被你说中了……我一般就 try: ... finally: f.close()，但 finally 写多了又臭又长。

墨叔：对，这正是 with 要解决的。with open(...) as f: 干的事，底层就是「进入时拿资源、return/raise/正常结束都保证清理」。今天咱们把它的三块骨头拆开：__enter__/__exit__ 协议、@contextmanager 用 yield 写、还有 contextvars 在并发里的上下文隔离——最后一块，是 Go 没直接对应物的硬核货。

一、with 协议到底是什么（怎么用）

先说清楚 with 不是语法糖，它背后是两个「魔术方法」在撑腰：

__enter__(self)：进入 with 块时调用，返回的东西绑定给 as 后面的变量。
__exit__(self, exc_type, exc_val, exc_tb)：退出 with 块时一定调用，无论正常结束、遇 return、还是抛异常。三个参数是异常信息（没有异常时都是 None）。

class Connection:
    def __init__(self, dsn: str) -> None:
        self.dsn = dsn
 
    def __enter__(self) -> "Connection":
        print(f"建立连接 {self.dsn}")
        return self
 
    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        print("断开连接")          # 不管出没出事都执行
        # 返回 None（或 False）表示异常照常往外抛
        # 返回 True 表示「我吞了这异常」，外面收不到
 
    def query(self, sql: str) -> list:
        return [f"结果:{sql}"]
 
 
with Connection("postgres://db") as conn:
    rows = conn.query("select 1")
    print(rows)

你品这块代码的对称性：__enter__ 拿资源，__exit__ 还资源，中间业务逻辑干干净净。对照 Go：

func query() error {
    conn, err := dial("postgres://db")
    if err != nil { return err }
    defer conn.Close()          // 等价 __exit__，但 defer 注册在函数帧上
    return conn.Query("select 1")
}

墨叔：看出区别没？Go 的 defer 是函数级的，绑在函数栈帧；Python 的 __exit__ 是块级的，绑在 with 这个缩进块。一进 with 就 enter，一出块就 exit，哪怕同一函数里套三层 with，各自进退分明。这对「嵌套资源」特别友好——

with Connection("db") as db:
    with Lock() as lk:
        with open("audit.log") as f:
            ...   # 三个资源，按栈顺序 enter，逆序 exit

__exit__ 的返回值是个关键开关，很多人栽在这。返回「真值」等于告诉解释器「异常我已经处理了，别再往上抛」。默认一定返回 None，让异常照常传播——除非你真要吞异常，否则别手痒返回 True，那会悄悄吃掉bug。

class SuppressCloseError:
    def __enter__(self): return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is OSError:
            return True        # 只吞 OSError，其他照抛
        return False

二、什么场景该用 with（对照 Go / 边界）

with 的适用边界就一句话：任何「拿了要还」的资源，都用它。具体清单：

文件、socket：IO 句柄有限，不关会耗尽（Go 里你也会 defer f.Close()）。
数据库连接、事务：with conn.begin() as tx: 提交/回滚自动化。
锁：with lock: 进入加锁、退出释放，杜绝「忘了 unlock 死锁」（Go 的 defer mu.Unlock() 同款）。
临时状态切换：比如「临时改时区、退出还原」。

看看事务这个真实场景，多清爽——

import sqlite3
 
def transfer(conn: sqlite3.Connection, src: int, dst: int, amount: int) -> None:
    with conn:                       # sqlite 连接本身支持 with：成功提交，异常回滚
        conn.execute("UPDATE acct SET bal=bal-? WHERE id=?", (amount, src))
        conn.execute("UPDATE acct SET bal=bal+? WHERE id=?", (amount, dst))

对比 Go 你得手写 tx, _ := db.Begin(); defer tx.Rollback(); ...; return tx.Commit()，还得靠 Rollback 幂等（已提交再 Rollback 不报错）来兜底。Python 这套把「提交还是回滚」直接交给 with：块正常走完→提交，块里抛异常→回滚。语义封装在协议里，你不会写错顺序。

老哥：那是不是所有函数都该包一层 with？

墨叔：不是。with 解决的是「资源生命周期 = 代码块作用域」的场景。如果你要跨函数持有资源（比如连接池借出去给好几个调用方用），那不适合塞进一个 with，得用连接池自己管生命周期。另一个边界：计算型、纯内存、没有外部资源的东西，别硬套 with，那只会让代码变怪。

还有个坑：__exit__ 里抛的新异常会掩盖原异常。如果你在清理时又出错，Python 会用新异常替换掉原来的，原始异常就丢了。清理代码要自己兜底好，别在里面裸奔。

三、收益与代价（为什么这么设计）

收益摆三块：

异常安全。最值钱的好处——任何退出路径（正常、return、raise）资源都回收。try/finally 也能做到，但 with 把样板收进协议，业务代码不再被 finally 污染。
可读性。资源的「生」和「死」贴在代码同一个位置看，心智负担小。
可组合。一个对象实现协议，就能进任意 with，还能 contextlib.closing 给那些没有协议的老对象补一层（比如某些返回 file-like 的库）。

代价呢？

心智模型：你得理解「异常信息三个参数」这套机制，新人容易在 __exit__ 返回值的语义上栽跟头（吞还是不吞）。
调试栈：with 块抛出的异常，栈顶会经过 __exit__，调试时多看一层。
和 defer 的思维切换：Go 里 defer 是「函数末尾执行」，可以写多个、后进先出；Python with 是「块末尾」，作用域更窄。你从 Go 过来最容易犯的错是：把该写进 with 的资源清理，习惯性地写成函数末尾的「类 defer」逻辑，结果函数里提前 return 就漏了——在 Python 里，请一律交给 with。

四、@contextmanager：用 yield 写管理器

每次都写 __enter__/__exit__ 两个方法太啰嗦。标准库 contextlib 给你一个装饰器，一行 yield 把函数劈成两半：yield 之前是 __enter__（yield 的值就是 as 拿到的），yield 之后是 __exit__。

from contextlib import contextmanager
import time
 
@contextmanager
def timer(label: str):
    start = time.perf_counter()
    try:
        yield                    # 这里「暂停」，把控制权交给 with 块
    finally:                     # with 块的任何退出都走到这
        cost = time.perf_counter() - start
        print(f"{label} 耗时 {cost:.3f}s")
 
with timer("批量导入"):
    sum(i * i for i in range(100_000))

注意我包了 try/finally：yield 那行如果 with 块抛异常，异常会从 yield 处重新抛出到 try 里，于是 finally 执行清理，异常继续往上冒。如果你不包 try，yield 抛出的异常会直接炸出来、后面的清理代码不执行——这点务必记牢。

再看个贴近项目的：临时切换日志级别，退出还原。

import logging
from contextlib import contextmanager
 
@contextmanager
def temp_level(logger: logging.Logger, level: int):
    old = logger.level
    logger.setLevel(level)
    try:
        yield logger
    finally:
        logger.setLevel(old)     # 无论出啥事，日志级别还原
 
with temp_level(logging.getLogger("biz"), logging.DEBUG):
    logging.getLogger("biz").debug("这条在 DEBUG 下才打")

这种「进入改状态、退出还原」的模式，@contextmanager 写起来比双方法类短一半。但要注意：它本质是生成器，yield 只能有一个。如果你需要管理多个资源、或 enter/exit 逻辑不对称到很复杂，老老实实写类更清楚。

老哥：那 yield 抛异常时，我在 with 块外能 except 到吗？

墨叔：能。异常从 yield 点冒出来，先经过你 try/finally 的 finally（执行清理），然后继续向外传播，你在 with 外面 try/except 正常接住。协议层和类写法完全一致。这就是「异常安全」在两种写法下都成立的保证。

五、contextvars：并发时代的上下文隔离

墨叔：这块是 Go 工程师最容易低估的。你想想 Go 的 context.Context——它顺着调用链传，goroutine 之间靠「显式传参」隔离。Python 早期没这东西，于是出了个大坑：asyncio 里，多个协程跑在同一个线程、共享全局变量，你想给「当前请求」挂个 request_id 做链路追踪，全局变量会被所有请求混在一起。

contextvars 就是为这个诞生的：它让变量「按执行流（任务/线程）隔离」，每个协程/线程拿到自己的副本。

import contextvars, asyncio
 
request_id: contextvars.ContextVar[str] = contextvars.ContextVar("request_id")
 
async def handle(rid: str) -> None:
    request_id.set(rid)              # 只影响「当前这个协程」的视图
    await step_a()
    await step_b()
 
async def step_a() -> None:
    print("step_a 看到", request_id.get())
 
async def step_b() -> None:
    print("step_b 看到", request_id.get())
 
async def main() -> None:
    await asyncio.gather(handle("req-1"), handle("req-2"))
    # 两个请求各自的 request_id 互不串台

关键点：request_id.set("req-1") 在 handle 协程里设的值，step_a/step_b 能读到，但另一个并发的 handle("req-2") 读到的是自己的 req-2。它们是「逻辑上同一调用链」共享，跨链隔离。这跟 Go「每个请求新建一个 ctx 往下传」一个意思，只是 Python 不用你手写参数传递，框架（FastAPI 等）在中间件里 set 一次，后面处理函数直接 get 就拿到当前请求的上下文。

线程场景同理：每个线程一份副本，线程池里跑任务也不会串。

还有个高级玩法——copy_context() 配合 context.run()，能把一个上下文「快照」下来交给新任务：

ctx = contextvars.copy_context()
ctx.run(some_func)        # some_func 及其调用链看到的 contextvars 是 ctx 的副本

这在写中间件、测试隔离、把请求上下文注入后台任务时极有用。代价是：它不是全局变量的银弹，你得想清楚「这状态该不该跟随执行流」——能当参数传的就传参，contextvars 留给「横切、贯穿全链路、又不该写进每个函数签名」的状态（trace id、当前用户、locale）。滥用它会让数据流变隐晦，调试时不知道值哪来的。

六、贴近项目的小例子：带超时的数据库重试块

把前面几块揉一起：@contextmanager + contextvars + 异常安全，写一个「带超时与链路 id 的查询上下文」，项目里能直接套。

import contextvars, time, sqlite3
from contextlib import contextmanager
 
trace_id: contextvars.ContextVar[str] = contextvars.ContextVar("trace_id", default="-")
 
@contextmanager
def db_session(conn: sqlite3.Connection, tid: str, timeout: float = 2.0):
    token = trace_id.set(tid)              # 进入：标记当前链路
    deadline = time.perf_counter() + timeout
    try:
        cur = conn.cursor()
        yield cur
        conn.commit()                       # 正常走完提交
    except sqlite3.Error as e:
        conn.rollback()                     # 出错回滚
        print(f"[{trace_id.get()}] 查询失败已回滚: {e}")
        raise
    finally:
        trace_id.reset(token)               # 退出：还原，避免串到下一个请求
        if time.perf_counter() > deadline:
            print(f"[{tid}] 查询超阈值 {timeout}s")
 
with db_session(sqlite3.connect(":memory:"), "trace-abc") as cur:
    cur.execute("CREATE TABLE t(x)")
    cur.execute("INSERT INTO t VALUES(1)")

这里 db_session 替你干了四件事：挂链路 id、提交/回滚二选一、异常安全、退出还原上下文。with 块里你只写业务 SQL。对照 Go，这基本是「defer tx.Rollback() + ctx 注入 + defer trace reset」三件套的 Python 合体版，但你没写一个 finally 也没引一个 defer。

七、contextlib 进阶：ExitStack 与 closing

墨叔：前面说的 with 都是「写死几个资源、静态嵌套」。但真实项目常遇到资源数量运行时才定的情况——比如你要同时关 N 个临时文件，N 是循环里算出来的；或者「根据配置动态决定进哪些上下文」。这时候把 with 套 N 层写不出来，得用 contextlib.ExitStack：它是一个「栈」，运行时往里 enter_context() 塞多少个管理器都行，退出时逆序统一清理。

from contextlib import ExitStack
 
def merge_sources(paths: list[str]) -> list[str]:
    with ExitStack() as stack:
        files = [stack.enter_context(open(p, encoding="utf-8")) for p in paths]
        # 不管 paths 几个，退出 with 时全部按逆序关闭
        return [line for f in files for line in f]

stack.enter_context(cm) 等价于「把 cm 的 __enter__ 执行、并把它的 __exit__ 注册进栈」，返回的还是 as 该拿的东西。这样循环里动态开多少资源，都能保证异常安全回收——这是 Go 的 defer 列表（你在 for 里反复 defer f.Close()）的 Python 对应物，而且 Python 帮你管了逆序和异常传播。

另一个高频工具是 closing：有些老库返回的对象有 .close() 但没实现上下文管理器协议（没有 __enter__/__exit__），with 直接套会 AttributeError。closing 给它包一层「伪协议」，让它能进 with：

from contextlib import closing
import urllib.request
 
with closing(urllib.request.urlopen("https://example.com")) as resp:
    body = resp.read()      # 退出时自动 resp.close()

你也可以给自己的类补协议：与其要求每个调用方手写 closing，不如在你的类上直接实现 __enter__/__exit__（通常 __enter__ 返回 self、__exit__ 调 self.close()），一劳永逸。

还有 nullcontext（Python 3.10+），在「有时要 with、有时不要」时当占位符，避免写 if 分支：

from contextlib import nullcontext
 
def run(verbose: bool):
    cm = timer("task") if verbose else nullcontext()   # 占位，啥也不做
    with cm:
        do_work()

收益：ExitStack 解决「动态数量资源的异常安全」，是普通 with 的有力补充；closing/nullcontext 让协议适配更顺。代价：ExitStack 多一层抽象，调试栈更深；且因为资源是运行时注册，静态读代码时不容易一眼看出「到底关了哪些」——所以资源数量固定时，宁可写显式嵌套 with，别为炫技用 ExitStack。

八、内置的实用管理器：suppress 与 redirect

墨叔：最后给你两个标准库自带的「小工具型」上下文管理器，平时写脚本、写测试特别顺手，不用自己造轮子。

第一个 contextlib.suppress：你想「忽略某种异常、但其他异常照常抛」，普通写法是 try: ... except X: pass，又臭又容易把 pass 的范围写大。suppress 让意图一行说清：

from contextlib import suppress
import os
 
with suppress(FileNotFoundError):
    os.remove("tmp.lock")       # 文件不在就当没这事，其他异常照样冒

对照 Go，这就是 if err == nil || !errors.Is(err, fs.ErrNotExist) { return err } 的简化版，但 suppress 把「忽略哪些异常」写在 with 参数里，范围精确锁在块内，不会误吞块外的异常。代价：新手容易 suppress(Exception) 一把忽略，把真 bug 也吃了——只 suppress 你明确知道可忽略的具体异常类型，绝不用 Exception。

第二个 contextlib.redirect_stdout / redirect_stderr：把 print、标准错误临时重定向到别处，退出还原。写测试时捕获某个函数打印了啥，不用改函数签名：

from contextlib import redirect_stdout
import io
 
buf = io.StringIO()
with redirect_stout(buf):
    print("hello")             # 这行不出现在终端，进了 buf
print(buf.getvalue())          # "hello\n"

真实价值：你有个老函数用 print 打日志（不该这么写，但遗产代码常有），你想在测试里断言它打印了特定内容，又不想让测试输出被污染——redirect_stdout 接住，断言完事。退出 with 自动还原回真正的 stdout，不污染后续。

这两个加上前面的 closing/nullcontext/ExitStack，就是你 contextlib 工具箱的常备六件套。一句话记法：凡是「进入做 A、退出还原/收尾」的模式，先想 contextlib 有没有现成的，没有再自己写协议或 @contextmanager。

【思考题】

你写过一个 with 块，里面 raise ValueError，但 __exit__ 你手滑 return True。外面能 except 到这个 ValueError 吗？为什么？如果 __exit__ 里自己又 raise RuntimeError，外面看到的是哪个异常？
@contextmanager 写的 timer，如果 with 块里抛异常，但你的生成器没包 try/finally（直接 start=...; yield; print(耗时)）。会发生什么？耗时还打不打得出来？和「类写法」比，哪里更危险？
你项目里做链路追踪，现在用全局变量 g_request_id。并发上来后不同请求串台了。用 contextvars 改造，需要改每个函数的签名吗？哪些场景你仍然应该坚持传参而不是用 contextvars？

【参考答案】

第一问：return True 等于告诉解释器「异常我吞了」，外面收不到 ValueError——这正是很多人「明明抛了错程序却没反应」的根因。第二问：如果 __exit__ 自身又 raise RuntimeError，这个新异常会取代原来的 ValueError 往外抛，原始异常丢失（除非用异常链 raise ... from e 保留）。教训：清理代码务必 try/finally 包好、别在里面裸抛；要吞异常想清楚再 return True；想同时保留原异常和加信息，用 raise NewError() from original。
没包 try/finally 时，yield 那行把 with 块的异常「原样传回」生成器，生成器在 yield 之后还有 print(耗时)，但异常会直接从 yield 处继续向外冒，后面的 print 不会执行——所以耗时打不出来，资源清理若写在 yield 之后也不会做。危险在于：@contextmanager 把清理逻辑藏在 yield 之后，一旦忘了 try/finally，异常一来清理就跳过了，而且代码看起来「好像写全了」。类写法里 __exit__ 是独立方法，框架保证调用，反而更难漏。所以写生成器版管理器，清理一律放 finally。
不需要改函数签名——这正是 contextvars 的卖点：在请求入口（中间件/处理器）trace_id.set(...) 一次，下游任意深度的函数直接 trace_id.get()，不用把 tid 当参数层层传。对照 Go 是「每个函数显式收 ctx」，Python 用隐式副本换来了签名干净。但仍该传参的场景：① 该状态只属于某几个函数、不是全链路横切的（传参更显式、更易测）；② 你希望在单元测试里精确控制输入、不依赖全局/隐式状态；③ 跨进程/跨线程「真正新起一个不继承当前上下文的任务」——contextvars 默认会继承，若你不想继承就得显式 copy_context() 控制。总原则：横切全链路且不该进签名的（trace id、当前用户、locale）用 contextvars；局部业务状态坚持传参。

徐术挺-漫思录

探索

05-上下文管理器

第 5 章上下文管理器：Python 的 defer

一、with 协议到底是什么（怎么用）

二、什么场景该用 with（对照 Go / 边界）

三、收益与代价（为什么这么设计）

四、@contextmanager：用 yield 写管理器

五、contextvars：并发时代的上下文隔离

六、贴近项目的小例子：带超时的数据库重试块

七、contextlib 进阶：ExitStack 与 closing

八、内置的实用管理器：suppress 与 redirect

目录

反向链接

徐术挺-漫思录

探索

05-上下文管理器

第 5 章 上下文管理器：Python 的 defer

一、with 协议到底是什么（怎么用）

二、什么场景该用 with（对照 Go / 边界）

三、收益与代价（为什么这么设计）

四、@contextmanager：用 yield 写管理器

五、contextvars：并发时代的上下文隔离

六、贴近项目的小例子：带超时的数据库重试块

七、contextlib 进阶：ExitStack 与 closing

八、内置的实用管理器：suppress 与 redirect

目录

反向链接

第 5 章上下文管理器：Python 的 defer