上下文管理器：with 语句与 contextlib

引言

前面几篇文章你已经认识了函数、作用域、异常处理（第四章第 8 篇）和装饰器（第四章第 5 篇）。现在把目光移到一个几乎每天都在用、却很少有人深究的语法结构上——with 语句。

你是否见过这样的代码：

f = open('data.csv')
data = f.read()
f.close()  # 万一 read() 之前抛异常了呢？

只要 f.read() 出错，f.close() 就永远不会执行，文件描述符就会泄漏。在操作系统层面，每个进程能打开的文件数是有上限的（通常 1024 个），泄漏累积到临界点程序直接崩溃。即便只是个人脚本，这种写法也在给未来的自己埋雷。更隐蔽的是，即便你写了 try/finally，代码也会变得冗长难读，尤其是在需要管理多个资源的时候。

本篇要回答三个问题：with 语句到底做了什么？谁来定义这个行为？Python 标准库为我们提供了哪些开箱即用的工具？ 读完后，你将能够自行实现上下文管理器，理解异常传播规则，并且正确使用 contextlib 模块中的核心工具。

阅读前提：熟悉函数定义（第 1 篇）、异常处理（第 8 篇）和类的基本语法（第五章第 1 篇）。本篇不引入异步上下文管理器，那是异步编程章节的内容。在开始之前，请确保你已经理解了 Python 的异常传播机制，因为上下文管理器的核心就建立在这一机制之上。

概念与原理

上下文管理器是什么

上下文管理器（Context Manager）是一个实现了特定协议的对象。所谓协议，在 Python 里不是接口类，而是一套「约定」：只要一个对象同时拥有 __enter__ 和 __exit__ 两个方法，它就是一个上下文管理器。

class FileManager:
    def __enter__(self):
        # 进入 with 块前执行：打开资源、获取锁等
        return self       # 返回值绑定到 as 变量

    def __exit__(self, exc_type, exc_val, exc_tb):
        # 离开 with 块时执行：关闭资源、释放锁等
        # 返回 True 表示"我已处理异常，不要向外传播"
        # 返回 False（或 None）表示"让异常继续传播"
        pass

with 语句只是这套协议的语法糖。当你写：

with manager as resource:
    do_something(resource)

Python 解释器实际执行的流程如下：

1. 求值 manager，确认它是一个上下文管理器（即有 __enter__ / __exit__）。
2. 调用 manager.__enter__()，将其返回值绑定到 resource。
3. 执行 with 块内部的代码。
4. 无论如何（正常退出还是抛出异常），都调用 manager.__exit__(exc_type, exc_val, exc_tb)。
   • 如果 __exit__ 返回 True，异常被吞掉，不向外传播。
   • 如果 __exit__ 返回 False，异常继续向外抛出。

这里 exc_type、exc_val、exc_tb 三个参数分别对应被抛异常的类、实例和追踪对象。如果 with 块内没有异常，它们全为 None。这三个参数是可选的——如果你的 __exit__ 不需要访问异常信息，可以直接写成 def __exit__(self, *args):。

为什么 finally 不够用

你可能会想：用 try/finally 也能做到”无论如何都清理”。没错，对于单个资源来说，try/finally 完全够用。但当资源数量增多或者需要组合多个资源时，with 语句的优势就显现出来了。

比如打开多个文件：

# try/finally 版本（繁琐且容易遗漏清理）
f1 = open('a.txt')
f2 = open('b.txt')
try:
    process(f1, f2)
finally:
    f1.close()
    f2.close()

# with 版本（简洁且自动处理嵌套异常）
with open('a.txt') as f1, open('b.txt') as f2:
    process(f1, f2)

第二个例子不仅代码更短，而且 with 的嵌套清理顺序是确定的（后进先出 LIFO），即便 open('b.txt') 抛出异常，f1 也不会被打开却无人关闭。这里有一个微妙的细节：如果 open('a.txt') 成功但 open('b.txt') 失败，f1 根本不会被赋值给任何变量，因此也不会有泄漏风险——因为 with 语句在 __enter__ 成功之后才会进入下一个 with。

__exit__ 的返回值控制异常传播

这是理解上下文管理器最关键的一点。__exit__ 返回 True 意味着它「消费」了异常，调用者不会感知到任何错误。这一特性常用于实现优雅的错误处理或重试逻辑。

下面这段代码演示两种行为的差异：

class SuppressCM:
    def __enter__(self):
        print('  entering')
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f'  exiting: {exc_type}')
        return True  # 吞掉异常

class PropagateCM:
    def __enter__(self):
        print('  entering')
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f'  exiting: {exc_type}')
        return False  # 传播异常

print('--- 吞掉异常 ---')
with SuppressCM() as cm:
    raise ValueError('will be suppressed')
print('no exception reached here')

print('--- 传播异常 ---')
try:
    with PropagateCM() as cm:
        raise ValueError('will propagate')
except ValueError:
    print('caught outside')

实际运行输出：

--- 吞掉异常 ---
  entering
  exiting: <class 'ValueError'>
no exception reached here
--- 传播异常 ---
  entering
  exiting: <class 'ValueError'>
caught outside

注意：如果 `__exit__` 自身抛出异常，那个新异常会**替换**原来的异常，成为最终的异常来源。这意味着在 `__exit__` 中应该尽量避免抛出异常，除非你是有意识地在做错误转换。

### 没有 `as` 子句的情况

`with` 语句可以不绑定返回值：

with open('data.txt') as f:
    print(f.read())

也可以省略 as：

class NoReturnCM:
    def __enter__(self):
        print('  entering')
    def __exit__(self, *args):
        print('  exiting')

with NoReturnCM():
    print('  inside')

此时 __enter__ 仍然会被调用（用于设置资源），但返回值不会被绑定到任何变量。这在一些只关心清理语义的场景下很有用，比如某些 logging 上下文管理器只需要在退出时 flush 缓冲区，不需要返回任何数据。

嵌套清理的顺序与异常传播

当多个上下文管理器嵌套使用时，Python 保证清理顺序遵循后进先出（LIFO）原则——最后进入的最先退出。这与栈的行为完全一致。

class CM:
    def __init__(self, name):
        self.name = name
    def __enter__(self):
        print(f'  enter {self.name}')
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            print(f'  release {self.name} (exception: {exc_type.__name__})')
        else:
            print(f'  release {self.name}')
        return False

with CM('A'), CM('B'):
    print('  inside')
    raise RuntimeError('boom')

运行结果：

--- acquire A, then B, exception in B ---
  acquire A
  acquire B
  inside
  release B (exception: RuntimeError)
  release A (exception: RuntimeError)
  caught outside

可以看到 B 先释放，A 后释放。这种 LIFO 清理顺序在嵌套锁、数据库事务等场景中非常重要——如果顺序反了，可能导致死锁或数据不一致。

### 内部实现细节：`with` 语句的字节码

从 CPython 的角度看，`with` 语句被编译为特殊的字节码指令。在 Python 3.10+ 中，它使用 `WITH` 指令，该指令会依次执行以下操作：调用 `__enter__`，将返回值压入栈，执行 `with` 块内的代码，然后调用 `__exit__` 并弹出返回值。如果 `with` 块内发生了异常，异常信息会被压入栈，然后调用 `__exit__`，由 `__exit__` 决定是否吞掉异常。这一底层机制保证了 `with` 语句的高效性和正确性，无需开发者手动编写 `try/finally`。

## 实现

### 手动实现一个简单的上下文管理器

我们来实现一个仿 `open()` 的文件管理器，用于教学目的（实际使用中直接用 `open()` 即可）：

class FileManager:
    def __init__(self, path, mode='r'):
        self.path = path
        self.mode = mode
        self.f = None

    def __enter__(self):
        self.f = open(self.path, self.mode)
        print(f'  opened {self.path}')
        return self.f

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.f:
            self.f.close()
            print(f'  closed {self.path}')
        return False

使用方式：

with FileManager('test.txt', 'w') as f:
    f.write('hello context manager\n')
    f.write('second line\n')
print('file closed after block')

运行结果：

  opened test.txt
file closed after block
  closed test.txt

### 使用 contextlib 简化创建

手动写类虽然直观，但很多场景下只需几行逻辑，用类显得太重。`contextlib` 模块提供了 `@contextmanager` 装饰器，让生成器函数变成上下文管理器：

from contextlib import contextmanager

@contextmanager
def managed_resource(name):
    print(f'  acquiring {name}')
    try:
        yield f'resource_{name}'
    finally:
        print(f'  releasing {name}')

使用：

with managed_resource('db') as res:
    print(f'  using: {res}')

运行结果：

--- contextmanager test ---
  acquiring db
  using: resource_db
  releasing db

`@contextmanager` 的工作原理：它把 `yield` 之前的代码当作 `__enter__`，`yield` 之后的代码（包括 `finally`）当作 `__exit__`。`yield` 的值就是 `as` 绑定的变量。

**重要细节**：如果 `with` 块内抛出异常，该异常会被传递给生成器的 `throw()` 方法，然后由 `__exit__` 逻辑处理。所以 `yield` 应该放在 `try` 中，`cleanup` 代码放在 `finally` 中，这样才能保证即使 `with` 块抛出异常，清理也一定会执行。

### contextlib 的其他实用工具

**`closing`**：适用于任何有 `close()` 方法但不实现上下文管理器协议的对象：

from contextlib import closing

class FakeStream:
    def write(self, s):
        print(f'  wrote: {s.strip()!r}')
    def close(self):
        print('  stream closed!')

with closing(FakeStream()) as s:
    s.write('hello')
    s.write('world')

运行结果：

wrote: 'hello'
wrote: 'world'
  stream closed!

**`redirect_stdout` / `redirect_stderr`**：临时重定向标准输出或标准错误，常用于测试或日志捕获：

from contextlib import redirect_stdout
import io

buf = io.StringIO()
with redirect_stdout(buf):
    print('hello to stdout')
    print('second line')

print('Captured stdout:', repr(buf.getvalue()))

运行结果：

Captured stdout: 'hello to stdout\nsecond line\n'

**`suppress`**：静默指定类型的异常，相当于 `try/except` 的缩写：

from contextlib import suppress

with suppress(ValueError, ZeroDivisionError):
    x = int('not_a_number')
    print('never reached')
print('after suppress, program continues')

运行结果：

--- suppress test ---
before
after suppress block

注意 `suppress` 只捕获指定的异常类型。如果抛出的是其他类型，异常会照常传播，程序会崩溃。这在测试中很有用，比如测试某个函数在预期异常下不会出错：

def may_fail():
    raise ValueError('expected error')

with suppress(ValueError):
    may_fail()
print('test passed - ValueError was expected')

嵌套上下文管理器

多个 with 可以嵌套或写在同一行：

with CM1() as a, CM2() as b:
    print(f'inside: a={a}, b={b}')

这等价于：

with CM1() as a:
    with CM2() as b:
        print(f'inside: a={a}, b={b}')

注意嵌套顺序：先执行的 with 最后退出（LIFO）。当 with 块内发生异常时，清理从最内层开始向外展开：

class CM:
    def __init__(self, name):
        self.name = name
    def __enter__(self):
        print(f'  enter {self.name}')
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            print(f'  release {self.name} (exception: {exc_type.__name__})')
        else:
            print(f'  release {self.name}')
        return False

with CM('A'), CM('B'):
    print('  inside')
    raise RuntimeError('boom')

运行结果：

--- acquire A, then B, exception in B ---
  acquire A
  acquire B
  inside
  release B (exception: RuntimeError)
  release A (exception: RuntimeError)
  caught outside

可以看到 B 先释放，A 后释放，这正是栈式清理的体现。

## 易错点与陷阱

### 陷阱一：`__enter__` 抛出异常时，`__exit__` 不会被调用

这是最常见的误解之一。`__exit__` 只在 `__enter__` 成功返回之后才会被调用。如果 `__enter__` 内部抛出异常，Python 认为资源根本没有成功获取，自然不需要走清理流程：

class BadEnter:
    def __enter__(self):
        print('  entering BadEnter')
        raise ValueError('fail at enter')
    def __exit__(self, *args):
        print('  exiting BadEnter (should NOT print)')
        return False

try:
    with BadEnter():
        print('inside (never reached)')
except ValueError as e:
    print(f'caught: {e}')

输出中不会出现 exiting BadEnter，只有：

  entering BadEnter
caught: fail at enter

理解这一点很重要：如果你需要在 `__enter__` 阶段也做清理（比如打开文件失败时要回滚前面的操作），必须在 `__enter__` 内部自己处理，`__exit__` 帮不上忙。

### 陷阱二：忘记 `return False` 或遗漏 `exc_type` 检查

很多人写 `__exit__` 时直接返回 `False`，但在需要吞掉异常的场景下忘记了修改返回值。另一种常见错误是在 `__exit__` 内部访问 `exc_val` 时没有检查 `exc_type` 是否为 `None`：

# 危险写法：if exc_val 在没有异常时可能为 None，但 exc_type 也是 None
if exc_val is not None:
    print(f'Handled exception: {exc_type.__name__}: {exc_val}')

正确做法是先判断 exc_type is not None，再访问 exc_val，因为当没有异常时 exc_type 为 None，而 exc_val 也为 None：

# 正确写法
if exc_type is not None:
    print(f'Handled exception: {exc_type.__name__}: {exc_val}')

虽然在这个特定场景下 exc_val is not None 也能工作，但它违背了意图表达的清晰度。用 exc_type is not None 来判断是否有异常发生更符合语义，也更容易让其他读者理解你的意图。

陷阱三：@contextmanager 中 yield 不在 try 内

from contextlib import contextmanager

@contextmanager
def broken_manager():
    print('setup')
    yield  # 没有 try/finally 包裹
    print('cleanup')

with broken_manager() as x:
    raise RuntimeError('boom')

在 CPython 3.10+ 中，如果 yield 没有放在 try 块内，当 with 块抛出异常时，生成器不会自动收到异常通知，cleanup 代码可能不会被执行，甚至抛出 RuntimeError: generator didn't stop。所以始终要把 yield 放在 try 中，把清理放在 finally 中：

@contextmanager
def safe_manager():
    print('setup')
    try:
        yield 'value'
    finally:
        print('cleanup')  # 无论是否异常都会执行

陷阱四：with 语句不能用于非上下文管理器的对象

有些对象有 open() 和 close() 方法，但没有实现 __enter__ 和 __exit__，直接使用 with 会抛出 AttributeError：

class NoCM:
    def open(self):
        print('opened')
    def close(self):
        print('closed')

obj = NoCM()
# with obj:  # 这会抛出 AttributeError
#     pass

如果你希望这类对象支持 with，可以用 contextlib.closing 包装，或者手动添加 __enter__ 和 __exit__ 方法。

工程实践中的上下文管理器模式

在实际项目中，上下文管理器不仅仅用于文件操作，还可以用于多种资源管理和流程控制场景。

数据库连接管理：这是上下文管理器最常见的应用场景之一。通过 __enter__ 获取连接，__exit__ 确保连接被正确关闭或回滚，即使在事务执行过程中发生异常也能保证数据一致性。

class DatabaseTransaction:
    def __init__(self, connection):
        self.conn = connection
    
    def __enter__(self):
        self.conn.begin()  # 开始事务
        return self.conn
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            self.conn.rollback()  # 异常时回滚
        else:
            self.conn.commit()  # 正常时提交
        return False  # 不吞掉异常

# 使用示例
with DatabaseTransaction(db) as conn:
    conn.execute("INSERT INTO users VALUES (?, ?)", (name, email))
    conn.execute("UPDATE accounts SET balance = balance - ?", (amount,))

性能测试与基准测量：用上下文管理器包装性能测试代码，自动计算和报告执行时间。这种模式在性能优化和分析中非常有用。

临时状态切换：当需要在某段代码执行期间临时改变全局状态，并在结束后恢复原状时，上下文管理器是理想的选择。比如临时修改日志级别、切换数据库连接池、修改环境变量等。

并发控制：结合 threading.Lock 或 asyncio.Lock，上下文管理器可以简化锁的获取和释放，避免死锁和资源泄漏。

小结

上下文管理器是 Python 资源管理的核心机制。with 语句通过 __enter__ / __exit__ 两个魔术方法，保证了资源在任何情况下都能被正确清理——无论正常退出还是异常退出。contextlib 模块提供了 @contextmanager、closing、redirect_stdout、suppress 等工具，让常见场景的实现成本降到最低。理解异常传播规则（__exit__ 返回 True 会吞掉异常）和清理顺序（LIFO 嵌套）是写出正确上下文管理器的关键。记住：__enter__ 失败时 __exit__ 不会被调用，@contextmanager 中的 yield 必须放在 try 里——这两个坑避开了，你就已经超过了大部分 Python 初学者。

练习与思考题

1. 实现一个计时上下文管理器：用 time.perf_counter() 测量 with 块执行时间，并在退出时打印耗时（秒），精确到微秒。示例输出：[sleep_test] took 0.050123s (50123.0us)。

2. 实现一个安全的数据库连接上下文管理器：模拟一个 DatabaseConnection 类，__enter__ 打印 "Connected to DB"，__exit__ 打印 "Disconnected from DB"。如果 with 块内抛出异常，仍然保证断开连接，并在 __exit__ 中打印异常类型。要求使用类方式实现，不用 @contextmanager。

3. 思考题：with open('f.txt') as f 内部的 open() 函数返回的对象是 _io.TextIOWrapper，它为什么天然支持 with 语句？去 Python 源码中查找 _io 模块，确认 __enter__ 和 __exit__ 的具体实现（无需背诵，了解即可）。

4. 进阶思考：为什么用 @contextmanager 实现”重试”逻辑会很困难？（提示：yield 在一个生成器生命周期内只执行一次。）如果需要重试，应该用什么模式替代上下文管理器？（答案：装饰器或显式循环，下一篇测验第 4 题会涉及。）