Python 嵌套上下文管理器

多个资源需要同时管理时，可使用嵌套上下文管理器，ExitStack 提供灵活的动态管理能力。

基本嵌套写法

# 传统写法
with open('file1.txt') as f1:
    with open('file2.txt') as f2:
        with open('file3.txt') as f3:
            content = f1.read() + f2.read() + f3.read()

# 单行写法
with open('file1.txt') as f1, open('file2.txt') as f2, open('file3.txt') as f3:
    content = f1.read() + f2.read() + f3.read()

ExitStack 基础

from contextlib import ExitStack

with ExitStack() as stack:
    f1 = stack.enter_context(open('file1.txt'))
    f2 = stack.enter_context(open('file2.txt'))
    f3 = stack.enter_context(open('file3.txt'))
    # 使用文件资源
    content = f1.read() + f2.read() + f3.read()
# 退出时按逆序关闭所有资源

动态数量资源

from contextlib import ExitStack

def process_files(filenames):
    with ExitStack() as stack:
        files = [stack.enter_context(open(f)) for f in filenames]
        return [f.read() for f in files]

# 处理任意数量的文件
contents = process_files(['a.txt', 'b.txt', 'c.txt'])

条件性资源管理

from contextlib import ExitStack

def process_data(input_file, output_file=None, log_file=None):
    with ExitStack() as stack:
        fin = stack.enter_context(open(input_file, 'r'))
        fout = stack.enter_context(open(output_file, 'w')) if output_file else None
        flog = stack.enter_context(open(log_file, 'a')) if log_file else None

        data = fin.read()
        if fout:
            fout.write(data.upper())
        if flog:
            flog.write(f"Processed {input_file}\n")

回调函数注册

from contextlib import ExitStack

def cleanup(name):
    print(f"清理 {name}")

with ExitStack() as stack:
    stack.callback(cleanup, "资源A")
    stack.callback(cleanup, "资源B")
    print("执行主要操作")
# 退出时按逆序执行回调

# 输出:
# 执行主要操作
# 清理 资源B
# 清理 资源A

获取已管理资源

from contextlib import ExitStack

with ExitStack() as stack:
    f = stack.enter_context(open('test.txt'))
    # 获取所有已注册的清理回调
    print(len(stack._exit_callbacks))  # 1

combine_contexts 函数

合并多个上下文管理器为一个。

from contextlib import contextmanager, ExitStack

@contextmanager
def cm1():
    print("进入 cm1")
    yield
    print("退出 cm1")

@contextmanager
def cm2():
    print("进入 cm2")
    yield
    print("退出 cm2")

def combine_contexts(*cms):
    with ExitStack() as stack:
        for cm in cms:
            stack.enter_context(cm())
        yield

with combine_contexts(cm1, cm2):
    print("主要操作")

# 输出:
# 进入 cm1
# 进入 cm2
# 主要操作
# 退出 cm2
# 退出 cm1

嵌套方式对比

要点总结

• 嵌套写法适合固定数量的资源管理
• ExitStack 支持动态数量的资源
• enter_context() 返回上下文管理器的返回值
• 退出时按 LIFO 顺序（后进先出）清理资源
• callback() 可注册任意清理函数
• ExitStack 是处理复杂资源场景的最佳选择