你去图书馆借书看，完整的流程是：
无论书好不好看、无论你中途接不接电话，第三步一定要做——书必须还回去，否则其他人借不到。
编程中也有类似的"借还"场景：
| | |
|---|
| open() | close() |
| connect() | connection.close() |
| lock.acquire() | lock.release() |
| socket.connect() | socket.close() |
**上下文管理器（Context Manager）**就是帮你自动"归还"资源的机制。with 语句就是使用这个机制的语法。
f = open("data.txt", "r", encoding="utf-8")# 忘了 f.close()！文件句柄泄漏，可能导致其他程序无法访问该文件# === 错误写法2：中途出错，close 不会执行 ===f = open("data.txt", "r", encoding="utf-8")# 如果出错了，程序跳到 except，f.close() 被跳过# === 正确但啰嗦的写法3：try/finally === f = open("data.txt", "r", encoding="utf-8") f.close() # 无论如何都会执行，但写法很繁琐# 每次都要写 try/finally，太啰嗦了！withopen("data.txt", "r", encoding="utf-8") as f:# 出了 with 块，f 自动关闭，无需手动操作with 做的事情：你只要告诉它"借什么"，它自动帮你"还"。无论 with 块里的代码正常结束还是中途出错，资源都会被正确释放。
├── 基于类实现上下文管理器（__enter__ + __exit__） ├── __exit__ 异常处理机制（三个参数 + 返回值控制） │ ├── @contextmanager（用生成器快速实现） │ ├── closing()（把有 close 的对象变成上下文管理器） │ └── ContextDecorator（既是装饰器又是上下文管理器） └── 综合实战案例（计时器、临时清理、异常日志）
withopen("data.txt", "r") as f:Python 解释器在背后做了这些事：
步骤1：open("data.txt", "r") → 创建一个文件对象步骤2：调用 文件对象.__enter__() → 返回文件对象自身，赋值给 f步骤3：执行 with 块内的代码 → content = f.read()步骤4：无论步骤3是否出错，调用 文件对象.__exit__(exc_type, exc_val, traceback) - 如果步骤3没出错：传入 (None, None, None)管理器 = 表达式 # 例如：管理器 = open("data.txt")变量 = 管理器.__enter__() # 例如：f = 文件对象.__enter__()ifnot 管理器.__exit__(异常类型, 异常值, traceback):raise# __exit__ 返回 False → 重新抛出异常 管理器.__exit__(None, None, None)"""模拟文件打开的上下文管理器——帮助理解 with 原理"""def__init__(self, filename, mode):print(f" [__enter__] 打开文件：{self.filename}")self.file = open(self.filename, self.mode)returnself.file # 这个返回值赋给 as 后面的变量def__exit__(self, exc_type, exc_val, traceback):print(f" [__exit__] 关闭文件：{self.filename}")# 返回 False（或不返回）→ 异常继续向上抛with MyOpen("test.txt", "w") as f:print(" [with块] 写入数据...")# [__enter__] 打开文件：test.txt# [__exit__] 关闭文件：test.txtwith MyOpen("test.txt", "r") as f:raise ValueError("模拟一个错误！") # 故意抛出异常print(" [外部] 捕获到了异常，但文件已经关闭了！")# [__enter__] 打开文件：test.txt# [__exit__] 关闭文件：test.txt ← 即使出错，文件也关闭了！关键发现：即使在 with 块中抛出异常，__exit__ 仍然会被调用，资源仍然被正确释放。
withopen("a.txt") as f1, open("b.txt") as f2:# 写法3：不获取资源（纯控制）——某些上下文管理器 as 后面可以没有变量# 例如：with lock: 或 with redirect_stdout(f):# 写法4：多资源嵌套（Python 3.10+ 可用括号换行）open("input.txt") as f_in,open("output.txt", "w") as f_out,# 以下都是 Python 内置的上下文管理器，可以直接用 withwithopen("file.txt") as f:with lock: # 自动 acquire 和 releasewith sqlite3.connect("test.db") as conn: # 自动 commit/rollback 和 close cursor.execute("SELECT * FROM users")with tempfile.TemporaryDirectory() as tmpdir:from urllib.request import urlopenwith urlopen("https://www.python.org") as response:新手易踩坑：
坑 1：with 后面的表达式不能是任意对象 → 必须实现 __enter__ 和 __exit__ 方法坑 2：as f 只能获取 __enter__ 的返回值，不是 with 后面的表达式本身坑 3：出了 with 块后，资源已释放，不要再使用 f（如 f.read()）
用 with 语句实现：打开一个文件写入 "Hello World"，然后再次用 with 读出来并打印。对比用 try/finally 的写法，体会 with 的简洁。
点击查看参考答案withopen("hello.txt", "w", encoding="utf-8") as f:withopen("hello.txt", "r", encoding="utf-8") as f:print(f.read()) # Hello World# 对比 try/finally 写法（同样功能但更啰嗦） f = open("hello.txt", "w", encoding="utf-8") f = open("hello.txt", "r", encoding="utf-8")
三、基于类的上下文管理器：enter 与 exit要实现一个上下文管理器，类必须定义两个方法：
| | | |
|---|
__enter__(self) | | | |
__exit__(self, exc_type, exc_val, exc_tb) | | | True |
┌────────────────────────────────────────────┐│ with MyManager() as obj: ││ # 1. MyManager() 创建实例 ││ # 2. 调用实例.__enter__() → 返回 obj ││ # 4. 调用实例.__exit__(...) ││ # └─ 没出错 → (None, None, None) ││ # └─ 出错了 → (异常类型, 异常值, │└────────────────────────────────────────────┘"""文件操作的上下文管理器 —— 自动关闭文件"""def__init__(self, filename, mode="r", encoding="utf-8"):"""初始化：记录文件名和模式，但不打开文件"""self.file = None# 文件句柄，初始为 Noneprint(f"[打开] {self.filename}")self.file = open(self.filename, self.mode, encoding=self.encoding)returnself.file # 这个值赋给 as 后面的变量def__exit__(self, exc_type, exc_val, exc_tb):"""退出上下文：关闭文件（无论是否发生异常）"""print(f"[关闭] {self.filename}")# 返回 False（或不返回）：异常继续向上传播with ManagedFile("greeting.txt", "w") as f:with ManagedFile("greeting.txt", "r") as f:"""数据库连接的上下文管理器 —— 自动 commit/rollback 和关闭"""def__init__(self, db_path):print(f"[数据库] 连接 {self.db_path}")self.connection = sqlite3.connect(self.db_path)self.cursor = self.connection.cursor() CREATE TABLE IF NOT EXISTS students ( id INTEGER PRIMARY KEY AUTOINCREMENT,returnself.cursor # 返回游标，方便调用者直接执行 SQLdef__exit__(self, exc_type, exc_val, exc_tb):"""退出上下文：根据是否有异常决定 commit 还是 rollback"""self.connection.rollback()print(f"[数据库] 回滚事务 ✗ (原因: {exc_type.__name__}: {exc_val})")with DatabaseConnection("school.db") as db: db.execute("INSERT INTO students (name, score) VALUES (?, ?)", ("小明", 95)) db.execute("INSERT INTO students (name, score) VALUES (?, ?)", ("小红", 88))with DatabaseConnection("school.db") as db: db.execute("INSERT INTO students (name, score) VALUES (?, ?)", ("小刚", 75))raise ValueError("模拟数据库写入后发生的错误！")# [数据库] 回滚事务 ✗ (原因: ValueError: 模拟数据库写入后发生的错误！)with DatabaseConnection("school.db") as db: result = db.execute("SELECT * FROM students").fetchall()# 只有小明和小红（正常提交的），小刚的记录已回滚"""倒计时上下文管理器 —— __enter__ 返回自身"""def__init__(self, start):print(f"倒计时开始：{self.start}")returnself# 返回自身，调用方可以通过 as 拿到实例def__exit__(self, exc_type, exc_val, exc_tb):print(f"倒计时结束（已数到 {self.current}）")print(f" {self.current}...")# 使用：as 拿到的是 Countdown 实例本身新手易踩坑：
坑 1：__exit__ 方法忘了写第三个参数 → 参数数量不匹配，调用时报 TypeError坑 2：__exit__ 中 return False 写成了 return → return 等同于 return None，None 在布尔上下文中是 False，所以效果一样，但不够明确坑 3：在 __enter__ 里初始化了资源但在 __exit__ 里没有释放 → 资源泄漏，背离上下文管理器的设计初衷
自己写一个 ReversibleList 上下文管理器：进入时打印列表内容，退出时自动将列表反转。__enter__ 返回自身。
with ReversibleList([1, 2, 3, 4, 5]) as rl:# 退出（已反转）：[5, 4, 3, 2, 1]点击查看参考答案def__exit__(self, exc_type, exc_val, exc_tb):print(f"退出（已反转）：{self.data}")with ReversibleList([1, 2, 3, 4, 5]) as rl:
__exit__(self, exc_type, exc_val, exc_tb) 的三个参数：
| | | |
|---|
exc_type | | None | |
exc_val | | None | |
exc_tb | | None | |
"""演示 __exit__ 返回值对异常传播的影响"""def__exit__(self, exc_type, exc_val, exc_tb):print(f"[退出] exc_type={exc_type}, exc_val={exc_val}")# === 情况1：返回 False → 异常继续向上传播（默认行为） ===# === 情况2：返回 True → 吞掉异常，外部感知不到 ===# 测试：__exit__ 返回 True 会吞掉异常print("=== 测试：异常被吞掉 ===")raise ValueError("一个错误！")print("这行竟然被执行了！因为异常被 __exit__ 吞掉了。")# [退出] exc_type=<class 'ValueError'>, exc_val=一个错误！# 这行竟然被执行了！因为异常被 __exit__ 吞掉了。"""只吞掉 KeyError，其他异常正常抛出"""def__exit__(self, exc_type, exc_val, exc_tb):print(f"[已抑制] KeyError: {exc_val}")print("=== 测试1：KeyError 被吞掉 ===")print(data["age"]) # 触发 KeyError —— 被吞掉了print("=== 测试2：ValueError 正常抛出 ===")raise ValueError("其他错误") # ValueError —— 不会被吞# === 测试1：KeyError 被吞掉 ===# === 测试2：ValueError 正常抛出 ===def__init__(self, log_file="error.log"):def__exit__(self, exc_type, exc_val, exc_tb):withopen(self.log_file, "a", encoding="utf-8") as f: f.write(f"[异常类型] {exc_type.__name__}\n") f.write(f"[异常信息] {exc_val}\n")# traceback.format_tb() 获取格式化的调用栈 traceback.print_tb(exc_tb, file=f)print(f"异常已记录到 {self.log_file}")except ZeroDivisionError:print("外部捕获了 ZeroDivisionError")# [异常类型] ZeroDivisionError# [异常信息] division by zero# File "xxx.py", line xx, in process# File "xxx.py", line xx, in calculate# ---------------------------------------- ┌─── 没出错 ───→ __exit__(None, None, None) └─── 出错了 ───→ __exit__(ValueError, ValueError("msg"), traceback) ├──→ returnTrue → 吞掉异常，外部无感知 └──→ returnFalse → 异常继续向上传播到外部新手易踩坑：
坑 1：__exit__ 返回 True 吞掉了所有异常 → 调试时完全不知道哪里出错了！坑 2：在 __exit__ 里重新抛出异常（raise）→ 会覆盖原始异常，让调试变得更困难坑 3：__exit__ 中的代码本身可能抛异常 → 如果在关闭资源时出错，Python 会优先抛出 __exit__ 中的异常，原始的异常信息可能丢失
写一个 LogIfError 上下文管理器：只有当 with 块发生异常时，才将异常类型和异常信息打印出来。异常照常传播，不吞掉。
点击查看参考答案def__exit__(self, exc_type, exc_val, exc_tb):print(f"[异常日志] {exc_type.__name__}: {exc_val}")raise RuntimeError("测试异常")# [异常日志] RuntimeError: 测试异常
contextlib 模块提供了更便捷的方式来创建和使用上下文管理器，不需要每次都写一个完整的类。
5.1 @contextmanager：用生成器实现上下文管理器这是最常用的工具——用一个生成器函数就能创建上下文管理器，代码量减少 80%。
from contextlib import contextmanagerdefmanaged_file(filename, mode="r", encoding="utf-8"):"""用生成器实现的文件上下文管理器 —— 无需写类！"""print(f"[打开] {filename}") f = open(filename, mode, encoding=encoding) # __enter__ 的逻辑print(f"[关闭] {filename}")with managed_file("test.txt", "w") as f: f.write("用生成器实现的上下文管理器！")with managed_file("test.txt", "r") as f:@contextmanager 的工作原理：
@contextmanager 装饰的生成器函数：yield 之前的代码 → 等价于 __enter__yield 后面的值 → 等价于 __enter__ 的返回值（赋给 as）yield 之后的代码 → 等价于 __exit__（放在 finally 中确保执行）┌─────────────────────────────────┐│ ├── yield 之前的代码（准备资源） │ ← __enter__│ ├── yield 资源 → 交给 with 块 ││ ├── yield 之后的代码（清理资源） │ ← __exit__└─────────────────────────────────┘5.2 @contextmanager 的异常处理from contextlib import contextmanager"""演示 @contextmanager 中的异常处理"""yieldf"资源-{name}"# 正常返回资源# 如果捕获了异常，就不会向上传播（相当于 __exit__ 返回 True）print(f"[捕获] 在上下文管理器内部处理了：{e}")# 其他异常 → 在 finally 之前重新抛出print("[发现] 其他类型异常，重新抛出")# finally 块中的代码无论异常与否都会执行with safe_operation("测试1") as res:print("\n=== 测试2：ValueError ===")with safe_operation("测试2") as res:print("ValueError 被吞掉了，程序继续！")print("\n=== 测试3：RuntimeError ===")with safe_operation("测试3") as res:raise RuntimeError("运行时错误")except RuntimeError as e:from contextlib import contextmanagerdefbroken_manager(filename): f.close() # 如果 with 块中出现异常，这行不会执行！defcorrect_manager(filename):新手易踩坑：
坑 1：@contextmanager 里 yield 后面的代码没用 try/finally 包裹 → 出错时资源不会释放坑 2：@contextmanager 里 yield 只能出现一次 → 生成器只能 yield 一个值坑 3：yield 返回的值赋给了 as 后面的变量，yield 本身没有值
5.4 closing()：给有 close() 的对象"披上"上下文管理器外衣很多第三方库的对象有 close() 方法但没有实现 __enter__/__exit__。closing() 就是为这些对象服务的。
from contextlib import closingfrom urllib.request import urlopen# urlopen 返回的对象有 close() 但没有实现上下文管理协议（旧版本）response = urlopen("https://httpbin.org/get")with closing(urlopen("https://httpbin.org/get")) as response:# 退出 with 时自动调用 response.close()# def __init__(self, thing):# def __exit__(self, *args):from contextlib import suppress# 传统写法：用 try/except 忽略 KeyErrorprint(data["age"]) # KeyError 被吞掉，程序继续from contextlib import suppresswith suppress(FileNotFoundError): os.remove("maybe_not_exist.txt")with suppress(KeyError, IndexError, TypeError): value = some_risky_operation()5.6 ContextDecorator：既是上下文管理器又是装饰器from contextlib import ContextDecoratorclasslog_execution(ContextDecorator):"""既可以当上下文管理器用，也可以当装饰器用！"""def__init__(self, prefix=""):print(f"{self.prefix}[进入]")def__exit__(self, *args):print(f"{self.prefix}[退出]")with log_execution("[CM] "):# 用法2：装饰器（整个函数自动包裹在 with 中！）5.7 redirect_stdout / redirect_stderr：重定向标准输出from contextlib import redirect_stdout# 场景：捕获 print 的输出，而不是让它打印到控制台f = io.StringIO() # 一个内存中的"文件"with redirect_stdout(f): # 把 print 的输出重定向到 fprint("所有的 print 都进了 StringIO")output = f.getvalue() # 获取被截获的输出 | | |
|---|
@contextmanager | | |
closing(thing) | | |
suppress(*exceptions) | | |
ContextDecorator | | |
redirect_stdout(f) | | |
redirect_stderr(f) | | |
ExitStack | | |
用 @contextmanager 实现一个 timer 上下文管理器，统计 with 块中代码的运行时间（秒）。
点击查看参考答案from contextlib import contextmanager start = time.perf_counter() elapsed = time.perf_counter() - startprint(f"[{name}] 耗时：{elapsed:.4f} 秒") total = sum(range(10000000))
需求：统计 with 块内代码的执行时间，支持命名、支持获取详细统计。
from contextlib import contextmanager"""高性能计时器 —— 统计代码块运行时间"""def__init__(self, name="代码块"):# 使用 perf_counter 而非 time.time（精度更高，不受系统时间调整影响）self.start_time = time.perf_counter()returnself# 返回自身，外部可以获取详细数据def__exit__(self, *args):self.end_time = time.perf_counter()self.elapsed = self.end_time - self.start_timeprint(f"[Timer] {self.name}: {self.elapsed * 1_000_000:.2f} μs")print(f"[Timer] {self.name}: {self.elapsed * 1000:.2f} ms")print(f"[Timer] {self.name}: {self.elapsed:.4f} s")with Timer("大数据排序") as t: data = list(range(10000000, 0, -1))# [Timer] 大数据排序: 0.3241 s result = sum(range(1000))需求：在 with 块内创建一个临时文件（或目录），退出时自动删除，无论代码是否出错。
"""临时工作目录 —— 退出时自动清理所有内容"""def__init__(self, prefix="workspace_", keep_on_error=False): keep_on_error: 如果 with 块出错，是否保留目录（调试用）self.keep_on_error = keep_on_errorself.path = Path(tempfile.mkdtemp(prefix=self.prefix))print(f"[创建] 临时目录：{self.path}")def__exit__(self, exc_type, exc_val, exc_tb):if exc_type isnotNoneandself.keep_on_error:print(f"[保留] 因出错保留目录：{self.path}") shutil.rmtree(self.path, ignore_errors=True)print(f"[清理] 临时目录已删除：{self.path}")with TemporaryWorkspace("data_") as ws: (ws / "input.txt").write_text("临时数据", encoding="utf-8") (ws / "output.csv").write_text("列1,列2\n1,2", encoding="utf-8")# [创建] 临时目录：C:\Users\...\Temp\data_xxxxxx# [清理] 临时目录已删除：C:\Users\...\Temp\data_xxxxxxwith TemporaryWorkspace("debug_", keep_on_error=True) as ws: (ws / "partial_result.txt").write_text("部分数据", encoding="utf-8")raise RuntimeError("处理过程中出错！")# [创建] 临时目录：C:\Users\...\Temp\debug_xxxxxx# [保留] 因出错保留目录：C:\Users\...\Temp\debug_xxxxxx需求：包裹一段代码，捕获指定类型的异常，记录日志，支持重试。
from datetime import datetimeformat="%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S"def__init__(self, operation_name, retry=0, delay=1, catch=(Exception,), reraise=True): operation_name: 操作名称（用于日志）self.name = operation_nameprint(f"[操作] {self.name}")print(f"[时间] {datetime.now().strftime('%H:%M:%S')}")def__exit__(self, exc_type, exc_val, exc_tb):ifissubclass(exc_type, self.catch):print(f"[失败] 第 {self.attempt} 次尝试：{exc_type.__name__}: {exc_val}")# 还有重试次数 → 吞掉异常，让外部代码进入下一次尝试ifself.attempt <= self.retry:print(f"[重试] 等待 {self.delay} 秒后重试...")self.final_error = exc_valprint(f"[放弃] 已重试 {self.retry} 次，仍然失败")print(f"[不可处理] {exc_type.__name__}（不在捕获范围内）")if random.random() < 0.4:raise ConnectionError("网络连接超时")for _ inrange(5): # 外层最多尝试 5 次with ErrorMonitor("网络数据获取", retry=3, delay=1, catch=(ConnectionError, TimeoutError)) as monitor: result = flaky_operation()运行效果演示：
==================================================[失败] 第 1 次尝试：ConnectionError: 网络连接超时==================================================[失败] 第 2 次尝试：ConnectionError: 网络连接超时====================================================================================================from contextlib import contextmanager, redirect_stdoutdefmeasure_and_capture(name="操作"): start = time.perf_counter()# 嵌套两个上下文：redirect_stdout 在内部yield f # 把 StringIO 传给 with 块 elapsed = time.perf_counter() - startprint(f"[{name}] 耗时 {elapsed:.4f}s")print(f"[{name}] 产生的输出：")for line in output.strip().split("\n"):with measure_and_capture("数据处理") as captured: total = sum(range(1000000))print(f"步骤2：计算结果 = {total}")# │ 步骤2：计算结果 = 499999500000实现一个 Transaction 上下文管理器，模拟银行转账的事务处理：
无异常时打印"提交事务"，有异常时打印"回滚事务" + 异常原因点击查看参考答案def__init__(self, name="事务"):print(f"[{self.name}] 开始")def__exit__(self, exc_type, exc_val, exc_tb):print(f"[{self.name}] 提交 ✓")print(f"[{self.name}] 回滚 ✗ ({exc_type.__name__}: {exc_val})")raise ValueError("李四账户不存在！")# [转账] 回滚 ✗ (ValueError: 李四账户不存在！)
错误 1：把 with 当成"作用域"来限制变量访问
# 错误理解：以为出了 with 就不能用 f 了withopen("test.txt") as f:# f 仍然存在！只是 f.close() 已经被调用了print(f.closed) # True —— 文件已关闭# f.read() # ValueError: I/O operation on closed file.# 如果调用 f.read()，会报错——不是变量不存在，而是文件已关闭错误 2：在 @contextmanager 中忘了 try/finally
# 错误：yield 后的代码不在 finally 中defbad_manager(filename): f.close() # 如果 yield 处抛出异常，这行不会执行！defgood_manager(filename):错误 3：exit 吞掉异常但不做任何处理
def__exit__(self, exc_type, exc_val, exc_tb):returnTrue# 吞掉所有异常！调试时完全不知道问题在哪# 正确：除非你清楚自己在做什么，否则 return False错误 4：混淆 with 和 try/except 的作用
# with 的作用是管理资源（自动释放），不是捕获异常# 错误期望：以为 with 会捕获 FileNotFoundErrorwithopen("not_exist.txt") as f: # FileNotFoundError 仍然会抛出！withopen("not_exist.txt") as f:except FileNotFoundError:错误 5：在 enter 里分配了资源但在 exit 出错时不处理
self.resource1 = acquire_resource1() # 分配成功self.resource2 = acquire_resource2() # 如果这里出错...# 问题：如果 resource2 分配失败，resource1 不会被释放！self.resource1 = acquire_resource1()self.resource2 = acquire_resource2()self.resource1.close() # 失败时手动释放 resource1 ┌─────────────┼─────────────┐ __enter__ with 代码块 __exit__ │ 没出错→(None,...) 出错了→(异常信息)核心心法三句话：
with = 自动借还：你只管用，__exit__ 帮你自动归还@contextmanager = 简化版：用生成器替代类，yield 前 = __enter__，yield 后 = __exit__**exit 返回 True 吞异常，False 传异常**：正常情况下用 False，除非你明确知道要吞掉
