Python 上下文管理器完全指南：从 with 语句到 contextlib 深度解析 | 资源管理的优雅之道

1
file = open("data.txt", "r")
2
try:
3
    content = file.read()
4
    # 处理 content
5
    print(content)
6
finally:
7
    file.close()  # 确保文件被关闭

1
file1 = open("input.txt", "r")
2
try:
3
    file2 = open("output.txt", "w")
4
    try:
5
        # 处理文件
6
        file2.write(file1.read())
7
    finally:
8
        file2.close()
9
finally:
10
    file1.close()

1
with open("data.txt", "r") as file:
2
    content = file.read()
3
    print(content)
4
# 文件在此处自动关闭，无论是否发生异常

1
with open("input.txt", "r") as f1, open("output.txt", "w") as f2:
2
    f2.write(f1.read())
3
# 两个文件都自动关闭

1
class SimpleContext:
2
    def __enter__(self):
3
        print("→ 进入上下文")
4
        return self  # 返回自身供 as 使用
5

6
    def __exit__(self, exc_type, exc_val, exc_tb):
7
        print("← 退出上下文")
8
        # exc_type: 异常类型 (如发生异常)
9
        # exc_val: 异常对象
10
        # exc_tb: 异常追踪栈
11
        return False  # 返回 True 会抑制异常
12

13
with SimpleContext() as ctx:
14
    print("正在执行代码块")

1
→ 进入上下文
2
正在执行代码块
3
← 退出上下文

1
class ErrorHandlingContext:
2
    def __enter__(self):
3
        return self
4

5
    def __exit__(self, exc_type, exc_val, exc_tb):
6
        if exc_type is not None:
7
            print(f"捕获异常: {exc_type.__name__}: {exc_val}")
8
            # 返回 True 表示已处理异常，不会继续向上抛出
9
            # 返回 False (默认) 表示异常继续传播
10
        return False  # 让异常继续传播
11

12
with ErrorHandlingContext():
13
    raise ValueError("测试错误")

1
class SuppressContext:
2
    def __init__(self, *exception_types):
3
        self.exception_types = exception_types
4

5
    def __enter__(self):
6
        return self
7

8
    def __exit__(self, exc_type, exc_val, exc_tb):
9
        if exc_type is not None and issubclass(exc_type, self.exception_types):
10
            print(f"已抑制异常: {exc_type.__name__}")
11
            return True  # 抑制指定类型的异常
12
        return False  # 其他异常继续传播
13

14
with SuppressContext(ValueError, TypeError):
15
    raise ValueError("这个错误会被抑制")  # 被抑制
16
    raise RuntimeError("这个错误不会被抑制")  # 会抛出

1
import sqlite3
2

3
class DatabaseConnection:
4
    def __init__(self, db_path):
5
        self.db_path = db_path
6
        self.connection = None
7

8
    def __enter__(self):
9
        self.connection = sqlite3.connect(self.db_path)
10
        return self.connection
11

12
    def __exit__(self, exc_type, exc_val, exc_tb):
13
        if self.connection:
14
            if exc_type is None:
15
                self.connection.commit()  # 正常退出时提交
16
            else:
17
                self.connection.rollback()  # 异常时回滚
18
            self.connection.close()
19
        return False
20

21
# 使用：自动管理事务
22
with DatabaseConnection("app.db") as conn:
23
    cursor = conn.cursor()
24
    cursor.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
25
# 如果一切正常，自动提交；发生异常自动回滚

1
from contextlib import contextmanager
2

3
@contextmanager
4
def simple_context():
5
    # __enter__ 部分：yield 之前
6
    print("→ 进入上下文")
7

8
    # yield 分隔进入与退出逻辑
9
    # yield 后的值会作为 as 变量的值
10
    yield "上下文内部数据"
11

12
    # __exit__ 部分：yield 之后
13
    print("← 退出上下文")
14

15
# 使用
16
with simple_context() as value:
17
    print(f"获取到: {value}")

1
→ 进入上下文
2
获取到: 上下文内部数据
3
← 退出上下文

1
@contextmanager
2
def managed_file(path, mode="r"):
3
    file = open(path, mode, encoding="utf-8")
4
    try:
5
        yield file  # 将控制权交给 with 代码块
6
    finally:
7
        file.close()  # 无论是否发生异常都关闭
8
        print("文件已关闭")
9

10
# 使用
11
with managed_file("data.txt", "w") as f:
12
    f.write("测试内容")

1
from contextlib import closing
2

3
# closing() 会在退出时自动调用对象的 close() 方法
4
with closing(open("data.txt", "r")) as f:
5
    content = f.read()
6
# 等效于普通 with open(...)，但适用于无 __exit__ 的对象
7

8
# 实际场景：网络请求
9
from urllib.request import urlopen
10

11
with closing(urlopen("https://example.com")) as response:
12
    html = response.read()
13
# urlopen 返回的对象可能没有完整实现上下文管理器，用 closing 来包装

1
from contextlib import suppress
2
import os
3

4
# 删除文件，如果不存在也不报错
5
with suppress(FileNotFoundError):
6
    os.remove("temp_file.txt")
7

8
# 忽略多种异常
9
with suppress(FileNotFoundError, PermissionError):
10
    os.remove("temp_file.txt")
11
    os.rmdir("temp_dir")

1
from contextlib import nullcontext
2

3
# 根据条件决定是否使用文件
4
def process_data(source=None):
5
    # 如果提供了文件路径，打开文件；否则使用标准输出
6
    if source:
7
        ctx = open(source, "w")
8
    else:
9
        ctx = nullcontext(enter_result=None)  # 提供一个无操作的上下文
10

11
    with ctx as f:
12
        if f:
13
            f.write("处理结果")
14
        else:
15
            print("处理结果")
16

17
# 或者更简洁的写法
18
from contextlib import nullcontext
19

20
ctx = open("file.txt") if use_file else nullcontext()
21
with ctx as resource:
22
    pass

1
from contextlib import ExitStack
2

3
filenames = ["file1.txt", "file2.txt", "file3.txt"]
4

5
with ExitStack() as stack:
6
    # 动态打开所有文件
7
    files = [stack.enter_context(open(name, "r")) for name in filenames]
8

9
    # 所有文件现在都被管理
10
    # 退出 with 时，所有文件按打开顺序的逆序关闭
11
    for f in files:
12
        print(f.read())

1
from contextlib import ExitStack
2
import sqlite3
3

4
def process_databases(db_paths):
5
    with ExitStack() as stack:
6
        # 动态打开所有数据库连接
7
        connections = [
8
            stack.enter_context(sqlite3.connect(path))
9
            for path in db_paths
10
        ]
11

12
        # 对所有连接执行操作
13
        for conn in connections:
14
            cursor = conn.cursor()
15
            cursor.execute("SELECT COUNT(*) FROM users")
16
            print(cursor.fetchone()[0])
17
        # 退出时所有连接自动关闭和提交

1
from contextlib import ExitStack
2
import os, tempfile
3

4
with ExitStack() as stack:
5
    # 创建临时目录，并注册退出时删除的回调
6
    temp_dir = tempfile.mkdtemp()
7
    stack.callback(lambda: os.rmdir(temp_dir))
8
    print(f"临时目录: {temp_dir}")
9

10
    # 创建临时文件，注册删除回调
11
    temp_file = os.path.join(temp_dir, "data.txt")
12
    with open(temp_file, "w") as f:
13
        f.write("临时数据")
14
    stack.callback(lambda: os.remove(temp_file))
15

16
    # 使用这些临时资源
17
    # ...
18
# 退出时：先删除临时文件，再删除临时目录（逆序执行）

1
from contextlib import ExitStack
2

3
def acquire_resources():
4
    stack = ExitStack()
5
    try:
6
        # 获取一些资源
7
        f = stack.enter_context(open("data.txt"))
8
        conn = stack.enter_context(sqlite3.connect("app.db"))
9
    except:
10
        stack.close()  # 获取失败时释放已有资源
11
        raise
12

13
    # 获取成功，将资源所有权转移
14
    return stack.pop_all()  # 返回一个新的 ExitStack，包含所有资源
15

16
# 调用方负责管理
17
stack = acquire_resources()
18
with stack:
19
    # 使用资源
20
    pass
21
# 资源在此释放

1
# Python 3.10+ 支持的新语法
2
with (
3
    open("input.txt", "r") as f1,
4
    open("output.txt", "w") as f2,
5
    open("log.txt", "a") as f3,
6
):
7
    f2.write(f1.read())
8
    f3.write("处理完成\n")

1
from contextlib import contextmanager
2

3
@contextmanager
4
def nested_operations():
5
    with open("in.txt", "r") as f1, open("out.txt", "w") as f2:
6
        yield (f1, f2)  # 将多个资源作为元组 yield
7

8
with nested_operations() as (input_file, output_file):
9
    output_file.write(input_file.read())

1
from contextlib import ExitStack
2

3
def composite_context():
4
    stack = ExitStack()
5
    # 逐个进入上下文
6
    files = [
7
        stack.enter_context(open(f"file{i}.txt"))
8
        for i in range(5)
9
    ]
10
    return stack.pop_all()  # 返回组合后的资源管理
11

12
with composite_context() as managed_files:
13
    pass

1
import time
2
from contextlib import contextmanager
3

4
@contextmanager
5
def timer(label="任务"):
6
    start = time.perf_counter()
7
    try:
8
        yield
9
    finally:
10
        elapsed = time.perf_counter() - start
11
        print(f"[{label}] 耗时: {elapsed:.4f} 秒")
12

13
# 使用
14
with timer("数据处理"):
15
    total = sum(range(1_000_000))
16
# 输出: [数据处理] 耗时: 0.0345 秒
17

18
# 嵌套使用
19
with timer("整体任务"):
20
    with timer("子任务 A"):
21
        time.sleep(0.1)
22
    with timer("子任务 B"):
23
        time.sleep(0.2)

1
import os
2
from contextlib import contextmanager
3

4
@contextmanager
5
def change_directory(path):
6
    original_path = os.getcwd()
7
    os.chdir(path)
8
    try:
9
        yield
10
    finally:
11
        os.chdir(original_path)  # 恢复原目录
12

13
# 使用
14
print(f"当前: {os.getcwd()}")  # /home/user/project
15

16
with change_directory("/tmp"):
17
    print(f"临时: {os.getcwd()}")  # /tmp
18

19
print(f"恢复: {os.getcwd()}")  # /home/user/project

1
import threading
2
from contextlib import contextmanager
3

4
# threading.Lock 本身就是上下文管理器
5
lock = threading.Lock()
6

7
# 标准用法
8
with lock:
9
    # 临界区代码
10
    pass
11

12
# 自定义可重入锁管理
13
@contextmanager
14
def acquire_lock(lock_obj, timeout=None):
15
    acquired = lock_obj.acquire(timeout=timeout)
16
    if not acquired:
17
        raise TimeoutError("无法在超时时间内获取锁")
18
    try:
19
        yield
20
    finally:
21
        if acquired:
22
            lock_obj.release()

1
from contextlib import contextmanager
2

3
class Settings:
4
    debug_mode = False
5
    verbose_logging = False
6

7
settings = Settings()
8

9
@contextmanager
10
def temp_settings(**kwargs):
11
    # 保存原始值
12
    original = {}
13
    for key, value in kwargs.items():
14
        if hasattr(settings, key):
15
            original[key] = getattr(settings, key)
16
            setattr(settings, key, value)
17

18
    try:
19
        yield settings
20
    finally:
21
        # 恢复原始值
22
        for key, value in original.items():
23
            setattr(settings, key, value)
24

25
# 使用
26
print(f"调试模式: {settings.debug_mode}")  # False
27

28
with temp_settings(debug_mode=True, verbose_logging=True):
29
    print(f"临时开启: {settings.debug_mode}")  # True
30
    # 执行需要调试的代码
31

32
print(f"恢复: {settings.debug_mode}")  # False

1
# ❌ 错误：如果第二个 open 失败，第一个文件不会被关闭
2
class BadMultiFile:
3
    def __enter__(self):
4
        self.f1 = open("file1.txt")
5
        self.f2 = open("file2.txt")  # 这里失败的话 f1 泄漏
6
        return self
7

8
# ✅ 正确：使用 ExitStack 保证部分成功也能释放
9
from contextlib import ExitStack
10

11
class GoodMultiFile:
12
    def __enter__(self):
13
        self._stack = ExitStack()
14
        try:
15
            self.f1 = self._stack.enter_context(open("file1.txt"))
16
            self.f2 = self._stack.enter_context(open("file2.txt"))
17
        except:
18
            self._stack.close()  # 释放已成功打开的资源
19
            raise
20
        return self
21

22
    def __exit__(self, *args):
23
        self._stack.close()
24
        return False

1
# ❌ 危险：如果代码块中抛出异常，finally 部分永远不会执行
2
@contextmanager
3
def dangerous():
4
    resource = acquire()
5
    yield resource  # 如果这里抛出异常，下方不会执行
6
    release(resource)
7

8
# ✅ 正确
9
@contextmanager
10
def safe():
11
    resource = acquire()
12
    try:
13
        yield resource
14
    finally:
15
        release(resource)

1
# ❌ 危险：隐藏所有错误，使调试变得困难
2
class SilentContext:
3
    def __exit__(self, *args):
4
        return True  # 抑制所有异常
5

6
# ✅ 正确：只处理预期的异常
7
def __exit__(self, exc_type, exc_val, exc_tb):
8
    if exc_type is ValueError:
9
        print(f"处理已知错误: {exc_val}")
10
        return True
11
    return False  # 其他异常继续抛出

1
from contextlib import ContextDecorator
2

3
class Timer(ContextDecorator):
4
    def __enter__(self):
5
        import time
6
        self.start = time.perf_counter()
7
        return self
8

9
    def __exit__(self, exc_type, exc_val, exc_tb):
10
        elapsed = time.perf_counter() - self.start
11
        print(f"耗时: {elapsed:.4f}s")
12
        return False
13

14
# 作为装饰器使用
15
@Timer()
16
def process_data():
17
    total = sum(range(100_000))
18
    return total
19

20
process_data()  # 自动计时

1
import asyncio
2

3
class AsyncDatabase:
4
    async def __aenter__(self):
5
        self.conn = await self._async_connect()
6
        return self.conn
7

8
    async def __aexit__(self, exc_type, exc_val, exc_tb):
9
        await self.conn.close()
10

11
    async def _async_connect(self):
12
        # 模拟异步连接
13
        await asyncio.sleep(0.1)
14
        return type("Connection", (), {"close": lambda self: asyncio.sleep(0.05)})()
15

16
# 使用 async with
17
async def main():
18
    async with AsyncDatabase() as conn:
19
        # 使用连接
20
        pass