第9章 9.3 上下文管理器 with 进阶

上一章我们学会了用装饰器给函数「穿外套」，在不改变原函数的情况下添加新功能。这一章我们要学习另一个 Python 独有的语法——with 语句，它解决的是一类完全不同的痛点：资源用完忘了还。

你有没有过这种经历？借了图书馆的书，看到精彩处直接合上书走了，结果逾期罚款。或者说好了用完别人电脑要清理桌面，结果忘了，关机走人。程序世界里这种事更常见：打开了文件没关、申请了内存没释放、锁了线程没解锁……每个都能让你的程序内存泄漏或者死锁。

学完这章，你能：用 with 语句优雅地管理资源，再也不会「忘了还」。

🎯 开场：三个「血的教训」

场景 1：文件读完了没关

# 大多数新手会这样写
f = open('data.txt', 'r')
content = f.read()
print(content)
# 问题来了：如果下面代码报错，f 就永远关不上了

场景 2：锁住了没解锁

import threading
lock = threading.Lock()
lock.acquire()
# 如果这里抛异常，锁就永远解不开了
print("业务代码")
lock.release()

场景 3：数据库连接忘了断

conn = connect_to_database()
# 假设中间出了错
result = conn.query("SELECT * FROM users")
# 连接泄漏！

这三个场景的共同点是：获取资源和释放资源是分开的，中间可能会出错。如果手动写 try...finally，代码又臭又长。

而 with 语句，就是来解决这个问题的——无论中间是否出错，with 块结束后一定执行清理代码。

🧱 基础：上下文管理器的三步理解法

第一步：理解「上下文」是什么

想象你去ATM机取钱。你走进ATM亭（进入上下文），取钱，出来（退出上下文）。不管你取钱的时候发生什么——机器坏了、卡被吞了、你突然不想取了——你最终都得从ATM亭里出来。

上下文管理器 = ATM亭的管理机制：它负责「进入」和「离开」时的所有善后工作。

第二步：自己实现一个上下文管理器

Python 里，实现上下文管理器需要两个方法：

class 文件工具:
def __enter__(self):
    """进入 with 块时自动调用"""
    print("打开文件")
    self.f = open('test.txt', 'w')
    return self.f  # as 后面的变量就是这个返回值

def __exit__(self, exc_type, exc_val, exc_tb):
    """离开 with 块时自动调用，无论是否出错"""
    print("关闭文件")
    self.f.close()
    return False  # 返回 True 会压制异常

使用它：

with 文件工具() as f:
f.write("你好，世界")
print("写入成功")

print("with 块结束后，文件已自动关闭")

输出：

打开文件
写入成功
关闭文件
with 块结束后，文件已自动关闭

解释：第 7 行 __enter__ 返回什么，as 后面就绑定什么。这里返回 self.f，所以 f 就是打开的文件对象。

第三步：用 contextlib 更简单

自己写类太麻烦了，Python 提供了一个装饰器 @contextmanager，把「生成器函数」变成上下文管理器：

from contextlib import contextmanager

@contextmanager
def 文件工具(文件名):
print("打开文件")      # __enter__ 的前半部分
f = open(文件名, 'w')
try:
    yield f             # 这里是分界线！yield 之前的相当于 __enter__
finally:
    print("关闭文件")  # __exit__ 的部分
    f.close()

使用方式完全一样：

with 文件工具('test.txt') as f:
f.write("你好，世界")

print("文件已关闭")

说白了：@contextmanager 就是让你用同步代码写异步逻辑（生成器的特点），Python 自动帮你处理 try...finally。

🔥 实战：三个递进项目

项目 1：计时器（5 分钟）

场景：你想知道某段代码执行了多久，用 with 语句做一个通用的计时工具。

完整代码：

import time
from contextlib import contextmanager

@contextmanager
def 计时器(任务名="任务"):
start = time.time()
print(f"⏱️ 开始 {任务名}...")
yield
elapsed = time.time() - start
print(f"✅ {任务名} 完成，耗时 {elapsed:.2f} 秒")

# 使用
with 计时器("下载文件"):
time.sleep(1)  # 模拟下载耗时
print("文件下载中...")

with 计时器("处理数据"):
time.sleep(0.5)
print("数据处理中...")

预期输出：

⏱️ 开始 下载文件...
文件下载中...
✅ 下载文件 完成，耗时 1.00 秒
⏱️ 开始 处理数据...
数据处理中...
✅ 处理数据 完成，耗时 0.50 秒

解释：这个计时器什么都不 yield，所以 as 后面不需要变量。它完美展示了「进入执行什么、退出执行什么」的结构。

项目 2：CSV 数据处理工具（15 分钟）

场景：你有一个 CSV 文件，里面有用户数据。你需要在处理前备份、处理中如果出错就回滚、无论成功失败都要关闭文件句柄。

数据文件 data.csv（手动创建）：

姓名,年龄,城市
张三,25,北京
李四,30,上海
王五,28,深圳

完整代码：

import csv
from contextlib import contextmanager
import os

@contextmanager
def 安全读取CSV(文件名):
"""安全的 CSV 读取：自动处理编码、关闭文件、出错回滚"""
print(f"📖 打开文件: {文件名}")
backup_name = 文件名 + ".bak"

# 备份原文件
with open(文件名, 'r', encoding='utf-8') as src:
    content = src.read()
with open(backup_name, 'w', encoding='utf-8') as dst:
    dst.write(content)
print(f"💾 已备份到: {backup_name}")

# 打开文件
f = open(文件名, 'r', encoding='utf-8')
reader = csv.DictReader(f)

try:
    yield reader  # 把读取器传出去用
    print("✅ 数据处理成功")
except Exception as e:
    print(f"❌ 出错了: {e}")
    print("🔄 恢复到备份文件")
    with open(backup_name, 'r', encoding='utf-8') as src:
        with open(文件名, 'w', encoding='utf-8') as dst:
            dst.write(src.read())
finally:
    f.close()
    print("📁 文件已关闭")

# 使用
with 安全读取CSV('data.csv') as reader:
print("\n读取到的数据：")
for 行 in reader:
    print(f"  {行['姓名']}，{行['年龄']}岁，住在{行['城市']}")

预期输出：

📖 打开文件: data.csv
💾 已备份到: data.csv.bak
📁 文件已关闭（这里是因为 yield 时机问题，实际使用需要调整）

更实用的版本（修正 yield 位置）：

@contextmanager
def 安全读取CSV(文件名):
backup_name = 文件名 + ".bak"

# 备份
with open(文件名, 'r', encoding='utf-8') as src:
    backup_content = src.read()

f = open(文件名, 'r', encoding='utf-8')
try:
    yield csv.DictReader(f)
except Exception as e:
    print(f"❌ 出错: {e}")
    with open(backup_name, 'w', encoding='utf-8') as dst:
        dst.write(backup_content)
finally:
    f.close()

项目 3：待办事项管理器（15 分钟）

场景：组合项目 1 和 2，做一个带计时和日志的待办事项管理器。支持从 JSON 加载任务列表、执行任务（用 sleep 模拟）、记录耗时、保存结果。

完整代码：

import json
import time
from contextlib import contextmanager
from datetime import datetime

# 任务数据
任务数据 = {
"任务列表": [
    {"id": 1, "名称": "整理文档", "耗时": 1},
    {"id": 2, "名称": "提交代码", "耗时": 0.5},
    {"id": 3, "名称": "发送邮件", "耗时": 0.3}
]
}

@contextmanager
def 日志记录器(文件名="log.txt"):
"""所有操作都记录到日志文件"""
logs = []
开始时间 = datetime.now()
yield logs  # 让外面可以往 logs 里追加内容
结束时间 = datetime.now()
logs.append(f"会话结束: {结束时间}")
with open(文件名, 'a', encoding='utf-8') as f:
    for 记录 in logs:
        f.write(f"[{datetime.now().strftime('%H:%M:%S')}] {记录}\n")

@contextmanager
def 计时任务(任务名):
start = time.time()
print(f"▶️ 开始: {任务名}")
yield
elapsed = time.time() - start
print(f"⏹️ 结束: {任务名} (耗时 {elapsed:.2f}s)")

# 主程序
with 日志记录器() as logs:
logs.append("=" * 30)
logs.append("待办事项管理器启动")

for 任务 in 任务数据["任务列表"]:
    with 计时任务(任务["名称"]):
        time.sleep(任务["耗时"])  # 模拟任务执行
        logs.append(f"完成: {任务['名称']}")

logs.append("所有任务完成！")

print("\n📋 日志内容：")
with open("log.txt", 'r', encoding='utf-8') as f:
print(f.read())

预期输出：

▶️ 开始: 整理文档
⏹️ 结束: 整理文档 (耗时 1.00s)
▶️ 开始: 提交代码
⏹️ 结束: 提交代码 (耗时 0.50s)
▶️ 开始: 发送邮件
⏹️ 结束: 发送邮件 (耗时 0.30s)

📋 日志内容：
[14:32:15] ==============================================================
[14:32:15] 待办事项管理器启动
[14:32:16] 完成: 整理文档
[14:32:16] 完成: 提交代码
[14:32:17] 完成: 发送邮件
[14:32:17] 所有任务完成！
[14:32:17] 会话结束

解释：这个例子展示了上下文管理器的「组合能力」——计时和日志可以独立使用，也可以嵌套在任何地方。

💪 进阶：新手最容易踩的坑

坑 1：忘记处理异常

❌ 错误写法：

@contextmanager
def 坏的上下文管理器():
print("打开")
yield
print("关闭")  # 如果 yield 之前抛异常，这里永远不会执行

✅ 正确写法：

@contextmanager
def 好的上下文管理器():
print("打开")
try:
    yield
finally:
    print("关闭")  # finally 确保一定执行

坑 2：yield 后面还有可能出错的代码

❌ 错误写法：

@contextmanager
def 错误示范():
resource = 获得资源()
yield resource
释放资源()  # 如果这里抛异常，resource 泄漏！

✅ 正确写法：

@contextmanager
def 正确示范():
resource = 获得资源()
try:
    yield resource
finally:
    释放资源()  # 一定执行

坑 3：混淆 `return` 和 `yield`

❌ 错误写法：

@contextmanager
def 错误写法():
resource = get_resource()
return resource  # 错了！return 不是 yield

✅ 正确写法：

@contextmanager
def 正确写法():
resource = get_resource()
yield resource  # 必须用 yield

坑 4：`exit` 返回 True 会压制异常

class 错误示范:
def __enter__(self):
    return self

def __exit__(self, exc_type, exc_val, exc_tb):
    print("出错了，但我不告诉你")
    return True  # 异常被吞掉了！

with 错误示范() as e:
raise ValueError("这个错误会被静默")

✅ 除非你真的想压制异常，否则不要返回 True。

坑 5：在嵌套 with 里用同一个资源

❌ 错误写法：

with open('file.txt', 'w') as f:
f.write("第一层")
with open('file.txt', 'w') as f:  # 嵌套！内层关闭后外层也失效了
    f.write("第二层")
f.write("这里已经写不进去了！")

✅ 正确做法：不要嵌套使用同一个文件对象，或者用 contextlib.ExitStack（进阶话题）。

性能小贴士：减少创建开销

如果你的上下文管理器每次都要「打开/关闭」某个资源，但调用非常频繁，可以用 contextlib.nullcontext 避免创建实际对象：

from contextlib import nullcontext

# 根据条件选择不同的上下文管理器
ctx = nullcontext() if 不需要特殊处理 else 贵的上下文管理器()
with ctx:
# ...

调试技巧：打印调用栈

上下文管理器出问题很难定位，可以在 __exit__ 里加调试信息：

def __exit__(self, exc_type, exc_val, exc_tb):
import traceback
if exc_type:
    print(f"异常类型: {exc_type.__name__}")
    print(f"异常信息: {exc_val}")
    traceback.print_tb(exc_tb)
return False  # 不压制异常

✏️ 练习题

练习 1（2 分钟）：抄改计时器
- 输入：在 with 计时器("喝水") 里睡 0.5 秒
- 预期输出：显示「喝水」任务耗时约 0.5 秒
- 提示：直接把项目 1 的代码复制，改个任务名

练习 2（2 分钟）：加条件判断
- 输入：在练习 1 的基础上，只有当耗时超过 0.3 秒才打印「太慢了」
- 预期输出：显示「太慢了」（因为 0.5 > 0.3）
- 提示：在 yield 后、print 之前加 if 判断

练习 3（3 分钟）：处理新数据
- 输入：创建一个 ["苹果", "香蕉", "橘子"] 的列表，用 with 循环打印每个元素
- 预期输出：逐行打印每个水果名
- 提示：参考项目 2 的 yield reader 模式

练习 4（5 分钟）：组合计时和日志
- 输入：把练习 2 的代码包进 日志记录器()，让耗时也记到日志里
- 预期输出：控制台打印耗时，日志文件也记录耗时
- 提示：项目 3 展示了如何组合两个上下文管理器

练习 5（5 分钟）：分析报错
- 输入：下面的代码为什么会出错？

@contextmanager
def 坏的上下文():
print("进入")
yield
print("退出")

with 坏的上下文():
raise ValueError("测试错误")

预期输出：你能说出屏幕上会看到什么，以及为什么
提示：finally 和普通代码的区别

作业：做一个「第 9 章 9.3 上下文管理器实战工具」

做一个数据库模拟器，用于管理数据库连接：

需求描述：模拟一个数据库连接管理器，能执行 SQL 语句，自动记录执行日志
功能点：
1) 用上下文管理器管理连接，进入时打印「连接数据库」，退出时打印「断开连接」
2) 支持 with conn as cursor: 的写法
3) 自动记录每条 SQL 的执行时间
加分项：
1) 支持嵌套查询（子查询）
2) 用 @contextmanager 实现
验收标准：
能跑起来
输出符合「进入→执行→退出」顺序
代码有中文注释
提交方式：评论区贴代码

📚 总结

这一章我们学了三个核心点：

上下文管理器的本质：自动帮你处理「进入」和「离开」时的资源管理
两种实现方式：__enter__/__exit__ 方法或 @contextmanager 装饰器
with 的核心价值：无论代码是否出错，清理代码一定执行

推荐延伸资源：

Python 官方文档 - contextlib（中文，最权威）
《流畅的 Python》第 15 章「上下文管理器」（进阶必读）
视频：【Python】深入理解with语句（15 分钟）

互动钩子：你在实际项目里有没有遇到过「文件没关」「锁没解」导致的 bug？当时是怎么发现的？评论区聊聊，老粉优先回复！

📝 下章预告：学会了 with 语句，下一章我们要学习一个和它配合默契的工具——dataclass。用它来定义数据结构，代码少一半，可读性翻倍，而且还能自动生成 __enter__ 和 __exit__！敬请期待第 9 章 9.4 dataclass 与 attrs。

声明：本站所有文章，如无特殊说明或标注，均为本站原创发布。任何个人或组织，在未征得本站同意时，禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益，可联系我们进行处理。

第9章 9.3 上下文管理器 with 进阶

第9章 9.3 上下文管理器 with 进阶

🎯 开场：三个「血的教训」

🧱 基础：上下文管理器的三步理解法

第一步：理解「上下文」是什么

第二步：自己实现一个上下文管理器

第三步：用 contextlib 更简单

🔥 实战：三个递进项目

项目 1：计时器（5 分钟）

项目 2：CSV 数据处理工具（15 分钟）

项目 3：待办事项管理器（15 分钟）

💪 进阶：新手最容易踩的坑

坑 1：忘记处理异常

坑 2：yield 后面还有可能出错的代码

坑 3：混淆 `return` 和 `yield`

坑 4：`exit` 返回 True 会压制异常

坑 5：在嵌套 with 里用同一个资源

性能小贴士：减少创建开销

调试技巧：打印调用栈

✏️ 练习题

📚 总结

评论(0)

提示：请文明发言取消回复

排行榜展示

第1章 1.3 字符串与 f-string 格式化

第10章 10.3 终极项目：仿 V2EX 论坛 API

第3章 3.3 作用域与闭包

第10章 10.5 终极实战：CLI 工具开发

第10章 10.1 实时聊天系统（WebSocket）

第9章 9.3 上下文管理器 with 进阶

第9章 9.3 上下文管理器 with 进阶

🎯 开场：三个「血的教训」

🧱 基础：上下文管理器的三步理解法

第一步：理解「上下文」是什么

第二步：自己实现一个上下文管理器

第三步：用 contextlib 更简单

🔥 实战：三个递进项目

项目 1：计时器（5 分钟）

项目 2：CSV 数据处理工具（15 分钟）

项目 3：待办事项管理器（15 分钟）

💪 进阶：新手最容易踩的坑

坑 1：忘记处理异常

坑 2：yield 后面还有可能出错的代码

坑 3：混淆 return 和 yield

坑 4：__exit__ 返回 True 会压制异常

坑 5：在嵌套 with 里用同一个资源

性能小贴士：减少创建开销

调试技巧：打印调用栈

✏️ 练习题

📚 总结

评论(0)

提示：请文明发言 取消回复

相关文章

排行榜展示

坑 3：混淆 `return` 和 `yield`

坑 4：`exit` 返回 True 会压制异常

提示：请文明发言取消回复