上下文管理器

程序运行中最常见的 bug 根源之一就是资源泄露——打开的文件忘了关、获取的锁忘了释放、数据库连接忘了归还。这些错误在短脚本中或许只是让进程多占几秒资源，但在长期运行的服务中，它们会像慢性的内存泄漏一样，最终耗尽文件描述符、连接池或者锁，导致服务崩溃。

Python 的上下文管理器（Context Manager）正是为解决这个问题而设计的。它提供了一种声明式的语法，确保资源的获取和释放成对发生——无论代码块是正常结束、抛出异常、还是被 return 提前退出。

为什么需要上下文管理器

资源管理的基本挑战

考虑一个最朴素的数据库操作：

python

# 需要 Python 3.8+
import sqlite3

conn = sqlite3.connect("app.db")
cursor = conn.cursor()
cursor.execute("SELECT * FROM users")
data = cursor.fetchall()
conn.close()

这段代码看似正常，但实际上存在三个问题：

如果 execute() 抛出异常，conn.close() 永远不会执行，连接泄漏
如果中间加了 return，连接同样不会关闭
代码重复：每个需要资源的地方都要写 try...finally

传统补救方案是用 try...finally：

python

conn = sqlite3.connect("app.db")
try:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    data = cursor.fetchall()
finally:
    conn.close()

这解决了资源泄漏问题，但引入了新的缺陷：代码冗长，且当多个资源需要同时管理时，嵌套的 try...finally 结构会迅速失控。

RAII 在 Python 中的实现

C++ 程序员熟悉 RAII（Resource Acquisition Is Initialization）模式：资源在构造函数中获取，在析构函数中释放，利用对象的生命周期管理资源。但 Python 没有确定的析构时机——__del__ 方法只有在对象被垃圾回收时才调用，而 GC 的时机是不确定的。

Python 的答案是上下文管理器协议：通过 with 语句，将资源获取与释放绑定到一个显式的代码块。with 语句块结束时，无论如何退出，释放代码都一定会执行。

python

# with 语句——Python 的 RAII 实现
with sqlite3.connect("app.db") as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    # 无论是否发生异常，conn 都会被关闭

EAFP 与资源管理

上下文管理器是 Python EAFP（Easier to Ask Forgiveness than Permission）哲学的完美体现：不提前检查资源状态，而是直接使用，将清理逻辑交给协议保证。

基于类的上下文管理器

协议定义

上下文管理器协议由两个方法组成：

方法	签名	用途
`__enter__`	`__enter__(self) -> Any`	进入 `with` 块时调用，返回值绑定到 `as` 后的变量
`__exit__`	`__exit__(self, exc_type, exc_val, exc_tb) -> bool`	退出 `with` 块时调用，接收异常信息，返回 `True` 可抑制异常

其中 __exit__ 的三个参数在没有异常发生时都是 None，有异常时则分别携带异常类型、异常值、回溯对象。

执行流程时序图

图表渲染中…

第一个上下文管理器：文件操作

python

# 需要 Python 3.10+
class ManagedFile:
    """一个展示上下文管理器协议的文件包装器"""

    def __init__(self, filepath: str, mode: str = "r", encoding: str = "utf-8"):
        self.filepath = filepath
        self.mode = mode
        self.encoding = encoding
        self.file = None

    def __enter__(self):
        print(f"[ManagedFile] 打开文件: {self.filepath}")
        self.file = open(self.filepath, self.mode, encoding=self.encoding)
        return self.file

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.file:
            print(f"[ManagedFile] 关闭文件: {self.filepath}")
            self.file.close()
        # 返回 False 表示不抑制异常，让异常继续传播
        return False


# 使用
with ManagedFile("test.txt", "w") as f:
    f.write("Hello, Context Manager!\n")

实战：数据库连接管理器

python

# 需要 Python 3.10+
import sqlite3
from typing import Optional


class DatabaseConnection:
    """线程不安全的 SQLite 连接管理器"""

    def __init__(self, db_path: str):
        self.db_path = db_path
        self.connection: Optional[sqlite3.Connection] = None
        self.cursor: Optional[sqlite3.Cursor] = None

    def __enter__(self):
        self.connection = sqlite3.connect(self.db_path)
        self.connection.row_factory = sqlite3.Row  # 支持列名访问
        self.cursor = self.connection.cursor()
        return self.cursor  # 返回 cursor 给 as 变量

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            # 正常退出：提交事务
            self.connection.commit()
        else:
            # 异常退出：回滚事务
            self.connection.rollback()
            print(f"[DatabaseConnection] 事务回滚，原因: {exc_type.__name__}: {exc_val}")

        if self.cursor:
            self.cursor.close()
        if self.connection:
            self.connection.close()
        return False  # 不抑制异常


# 使用
with DatabaseConnection("app.db") as cursor:
    cursor.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT)")
    cursor.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
    # 事务自动提交

基于生成器的上下文管理器

每次都写一个类来实现 __enter__ 和 __exit__ 过于繁琐。contextlib.contextmanager 装饰器允许你用一个生成器函数来定义上下文管理器，大幅简化代码。

工作原理

@contextmanager 将生成器函数转换为上下文管理器：

yield 之前的代码等价于 __enter__
yield 的值作为 as 的目标
yield 之后的代码等价于 __exit__
只有 yield 一个值（不能像迭代器那样多次 yield）

图表渲染中…

实战：计时器上下文管理器

python

# 需要 Python 3.10+
import time
from contextlib import contextmanager
from typing import Generator


@contextmanager
def timer(name: str = "代码块") -> Generator[None, None, None]:
    """测量代码块执行时间的上下文管理器"""
    start = time.perf_counter()
    try:
        yield  # 将控制权交给 with 代码块
    finally:
        elapsed = time.perf_counter() - start
        print(f"[{name}] 耗时: {elapsed:.4f} 秒")


# 使用
with timer("数据处理"):
    total = sum(range(10_000_000))
    print(f"计算结果: {total}")

实战：临时目录管理器

python

# 需要 Python 3.10+
import shutil
import tempfile
from pathlib import Path
from contextlib import contextmanager
from typing import Generator


@contextmanager
def temporary_directory() -> Generator[Path, None, None]:
    """创建临时目录，代码块结束后自动删除"""
    tmp_dir = Path(tempfile.mkdtemp())
    try:
        print(f"[TempDir] 创建临时目录: {tmp_dir}")
        yield tmp_dir
    finally:
        print(f"[TempDir] 清理临时目录: {tmp_dir}")
        shutil.rmtree(tmp_dir, ignore_errors=True)


# 使用
with temporary_directory() as tmp:
    data_file = tmp / "data.txt"
    data_file.write_text("临时数据", encoding="utf-8")
    print(f"文件大小: {data_file.stat().st_size} 字节")

contextlib 模块核心工具

contextlib 模块提供了比 @contextmanager 更丰富的工具集，覆盖了常见的上下文管理需求。

closing：将任意对象变为上下文管理器

有些对象有 close() 方法但没有实现上下文管理器协议（如 urllib 的某些对象）。closing 为它们包装出 with 支持。

python

# 需要 Python 3.10+
from contextlib import closing
from urllib.request import urlopen

# urlopen 返回的 HTTPResponse 有 close()，但不支持 with
with closing(urlopen("https://httpbin.org/get")) as response:
    data = response.read()
    print(f"响应状态: {response.status}, 数据长度: {len(data)}")
# response.close() 被自动调用

suppress：优雅地忽略特定异常

当某些异常在你的场景中是可接受的（比如删除一个可能不存在的文件），suppress 比 try...except...pass 更简洁。

python

# 需要 Python 3.10+
from contextlib import suppress
from pathlib import Path

# 删除临时文件，不存在也不报错
with suppress(FileNotFoundError):
    Path("/tmp/stale_lock.pid").unlink()

# 可以同时忽略多种异常
with suppress(FileNotFoundError, PermissionError):
    Path("/tmp/old_cache").unlink()

suppress vs 裸 except

suppress 只抑制指定类型的异常，不会像 except: pass 那样吞掉 KeyboardInterrupt 等不应被忽略的异常。始终优于裸 except。

redirect_stdout：捕获标准输出

将 print() 或其他写入 sys.stdout 的输出重定向到文件或 io.StringIO。

python

# 需要 Python 3.10+
import io
from contextlib import redirect_stdout

buffer = io.StringIO()

with redirect_stdout(buffer):
    print("这行进入了缓冲区")
    print("而不是控制台")

output = buffer.getvalue()
print(f"捕获到的输出:\n{output}")

ExitStack：动态管理多个上下文管理器

ExitStack 是 contextlib 中最强大的工具。当上下文管理器的数量在运行时才能确定，或者需要条件性地进入上下文时，ExitStack 提供了统一的清理回调机制。

python

# 需要 Python 3.10+
from contextlib import ExitStack
from pathlib import Path


def process_files(file_list: list[Path]) -> None:
    """同时打开多个文件，统一管理"""
    with ExitStack() as stack:
        files = []
        for filepath in file_list:
            try:
                f = stack.enter_context(open(filepath, "r", encoding="utf-8"))
                files.append(f)
            except FileNotFoundError:
                print(f"警告: 文件 {filepath} 不存在，跳过")

        # 所有成功打开的文件都在这里可用
        for f in files:
            first_line = f.readline().strip()
            print(f"{f.name}: {first_line}")
    # 所有文件自动关闭，以注册的相反顺序

ExitStack 还支持注册任意回调函数：

python

# 需要 Python 3.10+
from contextlib import ExitStack


def acquire_lock(name: str) -> str:
    print(f"获取锁: {name}")
    return f"lock_{name}"


def release_lock(lock_id: str):
    print(f"释放锁: {lock_id}")


with ExitStack() as stack:
    # 注册回调：退出时按注册的相反顺序调用
    lock_a = acquire_lock("A")
    stack.callback(release_lock, lock_a)

    lock_b = acquire_lock("B")
    stack.callback(release_lock, lock_b)

    print("持有两个锁，执行关键操作...")
# 退出时：先释放 lock_b，再释放 lock_a

contextlib 工具速查表

工具	用途	典型场景
`contextmanager`	将生成器函数转为上下文管理器	快速创建自定义上下文管理器
`closing(thing)`	为有 `close()` 的对象提供 `with` 支持	urllib 响应、旧版库对象
`suppress(*excs)`	忽略指定异常	删除可能不存在的文件
`redirect_stdout(target)`	重定向标准输出	捕获 `print` 输出、测试
`redirect_stderr(target)`	重定向标准错误	捕获警告和错误日志
`ExitStack`	动态管理多个上下文管理器	可选资源、运行时确定的资源数量
`nullcontext(value)`	返回一个不执行任何操作的上下文管理器	条件性地使用或不使用上下文管理器
`AbstractContextManager`	上下文管理器的抽象基类	类型提示中的 `isinstance` 检查
`ContextDecorator`	让上下文管理器同时可用作装饰器	将上下文管理器逻辑应用到函数

nullcontext：条件性使用上下文管理器

python

# 需要 Python 3.10+
from contextlib import nullcontext


def process_data(data: list, use_lock: bool = False):
    """根据条件决定是否加锁"""
    # 如果 use_lock 为 True，使用真正的锁；否则使用空上下文
    from threading import Lock
    lock = Lock()
    ctx = lock if use_lock else nullcontext()

    with ctx:
        # 只有在 use_lock=True 时才持有锁
        data.append(42)
        print(f"处理完成，数据长度: {len(data)}")

异步上下文管理器

在异步编程中，资源的获取和释放可能涉及 I/O 操作（如建立数据库连接、打开网络流），这些操作本身需要被 await。Python 提供了异步上下文管理器协议来解决这个问题。

协议定义

异步上下文管理器使用 __aenter__ 和 __aexit__ 方法，两者都是异步方法，需要用 async with 进入。

python

# 需要 Python 3.10+
import asyncio
from typing import Optional


class AsyncConnection:
    """模拟异步数据库连接"""

    def __init__(self, dsn: str):
        self.dsn = dsn
        self._connected = False

    async def __aenter__(self):
        # 模拟异步连接建立
        await asyncio.sleep(0.1)
        self._connected = True
        print(f"[AsyncConnection] 已连接到 {self.dsn}")
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            print(f"[AsyncConnection] 异常退出: {exc_type.__name__}")
        await asyncio.sleep(0.05)  # 模拟异步断开
        self._connected = False
        print(f"[AsyncConnection] 已断开 {self.dsn}")
        return False

    async def query(self, sql: str) -> str:
        if not self._connected:
            raise RuntimeError("未连接")
        await asyncio.sleep(0.05)
        return f"查询结果: {sql}"


async def main():
    async with AsyncConnection("postgresql://localhost/test") as conn:
        result = await conn.query("SELECT * FROM users")
        print(result)


asyncio.run(main())

异步上下文管理器的生成器写法

contextlib 提供了 asynccontextmanager 装饰器，语法与 @contextmanager 类似，但使用 async with 进入。

python

# 需要 Python 3.10+
import asyncio
from contextlib import asynccontextmanager
from typing import AsyncGenerator


@asynccontextmanager
async def async_timer(name: str = "异步代码块") -> AsyncGenerator[None, None]:
    """异步计时器上下文管理器"""
    start = asyncio.get_event_loop().time()
    try:
        yield
    finally:
        elapsed = asyncio.get_event_loop().time() - start
        print(f"[{name}] 耗时: {elapsed:.4f} 秒")


async def async_workload():
    async with async_timer("网络请求"):
        await asyncio.sleep(0.5)
        print("请求完成")


asyncio.run(async_workload())

异步上下文管理器实际应用

aiohttp.ClientSession、asyncpg.Connection、aioredis.Redis 等异步库都实现了异步上下文管理器协议。在异步代码中，始终使用 async with 管理这些资源。