为什么需要上下文管理器

在处理资源（文件、网络连接、锁等）时，有一个经典隐患：操作中途发生异常，导致 close() 或释放逻辑被跳过，资源泄漏：

# 不使用上下文管理器 - 容易出问题
file = open('data.txt', 'w')
file.write('hello')
file.close()  # 如果 write 抛出异常，这里永远不会执行

更常见的写法是用 try...finally，但代码冗长：

file = open('data.txt', 'w')
try:
    file.write('hello')
finally:
    file.close()  # 无论如何都会执行

上下文管理器（Context Manager）正是为解决这一问题而生——用优雅的 with 语句，将获取和释放资源的逻辑绑定在一起。

基于类的上下文管理器

实现上下文管理器，只需定义一个类，实现两个特殊方法：

class FileManager:
    """文件管理器的上下文管理器实现"""
    def __init__(self, filename, mode):
        self.filename = filename
        self.mode = mode
        self.file = None

    def __enter__(self):
        """进入上下文时执行，返回值绑定到 as 后的变量"""
        self.file = open(self.filename, self.mode)
        return self.file  # 赋值给 `as` 后的变量

    def __exit__(self, exc_type, exc_val, exc_tb):
        """退出上下文时执行，无论是否发生异常"""
        if self.file:
            self.file.close()
        # 返回 True 可以压制异常（但不推荐）
        return False

# 使用
with FileManager('test.txt', 'w') as f:
    f.write('hello, context manager!')
# 退出 with 块时，__exit__ 自动被调用

__exit__ 的三个参数分别是异常类型、异常值、追踪对象。如果 with 块内没有异常，这三个值都是 None。

压制异常

如果 __exit__ 返回 True，异常会被压制，不会向外传播：

class SuppressDivisionErrors:
    def __enter__(self):
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is ZeroDivisionError:
            print(f"捕获除零错误: {exc_val}")
            return True  # 压制异常
        return False

with SuppressDivisionErrors():
    result = 10 / 0
    print("这里不会执行")
print("程序继续执行")  # 这行会输出

contextlib 模块

手写类实现上下文管理器稍显繁琐，Python 标准库提供了更便利的工具。

@contextmanager 装饰器

contextlib.contextmanager 装饰器可以将生成器函数转换为上下文管理器，yield 之前的代码相当于 __enter__，之后的代码相当于 __exit__：

import contextlib

@contextlib.contextmanager
def managed_resource(name):
    """资源管理的便捷写法"""
    print(f"[获取资源] {name}")
    resource = f"Resource<{name}>"
    yield resource  # yield 前的代码是 __enter__，yield 后是 __exit__
    print(f"[释放资源] {name}")

with managed_resource("数据库连接") as res:
    print(f"使用资源: {res}")
# 输出:
# [获取资源] 数据库连接
# 使用资源: Resource<数据库连接>
# [释放资源] 数据库连接

异常处理也很自然——yield 后的代码在 finally 块中执行：

@contextlib.contextmanager
def safe_file_writer(filepath):
    f = open(filepath, 'w')
    try:
        yield f
    finally:
        f.close()
        print("文件已关闭")

with safe_file_writer('out.txt') as f:
    f.write("test")
    # 即使这里抛异常，finally 也会执行

@contextmanager 配合 try/except

@contextlib.contextmanager
def transaction(conn):
    """模拟数据库事务"""
    print("开始事务...")
    conn.begin()
    try:
        yield conn
        conn.commit()
        print("事务提交")
    except Exception as e:
        conn.rollback()
        print(f"事务回滚: {e}")
        raise

class FakeConn:
    def begin(self): print("  [conn] begin")
    def commit(self): print("  [conn] commit")
    def rollback(self): print("  [conn] rollback")

# 正常提交
with transaction(FakeConn()) as conn:
    print("  执行 SQL...")

# 模拟失败
try:
    with transaction(FakeConn()) as conn:
        raise RuntimeError("SQL 执行失败")
except RuntimeError as e:
    print(f"异常已传播: {e}")

closing() - 资源无需关闭时的简化

有些资源虽然实现了 close() 方法，但实际上不需要真正关闭（或者根本没有资源），可以用 contextlib.closing()：

import contextlib

class QueryResult:
    """模拟数据库查询结果集"""
    def __init__(self, data):
        self.data = data
    def close(self):
        print("关闭结果集")

    def __enter__(self):
        return self
    def __exit__(self, *args):
        self.close()

# 或者更简单地使用 closing
with contextlib.closing(QueryResult([1, 2, 3])) as result:
    print(result.data)
# 输出: [1, 2, 3]
# 输出: 关闭结果集

redirect_stdout - 标准输出重定向

contextlib.redirect_stdout 可以临时重定向标准输出，在测试时非常有用：

import contextlib
from io import StringIO

with open('output.txt', 'w') as f:
    with contextlib.redirect_stdout(f):
        print("这段话不会输出到终端")
        print("而是写入文件")

# 验证
with open('output.txt', 'r') as f:
    print(f.read())
# 输出:
# 这段话不会输出到终端
# 而是写入文件

类似的还有 redirect_stderr 重定向标准错误。

suppress() - 精确压制异常

contextlib.suppress(*exceptions) 可以精确压制指定的异常：

import contextlib

# 忽略 FileNotFoundError
with contextlib.suppress(FileNotFoundError):
    os.remove('不存在.txt')

# 等价于
try:
    os.remove('不存在.txt')
except FileNotFoundError:
    pass

嵌套 with 语句

ExitStack 允许在单个 with 块中管理多个上下文：

import contextlib

with contextlib.ExitStack() as stack:
    # 自动管理多个资源
    file1 = stack.enter_context(open('a.txt', 'w'))
    file2 = stack.enter_context(open('b.txt', 'w'))
    file3 = stack.enter_context(open('c.txt', 'w'))

    file1.write("file1")
    file2.write("file2")
    file3.write("file3")
# 所有文件自动关闭，无论是否发生异常

单次使用上下文管理器

nullcontext 为不需要真正上下文管理的场景提供占位：

import contextlib

# 根据条件选择不同的上下文管理器
with contextlib.nullcontext() if no_db else db_connection as conn:
    conn.execute("SELECT 1")

小结

• with 语句自动管理资源获取与释放，避免遗漏
• 自定义类实现 __enter__ + __exit__ 可创建任意上下文管理器
• contextlib 提供了实用的装饰器和工具：
  • @contextmanager：将生成器函数转为上下文管理器
  • closing()：资源不需要真正关闭时使用
  • redirect_stdout/stderr：标准输出重定向
  • suppress()：精确压制特定异常
  • ExitStack：管理多个上下文管理器

[[返回 Python 首页|python/index]]