上下文管理器与 with（Python 从精通到入门 · 04）

with open(...) 里的 with 到底是什么？它是一套「进入—退出」协议，管的是资源、状态和异常。

你将学到

• with 语句的完整执行流程，以及 __enter__ / __exit__ 的分工
• __exit__ 的返回值语义：为什么 return True 能吞掉异常
• contextlib 全家桶：contextmanager / closing / suppress / ExitStack
• 用上下文管理器做计时、事务、临时切换状态

前置知识

03 - 装饰器：从函数到框架。本篇会大量使用生成器与装饰器。

---

一、为什么需要 with

# ❌ 手动管理：中途抛异常，close 就没了
f = open("data.txt", "w")
f.write("hello")
raise RuntimeError("中途出错")     # 这行之后 f.close() 永远不会执行
f.close()

# ✅ with 保证退出时一定收尾
with open("data.txt", "w") as f:
    f.write("hello")
    raise RuntimeError("中途出错")   # 异常照样向上抛，但 f 一定被关闭

with 要解决的从来不是「少写一行 close」，而是在任意退出路径（正常、异常、return、break）上都能可靠收尾。

二、with 的执行流程

只要对象实现了 __enter__ / __exit__，它就能被 with 使用：

class Managed:
    def __enter__(self):
        print("enter")
        return "resource"            # 返回值会被 as 接收
    def __exit__(self, exc_type, exc, tb):
        print(f"exit exc_type={exc_type}")
        return False

with Managed() as r:
    print("body, r =", r)

# 输出:
# enter
# body, r = resource
# exit exc_type=None

with expr as var: 等价于下面这段（CPython 的真实语义）：

mgr = Managed()
r = mgr.__enter__()
try:
    print("body, r =", r)
except BaseException as e:                      # 出了异常
    if not mgr.__exit__(type(e), e, e.__traceback__):
        raise                                   # 返回值是假值 -> 继续向上抛
else:                                           # 正常结束
    mgr.__exit__(None, None, None)

三条要点：

1. __enter__ 的返回值赋给 as 后面的名字；
2. __exit__ 的三个参数是异常三元组，没有异常时全是 None；
3. __exit__ 返回真值，异常就被吞掉；返回假值（含 None）则继续上抛。

三、__exit__ 的返回值语义

class Suppress:
    def __enter__(self):
        return self
    def __exit__(self, exc_type, exc, tb):
        print(f"捕获 {exc_type.__name__}，但我不抛了")
        return True                  # ✅ 真值：吞掉异常

with Suppress():
    raise ValueError("boom")
print("继续执行")
# 输出: 捕获 ValueError，但我不抛了 / 继续执行

class Reraise:
    def __enter__(self):
        return self
    def __exit__(self, exc_type, exc, tb):
        return False                 # ✅ 假值：放行异常

try:
    with Reraise():
        raise ValueError("boom")
except ValueError as e:
    print("被外层捕获:", e)           # 输出: 被外层捕获: boom

另外两点容易忽略：

• 如果 __exit__ 自己 raise 了新异常，它会替换掉原本正在传播的异常；
• __exit__ 拿到的异常信息是只读的，想「翻译」异常就在里面 raise 新异常。

四、contextlib.contextmanager：用生成器写上下文

手写类太啰嗦。标准库给了装饰器，可以把一个生成器函数直接变成上下文管理器：

import contextlib, time

@contextlib.contextmanager
def timing(label):
    t0 = time.perf_counter()
    try:
        yield                        # yield 之前 = __enter__；之后 = __exit__
    finally:
        print(f"{label} 耗时 {time.perf_counter() - t0:.4f}s")

with timing("sleep"):
    time.sleep(0.1)
# 输出: sleep 耗时 0.10xx s

• yield 之前的代码相当于 __enter__；
• yield 的值就是 as 拿到的东西；
• yield 之后的代码相当于 __exit__；异常会在 yield 这一行抛出。

@contextlib.contextmanager
def opened(path, mode="r"):
    f = open(path, mode)
    try:
        yield f                      # with ... as f 拿到的就是这个文件对象
    finally:
        f.close()                    # 无论如何都关闭

想「吞掉」异常，就在 yield 外面 try/except 后让生成器正常结束；想让异常继续传播，就别捕获，或者捕获后再 raise。

五、closing / suppress / ExitStack

from contextlib import closing, suppress
import urllib.request

# closing：退出时调用对象的 close()
with closing(urllib.request.urlopen("")) as resp:
    print(resp.status)               # 输出: 200

# suppress：静默忽略指定异常，比 try/except/pass 更直白
with suppress(FileNotFoundError):
    import os
    os.remove("不存在的文件.txt")
print("没崩")                        # 输出: 没崩

资源数量不确定时，用 ExitStack——它按「后进先出」的顺序统一收尾：

from contextlib import ExitStack

def read_all(paths):
    with ExitStack() as stack:
        files = [stack.enter_context(open(p)) for p in paths]
        return [f.read() for f in files]
    # 离开 with 时，所有文件按进入顺序的逆序自动关闭

ExitStack 的杀手锏是 stack.callback(fn)：临时注册任意收尾函数，非常适合「动态拼装资源」的写法。

六、实战：事务与临时切换状态

事务：正常才提交，出错就回滚，并把异常继续抛出去：

import contextlib

@contextlib.contextmanager
def transaction(conn):
    conn.begin()
    try:
        yield conn
        conn.commit()                # 正常结束才提交
    except Exception:
        conn.rollback()              # 出错回滚
        raise                        # 别把异常吞了

class Conn:
    def begin(self): print("BEGIN")
    def commit(self): print("COMMIT")
    def rollback(self): print("ROLLBACK")

with transaction(Conn()):
    print("执行 SQL")
# 输出: BEGIN / 执行 SQL / COMMIT

临时切换状态：进入时改、退出时还原，天然支持嵌套：

@contextlib.contextmanager
def temporary_attr(obj, **kwargs):
    old = {k: getattr(obj, k) for k in kwargs}   # 先备份
    for k, v in kwargs.items():
        setattr(obj, k, v)
    try:
        yield obj
    finally:
        for k, v in old.items():                 # 退出时还原
            setattr(obj, k, v)

class Config:
    debug = False

with temporary_attr(Config, debug=True):
    print(Config.debug)     # 输出: True
print(Config.debug)         # 输出: False

contextlib.redirect_stdout、suppress、ExitStack 全都是这个套路；pytest 的 fixture、数据库连接池、文件锁也都在此之上。

常见坑

坑 1：__exit__ 没返回 True，却以为异常被处理了。

class M:
    def __enter__(self): return self
    def __exit__(self, *a): print("ignored")

# ❌ 只打印，不返回真值，异常照样往外抛
with M():
    raise ValueError("照样抛")

# ✅ 明确返回 True 才算吞掉
class M2:
    def __enter__(self): return self
    def __exit__(self, *a): return True

坑 2：contextmanager 里 yield 之后没写 try/finally。

# ❌ body 抛异常时，yield 之后的 close 永远执行不到
@contextlib.contextmanager
def opened(path):
    f = open(path)
    yield f
    f.close()          # 异常路径被跳过

# ✅ 把收尾放进 finally
@contextlib.contextmanager
def opened(path):
    f = open(path)
    try:
        yield f
    finally:
        f.close()

坑 3：生成器上下文管理器只能 yield 一次。

# ❌ 多 yield 会让 contextlib 报错
@contextlib.contextmanager
def bad():
    yield 1
    yield 2            # RuntimeError: generator didn't stop

# ✅ 一个 with 只对应一次进入/退出，需要多个资源就用 ExitStack

小结

• with 的价值在于任意退出路径都能收尾，而不是少写一行 close()。
• 协议只有两个方法：__enter__ 返回值给 as；__exit__ 收到异常三元组，返回真值则吞异常。
• __exit__ 里 raise 新异常会替换原异常。
• @contextlib.contextmanager 用「yield 前 / yield 后」映射 __enter__ / __exit__，且只允许 yield 一次。
• closing 管 close，suppress 忽略指定异常，ExitStack 管不确定数量的资源。
• 事务、临时改配置、计时——凡是「成对操作」，都该写成上下文管理器。

延伸阅读

• Python 官方文档 contextlib
• 《Fluent Python》第 18 章「With, Match, and Else Blocks」
• 读一遍 pytest 的 fixture 实现，看它如何用上下文管理器管理测试资源