contextlib 完全指南

官方文档：https://docs.python.org/3/library/contextlib.html
适用版本：Python 3.12（2026-05-08 核实）

contextlib 是 Python 标准库中用于支持上下文管理器的工具模块。它提供了一系列工具，让你无需手动实现 __enter__ / __exit__ 协议就能创建上下文管理器，同时也提供了管理多个上下文、异常抑制、输出重定向等实用工具。

基础概念：上下文管理协议

上下文管理器是实现了 __enter__ 和 __exit__ 两个方法的对象，配合 with 语句使用：

with expression as variable:
    body

等价于：

manager = expression
variable = manager.__enter__()
try:
    body
except:
    if not manager.__exit__(*sys.exc_info()):
        raise
else:
    manager.__exit__(None, None, None)

手动实现 __enter__ / __exit__ 协议

通过类实现上下文管理器是最基础的方式，适合需要维护状态的场景。

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

    def __enter__(self):
        # 获取资源，返回值绑定到 as 子句的变量
        print(f"获取资源: {self.name}")
        self.resource = f"resource:{self.name}"
        return self.resource

    def __exit__(self, exc_type, exc_val, exc_tb):
        # 释放资源
        print(f"释放资源: {self.name}")
        self.resource = None
        # 返回 True 表示异常已处理（吞掉异常）
        # 返回 False 或 None 表示异常继续传播
        return False


with ManagedResource("database") as res:
    print(f"使用资源: {res}")

__exit__ 方法的参数说明：

@contextmanager 装饰器

contextlib.contextmanager 允许用生成器函数来编写上下文管理器，代码更简洁直观。

from contextlib import contextmanager

@contextmanager
def managed_resource(name: str):
    # __enter__ 部分：yield 之前的代码
    print(f"获取资源: {name}")
    resource = f"resource:{name}"
    try:
        yield resource  # yield 的值绑定到 as 子句
        # __exit__ 正常退出部分：yield 之后的代码
    except Exception as e:
        # __exit__ 异常处理部分
        print(f"处理异常: {e}")
        raise  # 重新抛出，不吞掉异常
    finally:
        # 无论是否异常都会执行的清理
        print(f"释放资源: {name}")


with managed_resource("database") as res:
    print(f"使用资源: {res}")

@contextmanager 装饰器本身不接受参数，但被装饰的函数可以接受任意参数。

yield 前后的执行流程

with block 进入
  ↓
执行 yield 前的代码（相当于 __enter__）
  ↓
yield 值（绑定到 as 变量）
  ↓
执行 with block 体
  ↓
如果 with block 正常结束 → 执行 yield 后的代码
如果 with block 抛出异常 → 异常在 yield 处重新抛出，进入 except/finally

异常处理示例

from contextlib import contextmanager

@contextmanager
def transaction(conn):
    try:
        yield conn
        conn.commit()       # 正常完成则提交
    except Exception:
        conn.rollback()     # 异常则回滚
        raise               # 继续传播异常
    finally:
        conn.close()        # 始终关闭连接

contextlib.closing

closing 将任何拥有 close() 方法的对象包装为上下文管理器，适用于那些没有实现上下文管理协议但有 close() 方法的对象。

from contextlib import closing
import urllib.request

# urllib 返回的对象有 close() 但不是上下文管理器
with closing(urllib.request.urlopen("https://example.com")) as page:
    content = page.read()
# 离开 with 块后自动调用 page.close()

closing 类的参数：

contextlib.suppress

suppress 用于在 with 块中抑制指定类型的异常，让代码更简洁地忽略预期异常。

from contextlib import suppress
import os

# 传统写法
try:
    os.remove("nonexistent.txt")
except FileNotFoundError:
    pass

# 使用 suppress
with suppress(FileNotFoundError):
    os.remove("nonexistent.txt")

# 可以抑制多种异常
with suppress(FileNotFoundError, PermissionError):
    os.remove("protected_file.txt")

suppress 的参数：

注意：suppress 只抑制指定的异常类型，其他异常仍会正常传播。

contextlib.redirect_stdout

redirect_stdout 将 sys.stdout 临时重定向到另一个文件类对象，用于捕获标准输出。

from contextlib import redirect_stdout
import io

# 捕获函数的打印输出
output = io.StringIO()
with redirect_stdout(output):
    print("这段输出会被捕获")
    help(str.upper)   # help() 的输出也会被捕获

captured = output.getvalue()
print(f"捕获到的内容长度: {len(captured)}")

# 重定向到文件
with open("output.log", "w", encoding="utf-8") as f:
    with redirect_stdout(f):
        print("这段输出写入文件")

redirect_stdout 的参数：

contextlib.redirect_stderr

redirect_stderr 与 redirect_stdout 用法相同，但重定向的是 sys.stderr。

from contextlib import redirect_stderr
import io

error_output = io.StringIO()
with redirect_stderr(error_output):
    import warnings
    warnings.warn("这是一个警告")  # 警告输出到 stderr

errors = error_output.getvalue()

redirect_stderr 的参数：

contextlib.ExitStack

ExitStack 是最强大的 contextlib 工具，允许动态管理可变数量的上下文管理器。适用于需要在运行时决定进入哪些上下文的场景。

from contextlib import ExitStack

# 动态打开多个文件
filenames = ["a.txt", "b.txt", "c.txt"]
with ExitStack() as stack:
    files = [stack.enter_context(open(f, "w", encoding="utf-8")) for f in filenames]
    for i, f in enumerate(files):
        f.write(f"content {i}")
# 离开 with 块时，所有文件按逆序关闭

ExitStack 的主要方法

enter_context(cm)

callback(func, *args, **kwargs)

注册一个在退出时调用的回调函数，无需上下文管理器对象：

push(exit_func)

close()

立即展开栈并执行所有已注册的清理动作，等同于 __exit__(None, None, None)。

pop_all()

将当前栈的所有清理动作转移到一个新的 ExitStack，用于延迟清理所有权转移。

from contextlib import ExitStack

# 条件性地进入上下文
def process_files(filenames, output_filename=None):
    with ExitStack() as stack:
        files = [stack.enter_context(open(f, encoding="utf-8")) for f in filenames]

        # 条件性地打开输出文件
        if output_filename:
            out = stack.enter_context(open(output_filename, "w", encoding="utf-8"))
        else:
            import sys
            out = sys.stdout

        # 注册回调
        stack.callback(print, "所有文件处理完毕")

        for f in files:
            out.write(f.read())

使用 ExitStack 实现可选的上下文管理

from contextlib import ExitStack, nullcontext

def process(data, debug=False):
    with ExitStack() as stack:
        if debug:
            log_file = stack.enter_context(open("debug.log", "w", encoding="utf-8"))
        else:
            log_file = None
        # ...处理逻辑

contextlib.nullcontext

nullcontext 是一个什么都不做的上下文管理器，用于简化"可选上下文"的代码分支（Python 3.7+）。

from contextlib import nullcontext

def process(data, lock=None):
    # 如果传入了锁就使用，否则用 nullcontext 占位
    with lock if lock is not None else nullcontext():
        # 处理数据
        pass

# 更简洁的写法（Python 3.10+）
def process_v2(data, lock=None):
    with lock or nullcontext():
        pass

nullcontext 的参数：

@asynccontextmanager 异步上下文管理器

asynccontextmanager 是 contextmanager 的异步版本，用于编写 async with 语句的上下文管理器。

from contextlib import asynccontextmanager
import asyncio

@asynccontextmanager
async def async_managed_resource(name: str):
    # 异步初始化
    print(f"异步获取资源: {name}")
    await asyncio.sleep(0.1)  # 模拟异步操作
    resource = f"async_resource:{name}"
    try:
        yield resource
    finally:
        # 异步清理
        print(f"异步释放资源: {name}")
        await asyncio.sleep(0.1)


async def main():
    async with async_managed_resource("database") as res:
        print(f"使用资源: {res}")
        await asyncio.sleep(0.5)

asyncio.run(main())

异步数据库连接池示例

from contextlib import asynccontextmanager

@asynccontextmanager
async def get_db_connection(pool):
    conn = await pool.acquire()
    try:
        yield conn
        await conn.commit()
    except Exception:
        await conn.rollback()
        raise
    finally:
        await pool.release(conn)


async def fetch_user(pool, user_id: int):
    async with get_db_connection(pool) as conn:
        return await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)

@asynccontextmanager 的要求与 @contextmanager 相同：被装饰的函数必须是 async def，且必须恰好 yield 一次。

与 with 语句结合的最佳实践

1. 嵌套 vs 多目标语法

# Python 3.1+ 支持多目标，避免过深嵌套
with open("input.txt", encoding="utf-8") as fin, open("output.txt", "w", encoding="utf-8") as fout:
    fout.write(fin.read())

# 而非
with open("input.txt", encoding="utf-8") as fin:
    with open("output.txt", "w", encoding="utf-8") as fout:
        fout.write(fin.read())

2. 确保资源总是被释放

from contextlib import contextmanager

@contextmanager
def acquire_lock(lock):
    lock.acquire()
    try:
        yield
    finally:
        lock.release()  # finally 保证即使有异常也会释放

3. 上下文管理器工厂函数

from contextlib import contextmanager
from typing import Generator

@contextmanager
def timer(label: str) -> Generator[None, None, None]:
    import time
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        print(f"{label}: {elapsed:.4f}s")


with timer("数据处理"):
    # 需要计时的代码
    result = sum(range(1_000_000))

4. 用 ExitStack 实现可撤销操作

from contextlib import ExitStack
import shutil
import os

def safe_replace(src: str, dst: str):
    backup = dst + ".bak"
    with ExitStack() as stack:
        # 注册撤销操作（先注册的后执行）
        if os.path.exists(dst):
            shutil.copy2(dst, backup)
            stack.callback(lambda: shutil.move(backup, dst) if os.path.exists(backup) else None)

        shutil.copy2(src, dst)
        # 成功后清除备份
        stack.pop_all()  # 取消所有已注册的撤销操作
        if os.path.exists(backup):
            os.remove(backup)

踩坑与注意事项

踩坑 1：yield 只能出现一次

@contextmanager 装饰的生成器函数中，yield 必须恰好出现一次。多次 yield 会导致 RuntimeError。

from contextlib import contextmanager

# 错误：多次 yield
@contextmanager
def bad_context():
    yield "first"
    yield "second"  # RuntimeError: generator didn't stop

# 正确：只 yield 一次
@contextmanager
def good_context():
    yield "only once"

踩坑 2：忘记处理异常导致资源泄露

from contextlib import contextmanager

# 错误：没有 try/finally，异常时资源不会释放
@contextmanager
def bad_resource():
    resource = acquire()
    yield resource
    release(resource)  # 若 yield 处发生异常，这行不会执行！

# 正确：用 try/finally 保证清理
@contextmanager
def good_resource():
    resource = acquire()
    try:
        yield resource
    finally:
        release(resource)  # 无论是否异常都会执行

踩坑 3：exit 返回值的陷阱

class SuppressAll:
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        return True  # 吞掉所有异常，包括 KeyboardInterrupt 和 SystemExit！


# 更安全的做法：只吞掉特定异常
class SuppressValueError:
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        return exc_type is ValueError  # 只吞掉 ValueError

踩坑 4：异常传播规则

在 @contextmanager 中，如果 yield 处发生的异常在生成器内部被捕获但没有重新抛出，效果等同于 __exit__ 返回 True（吞掉异常）：

from contextlib import contextmanager

@contextmanager
def swallows_exceptions():
    try:
        yield
    except Exception:
        pass  # 捕获但不重新抛出 → 相当于吞掉异常

with swallows_exceptions():
    raise ValueError("这个异常会被吞掉")

print("程序继续执行")  # 这行会执行

踩坑 5：不要在 __exit__ 中引发新异常

class BadContext:
    def __exit__(self, exc_type, exc_val, exc_tb):
        # 如果这里抛出新异常，原始异常会被丢弃
        raise RuntimeError("清理时出错")  # 原始异常丢失！


class GoodContext:
    def __exit__(self, exc_type, exc_val, exc_tb):
        try:
            self.cleanup()
        except Exception as cleanup_error:
            # Python 3.11+ 可以用 ExceptionGroup
            # 或者记录日志后继续
            import logging
            logging.exception("清理失败")
        return False  # 不吞掉原始异常

踩坑 6：suppress 不适用于需要感知异常的场景

from contextlib import suppress

result = None
with suppress(ValueError):
    result = int("not a number")
# result 仍然是 None，但代码不会报错
# 如果需要知道是否发生了异常，不应该用 suppress

最佳实践

@contextmanager 优先于手写类：除非需要多次进入或继承，否则用 @contextmanager 装饰器定义上下文管理器比实现 __enter__/__exit__ 类更简洁，且 yield 分隔入口和出口逻辑，一目了然。

contextlib.suppress 替代 try/except: pass：明确表达"有意忽略此异常"的语义，比空 except 块更可读，也更容易被代码审查工具识别为有意行为。

ExitStack 管理数量不定的上下文：动态文件列表、数量不确定的网络连接等，用 ExitStack 统一管理退出，避免嵌套 with 语句：

from contextlib import ExitStack

with ExitStack() as stack:
    files = [stack.enter_context(open(f)) for f in filenames]
    process(files)

contextlib.closing 包装只有 close() 的对象：对于没有实现上下文管理协议但有 close() 方法的对象（如某些数据库游标），用 closing() 确保资源释放。

AsyncExitStack 用于异步上下文的动态管理：与 ExitStack 类似但支持异步上下文管理器，是异步代码中管理多个 async with 资源的最佳方案。

常见陷阱

陷阱：@contextmanager 函数中 yield 必须恰好执行一次

现象： @contextmanager 函数中有条件分支，某些情况下不执行 yield 或执行多次，运行时报 RuntimeError: generator didn't yield。
原因： 上下文管理器协议要求 __enter__ 和 __exit__ 各调用一次，@contextmanager 严格检查 yield 次数。
解决： 确保函数无论哪条分支都恰好执行一次 yield，用 try/finally 保护清理逻辑。

陷阱：suppress 误抑制了预期外的异常

现象： with suppress(Exception): 抑制了所有异常，包括 SystemExit、KeyboardInterrupt 以外的 bug 异常，导致错误无声地被忽略。
原因： suppress 接受的异常类型太宽泛时，业务异常也被吞掉。
解决： 始终指定具体异常类型，如 suppress(FileNotFoundError)，不用 Exception 或更宽的基类。

陷阱：contextmanager 中异常处理遗漏 finally

现象： with 块内抛出异常时，yield 后的清理代码没有执行，资源泄漏。
原因： 若 yield 没有被包在 try/finally 中，异常会在 yield 处传播并终止生成器，跳过后续代码。
解决： 所有 @contextmanager 清理代码写在 finally 块中：

@contextmanager
def managed_resource():
    resource = acquire()
    try:
        yield resource
    finally:
        release(resource)  # 无论是否异常都执行

参见

装饰器与函数高级
asyncio异步编程完全指南