← 返回Python标准库精讲目录
← 返回学习笔记首页
专题: Python标准库精讲系统学习
关键词: Python, 标准库, contextlib, 上下文管理器, contextmanager, with, ExitStack, suppress, closing
一、上下文管理器协议
上下文管理器(Context Manager)是Python中一套用于资源管理的协议,它通过 with 语句让开发者能够以简洁、安全的方式管理资源(如文件、锁、网络连接、数据库会话等)的生命周期。其核心思想是:无论代码块正常执行还是抛出异常,都能确保资源被正确释放。
1. __enter__ 和 __exit__ 协议
任何实现了 __enter__ 和 __exit__ 两个特殊方法的对象都可以作为上下文管理器使用。
__enter__(self) :进入运行时上下文时调用。返回值会被 with 语句的 as 子句捕获。
__exit__(self, exc_type, exc_val, exc_tb) :离开运行时上下文时调用。三个参数分别代表异常类型、异常值和回溯信息。若方法返回 True,则异常被抑制,不会向外传播。
# 自定义上下文管理器 —— 文件资源管理
class FileManager:
def __init__(self, filename, mode='r'):
self.filename = filename
self.mode = mode
self.file = None
def __enter__(self):
print(f'[进入] 打开文件: {self.filename}')
self.file = open(self.filename, self.mode, encoding='utf-8')
return self.file # as 变量捕获此返回值
def __exit__(self, exc_type, exc_val, exc_tb):
if self.file:
self.file.close()
if exc_type:
print(f'[退出] 发生异常: {exc_type.__name__}: {exc_val}')
else:
print(f'[退出] 正常关闭文件')
return False # False 表示异常向外传播;True 表示抑制异常
2. with 语句的执行流程
with EXPR as VAR: 语句的完整执行流程可以分为四个阶段:
获取上下文管理器 :执行表达式 EXPR,得到一个上下文管理器对象。
调用 __enter__ :调用上下文管理器的 __enter__ 方法,其返回值绑定到 VAR(若使用了 as 子句)。
执行代码体 :执行 with 块内的代码。
调用 __exit__ :无论代码体是否抛出异常,都会调用 __exit__ 方法。若代码体正常结束,三个异常参数均为 None;若抛出异常,则传入对应的异常信息。
要点说明: 与传统 try/finally 相比,with 语句更加简洁且意图清晰。异常处理方面,若 __exit__ 返回 True,则异常被吞没(相当于 try 块中捕获异常但不重新抛出);若返回 False,则异常沿调用栈继续传播。在绝大多数情况下,应当返回 False 或 None 以让异常正常传播,仅在明确需要抑制异常时才返回 True。
# 执行流程演示
class FlowDemo:
def __enter__(self):
print('1. 调用 __enter__')
return 'hello'
def __exit__(self, exc_type, exc_val, exc_tb):
print(f'4. 调用 __exit__ (异常类型={exc_type})')
return False
print('--- 正常流程 ---')
with FlowDemo() as val:
print(f'2. 进入代码体, val={val}')
print('3. 代码体正常结束')
print()
print('--- 异常流程 ---')
with FlowDemo() as val:
print(f'2. 代码体中抛出异常')
raise ValueError('出错了')
print('这行不会被执行')
# 输出:
# --- 正常流程 ---
# 1. 调用 __enter__
# 2. 进入代码体, val=hello
# 3. 代码体正常结束
# 4. 调用 __exit__ (异常类型=None)
#
# --- 异常流程 ---
# 1. 调用 __enter__
# 2. 代码体中抛出异常
# 4. 调用 __exit__ (异常类型=)
# --- 随后异常向外传播 ---
官方文档: 上下文管理器是 Python 中用于运行时上下文(runtime context)的对象。with 语句在 PEP 343 中被引入,它使得异常处理和资源清理可以被优雅地封装和重用。
二、@contextmanager装饰器
contextlib.contextmanager 装饰器提供了一种更加简洁的创建上下文管理器的方式。你只需要编写一个生成器函数,用 yield 语句将函数体分割为两个部分:yield 之前的代码对应 __enter__ 的逻辑,yield 之后的代码对应 __exit__ 的逻辑。
1. 基本用法:yield 分割(前=进入 / 后=退出)
使用 @contextmanager 装饰的生成器函数,yield 语句之前的代码在 __enter__ 阶段执行,yield 之后的代码在 __exit__ 阶段执行。yield 表达式的值将成为 __enter__ 的返回值,被 as 子句捕获。
from contextlib import contextmanager
@contextmanager
def managed_file(filename, mode='r'):
"""使用 @contextmanager 简化文件管理器"""
print(' [进入] 打开资源')
f = open(filename, mode, encoding='utf-8')
try:
yield f # f 的值被 as 捕获
finally:
print(' [退出] 关闭资源')
f.close()
# 使用方式与基于类的上下文管理器完全一致
with managed_file('test.txt', 'w') as f:
f.write('Hello, contextmanager!')
上述代码等价于基于类的 FileManager。关键在于两点:第一,使用了 try/finally 结构来确保资源清理;第二,yield 的值会传递给 as 子句。
2. 异常处理机制
当 with 块内部抛出异常时,异常会在 yield 语句处被重新抛出。此时可以利用 try/except/finally 结构在生成器内部捕获并处理异常:
@contextmanager
def safe_resource(name):
print(f' 获取资源: {name}')
resource = {'name': name, 'data': []}
try:
yield resource
except Exception as e:
print(f' 捕获异常: {e}')
# 如果希望异常被抑制(不向外传播),则不重新抛出
# 如果希望异常继续传播,则重新 raise
raise
finally:
print(f' 释放资源: {name}')
print('--- 正常使用 ---')
with safe_resource('连接A') as res:
res['data'].append('查询结果')
print()
print('--- 异常使用 ---')
try:
with safe_resource('连接B') as res:
raise RuntimeError('网络中断')
except RuntimeError:
print(' 外层捕获到 RuntimeError')
关键理解: @contextmanager 的本质是把生成器函数包装成一个上下文管理器类。yield 之前的代码在 __enter__ 中执行,yield 语句本身在 __enter__ 中返回 yield 值,yield 之后的代码在 __exit__ 中执行。异常通过生成器的 throw() 方法注入到 yield 点。
3. yield 返回值细节
yield 语句可以同时作为"产出值"和"接收值"的双向通道。除了可以 yield 一个值给 as 子句外,@contextmanager 还允许你在 yield 处接收 __exit__ 传入的异常信息——不过这种用法较为少见。在实践中,只需记住:yield 的值作为 __enter__ 的返回值即可。
@contextmanager
def greeting(name):
print(f'你好, {name}!')
# yield 的值会成为 __enter__ 的返回值
yield f'欢迎 {name} 进入上下文'
print(f'再见, {name}!')
with greeting('Alice') as msg:
print(f'在上下文中收到: {msg}')
# 输出:
# 你好, Alice!
# 在上下文中收到: 欢迎 Alice 进入上下文
# 再见, Alice!
三、实用上下文工具
contextlib 模块还提供了一系列开箱即用的上下文工具,用于解决常见的资源管理场景,无需手动编写上下文管理器。
1. closing —— 自动调用 close 方法
closing(thing) 返回一个上下文管理器,在退出时无条件调用传入对象的 close() 方法。适用于那些提供了 close() 方法但本身不是上下文管理器的对象(如 urllib.request.urlopen 返回的对象)。
from contextlib import closing
from urllib.request import urlopen
# urlopen 返回的对象有 close() 但不是上下文管理器
# 使用 closing 包装后即可用于 with 语句
with closing(urlopen('https://www.python.org')) as response:
html = response.read(1024)
print(f'读取了 {len(html)} 字节')
# 自动调用 response.close()
# closing 的实现出奇地简单:
# class closing:
# def __init__(self, thing):
# self.thing = thing
# def __enter__(self):
# return self.thing
# def __exit__(self, *exc_info):
# self.thing.close()
2. nullcontext —— 无操作上下文
nullcontext(enter_result=None) 返回一个什么也不做的上下文管理器。在某些需要条件性地使用上下文管理器的场景中非常有用——它充当了"空操作"的占位符,避免了编写分支逻辑。
from contextlib import nullcontext
import threading
lock = threading.Lock()
data = []
def append_item(item, use_lock=False):
"""根据 use_lock 决定是否加锁"""
# 若 use_lock=True → 使用 lock 上下文
# 若 use_lock=False → 使用 nullcontext(无操作)
ctx = lock if use_lock else nullcontext()
with ctx:
data.append(item)
# nullcontext().__exit__() 什么也不做
# 在不需要锁时,避免了写 if/else 分支
append_item('A', use_lock=False)
append_item('B', use_lock=True)
print(f'data = {data}')
# 也支持传入 enter_result 参数
with nullcontext('default_value') as val:
print(val) # 输出: default_value
3. suppress —— 忽略指定异常
suppress(*exceptions) 用于抑制指定的异常类型。当 with 块内抛出了被抑制的异常时,异常被静默地忽略掉,程序继续正常执行。相比使用 try/except/pass,suppress 的意图更加明确。
from contextlib import suppress
import os
# 场景:尝试删除一个可能不存在的文件
# 传统方式:
try:
os.remove('temp.txt')
except FileNotFoundError:
pass
# 使用 suppress(更简洁、意图更清晰):
with suppress(FileNotFoundError):
os.remove('temp.txt')
# 可以同时抑制多种异常类型
with suppress(FileNotFoundError, PermissionError):
os.remove('/etc/protected_config.ini')
# 注意:suppress 抑制的是整个代码块中出现的指定异常
# 如果抛出的是未被抑制的异常类型,仍然正常传播
工具 作用 典型场景
closing 自动调用 close() 旧版API、非上下文管理器对象
nullcontext 无操作占位 条件性上下文、接口兼容
suppress 忽略指定异常 文件清理、资源释放中的异常忽略
四、重定向工具
contextlib 提供了两个用于临时重定向标准输出和标准错误流的上下文工具,在编写测试、捕获日志或静默输出时非常实用。
1. redirect_stdout —— 临时重定向标准输出
redirect_stdout(new_target) 将 sys.stdout 临时替换为 new_target 对象,with 块结束后自动恢复原来的 stdout。任何写入 stdout 的输出(包括 print() 函数)都会被重定向。
from contextlib import redirect_stdout
import io
# 将输出捕获到 StringIO 缓冲区
buffer = io.StringIO()
with redirect_stdout(buffer):
print('这是被重定向的输出')
print('它们不会显示在控制台上')
help(print) # help() 的输出也会被重定向
# 获取捕获的内容
output = buffer.getvalue()
print(f'捕获到 {len(output)} 字符')
# 也可以重定向到文件
with open('output.log', 'w', encoding='utf-8') as f:
with redirect_stdout(f):
print('这条信息写入日志文件')
print('控制台是干净的')
2. redirect_stderr —— 临时重定向标准错误
redirect_stderr(new_target) 与 redirect_stdout 类似,区别在于它重定向的是 sys.stderr。适用于捕获异常栈信息、日志框架的错误输出等。
from contextlib import redirect_stderr
import sys
import io
# 将错误输出捕获到缓冲区
err_buffer = io.StringIO()
with redirect_stderr(err_buffer):
print('错误信息', file=sys.stderr)
try:
1 / 0
except ZeroDivisionError:
print('除零异常', file=sys.stderr)
# 检查捕获的错误输出
print(f'stderr 捕获内容: {err_buffer.getvalue()}')
# 实际应用:将 stderr 同时写入文件和终端
class Tee:
"""同时输出到多个流"""
def __init__(self, *streams):
self.streams = streams
def write(self, text):
for s in self.streams:
s.write(text)
def flush(self):
for s in self.streams:
s.flush()
with open('error.log', 'a') as log:
tee = Tee(sys.stderr, log)
with redirect_stderr(tee):
print('这条错误信息同时写入日志文件', file=sys.stderr)
使用注意: redirect_stdout 和 redirect_stderr 不是线程安全的。在多线程环境中重定向 stdout/stderr 会影响整个进程的所有线程。在这类场景中,建议使用日志模块(logging)替代 print(),而非依赖重定向工具。
五、ExitStack栈管理
ExitStack 是 contextlib 中最强大、最灵活的上下文管理工具。它实现了一个"退出栈"(exit stack),允许在运行时动态地添加和移除上下文管理器,所有注册的上下文管理器的 __exit__ 方法将在 ExitStack 退出时以后进先出(LIFO)的顺序依次调用。
1. enter_context —— 动态添加上下文管理器
enter_context(cm) 将一个上下文管理器压入栈中,并返回其 __enter__ 的返回值。可以在 with 块内多次调用,动态管理任意数量的资源。
from contextlib import ExitStack
with ExitStack() as stack:
# 在运行时动态打开文件
files = []
for i in range(3):
fname = f'dynamic_{i}.txt'
f = stack.enter_context(open(fname, 'w'))
files.append(f)
f.write(f'这是第 {i} 个动态文件\n')
# 退出时,3个文件会按 LIFO 顺序自动关闭
2. enter_contexts —— 批量添加上下文管理器
Python 3.11 引入了 enter_contexts 方法,可以一次性压入多个上下文管理器。它接受一个可迭代对象,返回所有 __enter__ 返回值的元组。
# Python 3.11+ 支持
with ExitStack() as stack:
cms = [
open('a.txt', 'w'),
open('b.txt', 'w'),
open('c.txt', 'w'),
]
files = stack.enter_contexts(cms)
# files 是 (file_a, file_b, file_c) 的元组
for f in files:
f.write('批量写入')
# 退出时全部自动关闭
3. callback —— 注册清理回调
callback(callable, *args, **kwargs) 注册一个在退出时被调用的清理函数。这对于那些不是上下文管理器但需要清理的资源非常有用。
from contextlib import ExitStack
def cleanup_resource(resource_id, reason='normal'):
print(f'清理资源 #{resource_id}, 原因: {reason}')
with ExitStack() as stack:
# 注册多个清理回调
stack.callback(cleanup_resource, 1)
stack.callback(cleanup_resource, 2, reason='early')
stack.callback(lambda: print('匿名清理函数'))
print('工作中...')
# 退出时按注册顺序的逆序执行:
# 匿名清理函数
# 清理资源 #2, 原因: early
# 清理资源 #1, 原因: normal
# callback 也可以包装成上下文管理器:
stack.callback(cleanup_resource, 3)
# 等价于:
@contextmanager
def cleanup_cm(resource_id):
try:
yield
finally:
cleanup_resource(resource_id)
4. pop_all —— 转移上下文所有权
pop_all() 将当前 ExitStack 中所有已注册的上下文管理器转移到新的 ExitStack 中。这在需要将上下文的生命周期延长的场景下非常有用(例如,函数内部管理资源,但将清理责任返还给调用者)。
from contextlib import ExitStack
def setup_resources():
"""准备资源并转移调用者"""
stack = ExitStack()
try:
f1 = stack.enter_context(open('res1.txt', 'w'))
f2 = stack.enter_context(open('res2.txt', 'w'))
# 将栈中所有上下文转移给调用者
return stack.pop_all()
except:
stack.close() # 如果失败,清理已打开的资源
raise
# 调用者负责清理
outer_stack = ExitStack()
with outer_stack:
inner_stack = setup_resources()
outer_stack.enter_context(inner_stack)
# 使用资源...
print('使用 setup_resources 提供的资源')
# outer_stack 退出时清理所有资源
# pop_all 的典型模式:
# 1. 内部函数创建 ExitStack 并 enter_context
# 2. pop_all 将责任转移出去
# 3. 调用者接管并最终清理
5. 多上下文动态管理完整示例
ExitStack 在实际开发中最强大的应用场景是需要根据运行时条件动态决定管理哪些资源的场景。以下是一个完整的示例:
from contextlib import ExitStack, contextmanager
import os
@contextmanager
def temp_dir(prefix='tmp_'):
"""创建一个临时目录,用完后删除"""
import tempfile, shutil
d = tempfile.mkdtemp(prefix=prefix)
try:
yield d
finally:
shutil.rmtree(d)
def process_files(file_list, need_temp=False, need_log=False):
"""根据运行时配置动态管理资源"""
with ExitStack() as stack:
# 条件性打开临时目录
if need_temp:
tmp = stack.enter_context(temp_dir('process_'))
print(f'临时目录: {tmp}')
# 条件性创建日志文件
if need_log:
log = stack.enter_context(open('process.log', 'w'))
log.write('开始处理\n')
# 批量打开输入文件
files = []
for fname in file_list:
if os.path.exists(fname):
f = stack.enter_context(open(fname, 'r'))
files.append(f)
else:
print(f'跳过不存在的文件: {fname}')
# 注册最终回调
stack.callback(lambda: print('所有资源清理完成'))
# 处理文件
for f in files:
print(f'处理: {f.name}')
# 灵活调用
process_files(['a.txt', 'b.txt'], need_temp=True, need_log=True)
# 所有资源(临时目录、日志文件、输入文件)在退出时自动清理
六、AsyncContextManager
contextlib 也支持异步上下文管理器。异步上下文管理器(Async Context Manager)实现了 __aenter__ 和 __aexit__ 协议,与 async with 语句配合使用。Python 3.7+ 提供了 @asynccontextmanager 装饰器,用于简洁地创建异步上下文管理器。
1. @asynccontextmanager 装饰器
用法与 @contextmanager 完全类似,区别在于被装饰的函数必须是异步生成器函数(包含 yield 的 async def 函数)。yield 之前的代码在 __aenter__ 中执行,yield 之后的代码在 __aexit__ 中执行。
from contextlib import asynccontextmanager
import asyncio
@asynccontextmanager
async def async_resource(name):
"""异步资源管理器"""
print(f' [异步进入] 连接 {name}')
resource = {'name': name, 'connected': True}
try:
yield resource
finally:
print(f' [异步退出] 断开 {name}')
resource['connected'] = False
async def main():
print('=== 异步上下文管理器演示 ===')
async with async_resource('数据库连接池') as res:
print(f' 使用资源: {res}')
await asyncio.sleep(0.5)
print('=== 演示结束 ===')
asyncio.run(main())
2. 异步上下文管理器的协议
异步上下文管理器的协议与同步版本一一对应:__aenter__ 对应 __enter__,__aexit__ 对应 __exit__,但都是协程方法,必须使用 await 调用。这使得在进入和退出上下文时可以进行异步操作。
import asyncio
class AsyncDatabase:
"""基于类的异步上下文管理器"""
async def __aenter__(self):
print('连接数据库...')
await asyncio.sleep(0.2) # 模拟异步连接
self.conn = {'connected': True}
return self.conn
async def __aexit__(self, exc_type, exc_val, exc_tb):
print('断开数据库...')
await asyncio.sleep(0.1) # 模拟异步断开
self.conn['connected'] = False
return False # 异常向外传播
async def query():
async with AsyncDatabase() as db:
print(f'数据库状态: {db}')
return '查询结果'
result = asyncio.run(query())
print(f'结果: {result}')
3. 实际应用场景
异步上下文管理器的典型场景包括:异步数据库连接池、aiohttp 客户端会话、异步文件 I/O 操作、异步锁和信号量等。在这些场景中,资源的获取和释放本身需要异步操作(如网络等待),因此传统的同步上下文管理器无法胜任。
# aiohttp 示例(伪代码,依赖 aiohttp 库)
# @asynccontextmanager
# async def http_session(base_url):
# session = aiohttp.ClientSession(base_url=base_url)
# try:
# yield session
# finally:
# await session.close()
#
# async def fetch_data():
# async with http_session('https://api.example.com') as session:
# async with session.get('/data') as resp:
# return await resp.json()
# 结合 asyncio 锁
@asynccontextmanager
async def async_lock(lock):
"""确保异步锁的获取和释放成对出现"""
await lock.acquire()
try:
yield
finally:
lock.release()
lock = asyncio.Lock()
async with async_lock(lock):
print('在锁保护的临界区内')
版本提示: @asynccontextmanager 在 Python 3.7 中正式加入标准库。如果你需要兼容 Python 3.6 及更早版本,可以考虑使用 async_generator 第三方库的 @asynccontextmanager 替代。
七、实战案例与总结
本节通过几个典型的实战案例,展示 contextlib 在实际项目中的高级应用模式,帮助你将前面学到的知识融会贯通。
1. 计时器上下文 —— 精确测量代码块执行时间
利用 @contextmanager 可以轻松实现一个可重入的计时器,用于测量代码块的执行时间,对性能分析和优化非常有帮助。
import time
from contextlib import contextmanager
@contextmanager
def timer(name='Task', report_func=print):
"""精确测量代码块执行时间
Args:
name: 任务名称,用于标识
report_func: 报告函数,默认为 print
"""
start = time.perf_counter()
try:
yield
finally:
elapsed = time.perf_counter() - start
report_func(f'[{name}] 耗时: {elapsed:.4f} 秒')
# 使用示例
import random
with timer('数据排序'):
data = [random.random() for _ in range(100000)]
data.sort()
# 嵌套计时
with timer('整体处理'):
with timer('数据加载'):
time.sleep(0.2)
with timer('数据计算', report_func=lambda msg: None): # 静默模式
time.sleep(0.3)
print('处理完成')
2. 数据库事务上下文 —— 自动提交/回滚
数据库操作中,事务管理是典型的需要上下文管理器的场景:正常执行时提交事务,发生异常时回滚事务。使用 @contextmanager 可以干净地封装这一模式。
from contextlib import contextmanager
@contextmanager
def transaction(connection, name='default'):
"""数据库事务管理器
正常退出 → 提交事务
异常退出 → 回滚事务
"""
print(f' [事务 {name}] 开始')
try:
yield connection
connection.commit()
print(f' [事务 {name}] 提交成功')
except Exception as e:
connection.rollback()
print(f' [事务 {name}] 回滚: {e}')
raise # 异常继续传播
# 模拟数据库连接
class FakeConnection:
def __init__(self):
self.closed = False
def execute(self, sql):
if 'ERROR' in sql:
raise RuntimeError('SQL 执行失败')
print(f' 执行: {sql}')
def commit(self):
print(f' 提交事务')
def rollback(self):
print(f' 回滚事务')
# 正常情况: 提交
print('--- 正常事务 ---')
conn = FakeConnection()
with transaction(conn, 'INSERT') as c:
c.execute('INSERT INTO users VALUES (1, "Alice")')
c.execute('INSERT INTO users VALUES (2, "Bob")')
# 异常情况: 回滚
print()
print('--- 异常事务 ---')
conn2 = FakeConnection()
try:
with transaction(conn2, 'UPDATE') as c:
c.execute('UPDATE users SET name="Charlie" WHERE id=1')
c.execute('UPDATE users SET ERROR') # 触发异常
except RuntimeError:
print(' 外层捕获到异常')
3. 资源池管理 —— ExitStack 的实战应用
ExitStack 在管理动态数量的资源方面具有独特优势。以下是一个实现简单数据库连接池管理的示例:
from contextlib import ExitStack, contextmanager
class ConnectionPool:
"""简单的连接池管理器"""
def __init__(self, min_size=2, max_size=5):
self.min_size = min_size
self.max_size = max_size
self._pool = []
self._init_pool()
def _init_pool(self):
for _ in range(self.min_size):
self._pool.append(self._create_conn())
def _create_conn(self):
return FakeConnection()
@contextmanager
def get_connection(self):
"""从池中获取连接"""
conn = self._pool.pop() if self._pool else self._create_conn()
try:
yield conn
finally:
if len(self._pool) < self.max_size:
self._pool.append(conn)
else:
conn.closed = True
print('连接池已满,关闭多余连接')
@contextmanager
def transaction_scope(self):
"""从池中获取连接并开启事务"""
with self.get_connection() as conn:
with transaction(conn):
yield conn
# 使用资源池
pool = ConnectionPool(min_size=2)
with ExitStack() as stack:
# 同时获取多个连接
conn1 = stack.enter_context(pool.get_connection())
conn2 = stack.enter_context(pool.get_connection())
conn1.execute('SELECT 1')
conn2.execute('SELECT 2')
# 退出时连接自动回池
# 使用事务范围
with pool.transaction_scope() as conn:
conn.execute('INSERT INTO log VALUES ("操作完成")')
4. 综合对比:contextlib 工具适用场景速查表
工具 适用场景 替代方案 推荐度
__enter__/__exit__ 类 需要手动管理状态的复杂上下文 @contextmanager 一般
@contextmanager 大多数上下文管理场景 类的协议实现 强烈推荐
closing 遗留API、close() 方法 手动 try/finally 推荐
nullcontext 条件性上下文占位 if/else 分支 推荐
suppress 忽略已知异常 try/except/pass 推荐
redirect_stdout 临时捕获/重定向输出 logging模块 测试场景推荐
redirect_stderr 捕获错误输出 logging模块 测试场景推荐
ExitStack 动态数量/条件性资源管理 嵌套 with、手动管理 强烈推荐
@asynccontextmanager 异步资源管理 __aenter__/__aexit__ 类 强烈推荐
5. 核心要点总结
1. 上下文管理器协议(__enter__/__exit__) 是 Python 资源管理的基石,with 语句确保资源在离开作用域时被正确清理。
2. @contextmanager 装饰器 提供了一种声明式的创建方式,yield 分割进入和退出逻辑,try/finally 确保清理代码必定执行。
3. 实用工具(closing, nullcontext, suppress) 针对常见场景提供开箱即用的解决方案,让代码更简洁、意图更明确。
4. redirect_stdout/redirect_stderr 在测试和日志捕获场景中非常实用,但注意非线程安全。
5. ExitStack 是最强大的上下文管理工具,支持动态添加、批量管理、回调注册、责任转移等高级模式。
6. @asynccontextmanager 是异步编程中管理资源的标准方式,与 async with 语句配合使用。
7. 选择原则: 简单场景用 @contextmanager,动态场景用 ExitStack,异步场景用 @asynccontextmanager,特殊需求用类协议实现。
6. 进一步思考
1. 上下文管理器与装饰器结合使用会碰撞出什么火花?例如实现一个自动重试的上下文管理器。
2. ExitStack 的 pop_all 在库的设计中如何用于"在初始化时分配资源,在用户关闭时释放"的模式?
3. 在大型项目中,如何组织自定义上下文管理器的目录结构?推荐的做法是将它们集中放在一个 contexts.py 或 managers.py 模块中。
4. 考虑性能开销:@contextmanager 比自定义类上下文管理器稍慢(因为涉及生成器的额外开销),在性能敏感型场景中需要注意。