从一段非常容易踩坑的代码开始说说我为什么想写这个话题。前阵子帮同事排查一个线上脚本问题症状很诡异程序跑完不报错但日志文件经常只有半截内容数据库连接偶尔提示“too many connections”。后来定位到根因其实就是文件没关闭、连接没归还资源在异常路径上泄漏了。当时他用的还是最原始的f open(...)、f.close()这种写法。我说你换成with open(...)不就好了他很困惑地问我with到底做了什么为什么以前教科书教我的手动close就不好使这个问题我觉得很有代表性。今天就用一整篇把Python上下文管理器with语句的原理、机制、实战和那些坑讲透不讲空泛理论所有结论都从可运行的代码里来。1. 资源泄漏的痛点为什么手动 close 靠不住先回到那个经典场景。刚学Python时几乎所有人都会写类似的代码f open(data.txt, r) data f.read() f.close() print(data)这段代码看起来人畜无害。但问题恰恰藏在这种“看起来没问题”的写法里。假如f.read()这行抛了异常比如文件被占用了、编码不对、磁盘IO出错了程序会直接跳过f.close()这行文件句柄就永远留在那里不释放了。一次两次没关系但脚本一旦跑在循环里、跑在长期运行的服务里文件描述符迟早会被吃光。Linux下默认文件描述符上限通常是1024超过之后open()直接报OSError: [Errno 24] Too many open files整个服务就瘫了。数据库连接、网络连接、锁资源的原理一模一样而且因为连接通常更昂贵、服务端有并发上限泄漏造成的危害更严重。你要说“我加个try...except...finally不就行了”确实可以这是标准解法try: f open(data.txt, r) data f.read() finally: f.close()但这不是Python社区推荐的做法。原因有二一是啰嗦每写一处资源操作就要套一个try...finally代码量翻倍二是容易漏人的注意力是有限的写业务逻辑时很容易忘了给某个资源加上finally兜底。Python的设计哲学里很重要的一条就是“让正确的事情更容易做”所以语言层面专门提供了with语句把“资源进来、用、确保释放”这三件事封装成一个不可分割的语法结构。这就是上下文管理器存在的根本动机它不是一个花哨的语法糖而是解决一个非常现实的工程问题资源的确定性释放。2. 协议拆解enter与exit这对双子星要理解with必须理解它背后的协议。Python里很多东西都是靠协议dunder方法运转的比如迭代器靠__iter__和__next__序列靠__len__和__getitem__上下文管理器靠的是两个特殊方法__enter__和__exit__。2.1 两者的职责与返回值__enter__(self)在进入with代码块之前被调用。它的返回值会赋给as后面的变量。需要注意的是它不是必须返回资源对象本身你可以返回任何东西甚至可以返回None。但惯例上如果这个上下文管理器代表一个资源就会把资源对象返回出去比如open()的返回值。__exit__(self, exc_type, exc_value, traceback)在代码块结束后被调用。无论代码块是正常跑完、还是中途抛了异常、还是被return/break/continue跳出它都会被调用。三个参数分别代表异常类型、异常实例、堆栈跟踪对象。如果代码块没有异常三个值都是None。关键来了__exit__的返回值是布尔类型实际上Python会对返回值做真值判断。如果返回True则表示“这个异常我已经处理掉了”with语句不会让它继续向外传播如果返回False或者None因为None会被当成假值异常会继续往外抛。2.2 最简单的自定义上下文管理器只看open()的内部实现你感受不到协议的具体工作方式自己动手写一个最直观。下面这个例子模拟一个“计时器”上下文管理器进入时记录开始时间退出时计算耗时import time class Timer: def __enter__(self): self.start time.perf_counter() return self def __exit__(self, exc_type, exc_value, traceback): self.elapsed time.perf_counter() - self.start print(f耗时: {self.elapsed:.4f} 秒) return False with Timer() as timer: # 模拟一些耗时操作 total sum(range(1000000)) print(timer.elapsed)这里有两个细节值得注意。第一as拿到的timer是Timer实例本身所以代码块执行完后还能通过timer.elapsed取到结果。第二return False是刻意写的表示“不拦截异常”。如果代码块里抛了异常这个计时器一样能正常统计耗时然后让异常继续跑出去。2.3 打开资源的标准范式再看一个资源型场景写一个上下文管理器来管理一次性临时目录。这是很多自动化测试脚本里刚需的功能import tempfile import shutil from pathlib import Path class TempDir: def __enter__(self): self.dir_path Path(tempfile.mkdtemp()) return self.dir_path def __exit__(self, exc_type, exc_value, traceback): shutil.rmtree(self.dir_path, ignore_errorsTrue) return False with TempDir() as tmp: target tmp / output.txt target.write_text(hello) print(target.exists()) # 离开with后临时目录已经被清理这类“用完必须清理清理动作绝对不能忘”的场景就是上下文管理器的主场。不需要你手动调用任何清理函数保证代码块里的逻辑无论成功还是失败shutil.rmtree一定会执行。3. with 语句底层到底做了什么执行流程全解析上一篇讲协议时可能有人已经在想__enter__的返回值给到as变量__exit__在结束时被调用但中间的异常是怎么流转的如果__enter__本身就抛了异常怎么办代码块里return了又会走到__exit__吗这一节把执行流程完整摊开讲。3.1 完整执行步骤以with EXPR as VAR:为例Python解释器实际执行的步骤可以拆成五步计算EXPR表达式得到上下文管理器对象。调用该对象的__enter__()方法把返回值赋给VAR。如果EXPR的求值本身就抛了异常后续全部不执行。执行with代码块里的完整语句体。执行完代码块后调用__exit__(None, None, None)。如果代码块内抛出了异常则带着异常信息调用__exit__(exc_type, exc_value, traceback)如果__exit__返回真值抑制异常如果返回假值异常继续抛给上层。这里有个容易误解的点__exit__并不是在“代码块正常结束”和“代码块异常”两条路径中各调用一次它只调用一次区别只在于传入的参数不同。调用方是统一的所谓“正常”与“异常”只是参数值不同。3.2 异常抑制的原理很多新手第一次看到__exit__能“吞掉异常”时会觉得不可思议。实际上这个机制非常巧妙。看下面这个场景我们希望某个操作中出现的特定异常静默处理class IgnoreValueError: def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): # 如果异常是 ValueError就返回 True 来抑制它 return exc_type is ValueError with IgnoreValueError(): number int(hello) # 触发 ValueError print(这里还能继续执行)正常情况下int(hello)会抛出ValueError导致程序崩溃但因为在__exit__里判断并返回了True这个异常就被“吸收”了。这也是标准库contextlib.suppress的实现原理后面再说。3.3enter的返回值与 as 变量的常见误区as不是必须写的。有些上下文管理器只关心进入和退出时的副作用返回值没有意义。但如果写了as你拿到什么完全取决于__enter__的设计。典型误区一以为拿到的是“被管理的资源”其实拿到的是上下文管理器自己。很多自定义类为了方便在__enter__里return self这时两者恰好相等但如果__enter__返回了别的东西比如file对象、cursor对象、或经过转换的对象那就完全不同。典型误区二as变量在离开with后仍然存在。Python用as赋值并不会像其他语言那样创造块级作用域变量在当前函数作用域里依然绑着那个对象。刚才的Timer例子就是利用了这一点。只是对象本身的资源状态可能已经“过期”了比如文件已经关闭。3.4 多个with怎么嵌套与合并同时打开两个文件做复制的常见需求两种写法等价# 写法一嵌套 with open(source.txt, r) as src: with open(target.txt, w) as dst: dst.write(src.read()) # 写法二逗号并列3.1 支持 with open(source.txt, r) as src, open(target.txt, w) as dst: dst.write(src.read())并列写法的执行顺序是从左到右先进入src再进入dst退出顺序则相反先退出dst再退出src。这符合资源释放的直觉就像先脱外套再脱帽子脱的顺序一定和穿的顺序相反。4. contextlib把上下文管理的复杂度藏起来前面写的都是基于类的实现需要定义__enter__和__exit__两个方法。但很多时候你只是想快速地把一小段逻辑变成一个上下文管理器不值得为此定义一个类。标准库contextlib就是专门解决这个问题的模块。4.1 contextmanager 装饰器用生成器伪装上下文管理器contextlib.contextmanager是一个装饰器可以把一个生成器函数变成上下文管理器。它内部替你做了一大堆状态机和异常处理的脏活让使用者只需关心yield前后的两段逻辑。基本形态如下from contextlib import contextmanager import time contextmanager def timer(): start time.perf_counter() try: yield # yield 之前的代码相当于 __enter__ finally: elapsed time.perf_counter() - start print(f耗时: {elapsed:.4f} 秒)重点在于这几点。第一yield之前的所有代码会在进入with时执行相当于__enter__。第二yield的产出值会赋给as后面的变量。第三**yield之后确切地说是finally块里的代码在退出时执行**相当于__exit__。第三点尤其关键yield后面的代码必须以try...finally包裹这样无论with块里发生什么正常结束也好、异常也好finally里的清理逻辑保证执行。再看一个生产实践中很常用的示例切换工作目录很多人写脚本时会有这个需求contextmanager def working_directory(path): import os origin os.getcwd() os.chdir(path) try: yield finally: os.chdir(origin) with working_directory(/tmp): # 在这里所有相对路径都相对于 /tmp with open(hello.txt, w) as f: f.write(hi) # 出来之后工作目录已恢复如果不用上下文管理器每次都要手动记住“进入前在哪”异常路径上分分钟忘记恢复后果就是后续所有相对路径操作全部错乱。我见过不少自动化测试脚本因为工作目录没恢复而串了环境排查起来极为痛苦。使用contextmanager时有一个容易踩的坑如果yield之后还想返回某个值去“抑制”异常靠普通return不行需要向异常处理的语义靠拢。实际上contextmanager内部帮我们实现了异常拦截如果在yield语句处捕获到异常它会以throw()的方式重新抛入生成器如果你在生成器内写的是try/except/finally结构就可以用except分支捕获再return非None值来抑制。但为了代码阅读方便绝大多数场景建议把“是否抑制”的决策留在用上下文管理器的代码处不要在这个封装里做过多的异常吞噬。4.2 ExitStack动态管理不可预知数量的资源有些场景是“运行前根本不知道要打开多少资源”比如批量处理多个文件、按配置连接多个数据库实例。手写一个while循环自己去管理每个资源的进入和退出很容易出事。ExitStack就是为了这种场景设计的。from contextlib import ExitStack filenames [a.txt, b.txt, c.txt] with ExitStack() as stack: files [stack.enter_context(open(name, r)) for name in filenames] # 此时三个文件句柄全部打开任何一步失败都会触发已打开文件的清理 contents [f.read() for f in files] # 离开with后三个文件全部自动关闭更有价值的是ExitStack允许你根据运行时的不同情况决定要不要进入某个上下文from contextlib import ExitStack def process_with_backup(primary, backupNone): with ExitStack() as stack: main_file stack.enter_context(open(primary, r)) if backup: backup_file stack.enter_context(open(backup, r)) else: backup_file None # 处理逻辑 data main_file.read() if backup_file: data backup_file.read() return data这个灵活性是普通with嵌套做不到的因为你没法在语法层面“条件性”地打开一个上下文。4.3 其他趁手工具一览contextlib模块里还有几个高频工具这里快速盘一下suppress(*exceptions)抑制指定异常。等价于自己写一个__exit__里返回issubclass(exc_type, exceptions)的上下文管理器。redirect_stdout(stream)/redirect_stderr(stream)临时把标准输出/错误重定向到指定流。写测试断言输出时非常有用。closing(obj)专为那些没有上下文管理协议但提供了close()方法的对象设计。比如urllib里面某些响应对象。nullcontext()什么都不做但兼容with的占位符常用于“可选上下文”的默认分支让代码结构一致。一个nullcontext的典型使用姿势from contextlib import nullcontext def process_data(pathNone): # 如果传了 path 就写日志文件否则就用一个什么都不做的占位 cm open(path, a) if path else nullcontext() with cm as log: # 统一在这里处理不区分有没有日志文件 pass5. 实战应用锁、事务、临时文件等场景串讲上下文管理器在真实项目中最大的价值不是“简化语法”而是让“资源生命周期”和“业务逻辑作用域”在代码结构上完全重合。这一节用几个高频实战场景把它串起来。5.1 多线程锁某些资源只允许一个线程同时访问最常见的写法是threading.Lock配合with。锁对象本身实现了上下文管理器协议进入时获取锁退出时释放锁。代码里不需要出现一个显式的acquire和release配对因为一旦在acquire和release之间抛了异常锁就可能永远得不到释放其他线程全部卡死import threading lock threading.Lock() shared_counter 0 def increment_many(): global shared_counter for _ in range(100000): with lock: shared_counter 1 threads [threading.Thread(targetincrement_many) for _ in range(4)] for t in threads: t.start() for t in threads: t.join() print(shared_counter) # 必然是 400000关键词是“必然”。如果不用lock而用裸的shared_counter 1在GIL下虽然不一定立即出错但面对高并发时结果不可预期如果手动lock.acquire()但忘了lock.release()程序会直接卡死。用with lock同时解决两个隐患。5.2 数据库事务数据库连接和事务管理是生产代码里无可争议的上下文管理器第一应用场景。以sqlite3为例连接对象支持上下文管理器协议但有一个非常容易踩的坑旧版Python里connection的__exit__只在提交或回滚上工作得很粗糙如果业务代码里写了with conn:异常时不一定保证回滚。更可靠的模式是自己封装一层或借助周边ORM。这里给出一个通用的自定义事务上下文管理器import sqlite3 from contextlib import contextmanager contextmanager def transactional_connection(db_path): conn sqlite3.connect(db_path) try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close() with transactional_connection(data.db) as conn: cursor conn.execute(UPDATE users SET level level 1 WHERE id ?, (1,))注意设计上的一个细节commit()放在try块内、yield之后意味着只有业务代码顺利执行完才会提交有异常则直接跳进except做回滚然后raise把异常丢出去保证上层能感知到“这次操作失败已回滚”。finally负责最终关闭连接。这正是数据库事务“要么全成功、要么全回滚”的语义。如果用ORM比如SQLAlchemy它的session内部已经把这套逻辑封装好了但原理依然相同with session.begin():在正常退出时commit在异常时rollback。5.3 临时文件与目录的自动清理写测试数据、导出临时报表、处理上传文件并清理这类场景标准库tempfile提供的类本身大部分实现了上下文管理器协议比如TemporaryFile、TemporaryDirectoryfrom tempfile import TemporaryDirectory with TemporaryDirectory(prefixmyapp-) as tmpdir: # tmpdir 自动生成且前缀可标识来源 import os fpath os.path.join(tmpdir, download.zip) # 假装下载并写文件 with open(fpath, wb) as f: f.write(bfake zip content) # 离开后整个临时目录连带文件自动删除如果项目里用的不是标准库封装自己写一个基于tempfile.mkdtemp()并配合__exit__里shutil.rmtree的类也是个好办法前面第2节已经演示过。工程里最怕的是临时文件只建不删尤其跑在CI上的测试任务容器磁盘被临时文件占满的例子屡见不鲜。5.4 网络连接和HTTP会话requests库的Session对象也支持上下文管理器它保证连接池在使用完毕后被清理import requests with requests.Session() as session: resp session.get(https://httpbin.org/json, timeout5) data resp.json() # 会话已关闭连接池已释放对于自建网络服务比如socket连接拿contextmanager包一层tls握手、连接保活、关闭是常见姿态。用with把每个请求的“连接生命周期”约束住既能避免长时间占用连接又方便统计链路状态。5.5 事件钩子和临时行为开关除了资源外上下文管理器还能用来管理“临时状态”。例如在调试时临时把某段代码的warning提升为errorimport warnings from contextlib import contextmanager contextmanager def strict_warnings(): with warnings.catch_warnings(): warnings.simplefilter(error, UserWarning) yield with strict_warnings(): warnings.warn(这条警告会被当成错误)这种“临时改全局状态用完必须还原”的场景如果手动管理非常容易漏掉还原步骤。上下文管理器天生适合做这件事因为__exit__保证还原逻辑永远执行。同理还可以做环境变量临时修改、日志级别临时切换、locale临时给某段逻辑改掉等。6. 进阶玩法与翻车记录几个值得警惕的细节上下文管理器用久了会发现它不只是“with语法”更是一种设计模式。但把模式用到极致前先聊聊我踩过的几个坑和实战中总结的经验。把这些细节处理好才算真正掌握这个特性。6.1 多个 with 带来的性能开销每个with进入和退出都要多一次函数调用__enter__和__exit__如果这个上下文管理器本身逻辑很薄、又处于高频循环里性能损耗是真实存在的。比如在百万级循环里用一个timer计时from contextlib import contextmanager import time contextmanager def trivial_cm(): yield start time.perf_counter() for _ in range(5_000_000): with trivial_cm(): pass print(time.perf_counter() - start)在我本机上裸循环耗时约0.3秒套上这个空上下文管理器后变成约1.5秒多出了5倍。结论不是“不要用上下文管理器”而是要注意使用边界。高频率的原子操作、硬循环内部尽量把with提到循环外面或者用普通函数调用替代资源型场景、异常安全敏感的代码该用还得用这5倍开销换来的是内存安全和代码可维护性完全值得。6.2exit里又抛异常的后果如果__exit__本身抛了新异常那原本代码块里的异常会被这个新异常替换掉整个错误信息会非常混乱。设想一下文件关闭时磁盘满了f.close()抛了异常而你正好在__exit__里做记录日志操作日志系统又出错……这时候看到的是日志系统的错误真正的文件错误反而被覆盖了。生产中遇到这种“错误信息对不上”的问题优先检查__exit__里的清理逻辑是否健全。清理代码应该只做最必要、最低风险的操作最好用try...except Exception: pass兜底。6.3 基于生成器的上下文管理器必须用 try/finally这是contextmanager装饰器最核心的纪律。很多人第一次写contextmanager def bad_cm(): print(enter) yield print(cleanup) # 如果异常发生这一行不会执行在yield后面的代码没有在try/finally里的话一旦with块中抛异常yield点会抛出异常打断生成器后面的print(cleanup)永远不执行。清理动作放错了位置等于没有清理。正确写法应该是contextmanager def good_cm(): print(enter) try: yield finally: print(cleanup) # 无论正常退出还是异常都能执行6.4 嵌套上下文管理器的异常优先级当同时使用多个with时异常的处理顺序也值得留意。从外层到内层__exit__按“最后进入先退出”的顺序被调用任何一个__exit__返回True都能阻止异常继续向外传播但这也意味着位于更外层的__exit__没有机会“看到”这个异常。设计时如果想在内层记录错误、在外层处理错误需要确保内层__exit__返回False不拦截让异常一路传到外层。6.5 使用 as 变量时注意作用域陷阱前面提过as变量不会自动销毁。这带来一个小小的反直觉场景with open(a.txt) as f: data f.read() # 此时 f 依然存在但文件已经关闭 print(f.closed) # True print(f.read()) # ValueError: I/O operation on closed file如果后续代码不小心复用了f这个变量名可能因为无意中使用了关闭后的对象而出现难以排查的异常。最好在with块外不要依赖as变量去做资源操作把读取结果提取到普通变量后立即和资源对象“断交”。6.6 七牛八凑的自定义组合最后给一个我个人很常用的组合技巧用ExitStack把多个资源与一个“整体成功/整体失败”的判断绑定。比如批量处理一批文件我希望“要么全部写入成功要么一个都别写”这时可以先把内容都处理到内存里再用ExitStack一次性打开全部目标文件统一写入。打开任何一个文件失败时前面已打开的文件由ExitStack自动关闭from contextlib import ExitStack outputs [r1.txt, r2.txt, r3.txt] contents [hello, world, python] # 假设提前算好 try: with ExitStack() as stack: files [stack.enter_context(open(name, w)) for name in outputs] for f, content in zip(files, contents): f.write(content) except OSError as e: print(写入失败已自动清理所有已打开文件, e)这类写法让“资源管理代码”看起来像“责任声明代码”一眼看过去就知道哪个资源什么时候被释放在工程维护上极其舒服。写在最后的一个实用提醒回到文章开头那个同事的问题。他后来把全项目里所有手动open/close、手动acquire/release、手动connect/disconnect的代码全部改成了with写法运行几周后再没出现过句柄泄漏或连接爆满的问题。这就是上下文管理器最大的价值它把“人一定会犯错”这件事从资源管理路径上彻底移除了。如果你正在重构老代码建议把“所有资源操作必须走with”列为一条硬性规范配合代码评审去执行。收获的不只是bug变少更是代码阅读者心智负担的大幅下降。 Python上下文管理器与with语句：原理、实战与避坑指南
从一段非常容易踩坑的代码开始说说我为什么想写这个话题。前阵子帮同事排查一个线上脚本问题症状很诡异程序跑完不报错但日志文件经常只有半截内容数据库连接偶尔提示“too many connections”。后来定位到根因其实就是文件没关闭、连接没归还资源在异常路径上泄漏了。当时他用的还是最原始的f open(...)、f.close()这种写法。我说你换成with open(...)不就好了他很困惑地问我with到底做了什么为什么以前教科书教我的手动close就不好使这个问题我觉得很有代表性。今天就用一整篇把Python上下文管理器with语句的原理、机制、实战和那些坑讲透不讲空泛理论所有结论都从可运行的代码里来。1. 资源泄漏的痛点为什么手动 close 靠不住先回到那个经典场景。刚学Python时几乎所有人都会写类似的代码f open(data.txt, r) data f.read() f.close() print(data)这段代码看起来人畜无害。但问题恰恰藏在这种“看起来没问题”的写法里。假如f.read()这行抛了异常比如文件被占用了、编码不对、磁盘IO出错了程序会直接跳过f.close()这行文件句柄就永远留在那里不释放了。一次两次没关系但脚本一旦跑在循环里、跑在长期运行的服务里文件描述符迟早会被吃光。Linux下默认文件描述符上限通常是1024超过之后open()直接报OSError: [Errno 24] Too many open files整个服务就瘫了。数据库连接、网络连接、锁资源的原理一模一样而且因为连接通常更昂贵、服务端有并发上限泄漏造成的危害更严重。你要说“我加个try...except...finally不就行了”确实可以这是标准解法try: f open(data.txt, r) data f.read() finally: f.close()但这不是Python社区推荐的做法。原因有二一是啰嗦每写一处资源操作就要套一个try...finally代码量翻倍二是容易漏人的注意力是有限的写业务逻辑时很容易忘了给某个资源加上finally兜底。Python的设计哲学里很重要的一条就是“让正确的事情更容易做”所以语言层面专门提供了with语句把“资源进来、用、确保释放”这三件事封装成一个不可分割的语法结构。这就是上下文管理器存在的根本动机它不是一个花哨的语法糖而是解决一个非常现实的工程问题资源的确定性释放。2. 协议拆解enter与exit这对双子星要理解with必须理解它背后的协议。Python里很多东西都是靠协议dunder方法运转的比如迭代器靠__iter__和__next__序列靠__len__和__getitem__上下文管理器靠的是两个特殊方法__enter__和__exit__。2.1 两者的职责与返回值__enter__(self)在进入with代码块之前被调用。它的返回值会赋给as后面的变量。需要注意的是它不是必须返回资源对象本身你可以返回任何东西甚至可以返回None。但惯例上如果这个上下文管理器代表一个资源就会把资源对象返回出去比如open()的返回值。__exit__(self, exc_type, exc_value, traceback)在代码块结束后被调用。无论代码块是正常跑完、还是中途抛了异常、还是被return/break/continue跳出它都会被调用。三个参数分别代表异常类型、异常实例、堆栈跟踪对象。如果代码块没有异常三个值都是None。关键来了__exit__的返回值是布尔类型实际上Python会对返回值做真值判断。如果返回True则表示“这个异常我已经处理掉了”with语句不会让它继续向外传播如果返回False或者None因为None会被当成假值异常会继续往外抛。2.2 最简单的自定义上下文管理器只看open()的内部实现你感受不到协议的具体工作方式自己动手写一个最直观。下面这个例子模拟一个“计时器”上下文管理器进入时记录开始时间退出时计算耗时import time class Timer: def __enter__(self): self.start time.perf_counter() return self def __exit__(self, exc_type, exc_value, traceback): self.elapsed time.perf_counter() - self.start print(f耗时: {self.elapsed:.4f} 秒) return False with Timer() as timer: # 模拟一些耗时操作 total sum(range(1000000)) print(timer.elapsed)这里有两个细节值得注意。第一as拿到的timer是Timer实例本身所以代码块执行完后还能通过timer.elapsed取到结果。第二return False是刻意写的表示“不拦截异常”。如果代码块里抛了异常这个计时器一样能正常统计耗时然后让异常继续跑出去。2.3 打开资源的标准范式再看一个资源型场景写一个上下文管理器来管理一次性临时目录。这是很多自动化测试脚本里刚需的功能import tempfile import shutil from pathlib import Path class TempDir: def __enter__(self): self.dir_path Path(tempfile.mkdtemp()) return self.dir_path def __exit__(self, exc_type, exc_value, traceback): shutil.rmtree(self.dir_path, ignore_errorsTrue) return False with TempDir() as tmp: target tmp / output.txt target.write_text(hello) print(target.exists()) # 离开with后临时目录已经被清理这类“用完必须清理清理动作绝对不能忘”的场景就是上下文管理器的主场。不需要你手动调用任何清理函数保证代码块里的逻辑无论成功还是失败shutil.rmtree一定会执行。3. with 语句底层到底做了什么执行流程全解析上一篇讲协议时可能有人已经在想__enter__的返回值给到as变量__exit__在结束时被调用但中间的异常是怎么流转的如果__enter__本身就抛了异常怎么办代码块里return了又会走到__exit__吗这一节把执行流程完整摊开讲。3.1 完整执行步骤以with EXPR as VAR:为例Python解释器实际执行的步骤可以拆成五步计算EXPR表达式得到上下文管理器对象。调用该对象的__enter__()方法把返回值赋给VAR。如果EXPR的求值本身就抛了异常后续全部不执行。执行with代码块里的完整语句体。执行完代码块后调用__exit__(None, None, None)。如果代码块内抛出了异常则带着异常信息调用__exit__(exc_type, exc_value, traceback)如果__exit__返回真值抑制异常如果返回假值异常继续抛给上层。这里有个容易误解的点__exit__并不是在“代码块正常结束”和“代码块异常”两条路径中各调用一次它只调用一次区别只在于传入的参数不同。调用方是统一的所谓“正常”与“异常”只是参数值不同。3.2 异常抑制的原理很多新手第一次看到__exit__能“吞掉异常”时会觉得不可思议。实际上这个机制非常巧妙。看下面这个场景我们希望某个操作中出现的特定异常静默处理class IgnoreValueError: def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): # 如果异常是 ValueError就返回 True 来抑制它 return exc_type is ValueError with IgnoreValueError(): number int(hello) # 触发 ValueError print(这里还能继续执行)正常情况下int(hello)会抛出ValueError导致程序崩溃但因为在__exit__里判断并返回了True这个异常就被“吸收”了。这也是标准库contextlib.suppress的实现原理后面再说。3.3enter的返回值与 as 变量的常见误区as不是必须写的。有些上下文管理器只关心进入和退出时的副作用返回值没有意义。但如果写了as你拿到什么完全取决于__enter__的设计。典型误区一以为拿到的是“被管理的资源”其实拿到的是上下文管理器自己。很多自定义类为了方便在__enter__里return self这时两者恰好相等但如果__enter__返回了别的东西比如file对象、cursor对象、或经过转换的对象那就完全不同。典型误区二as变量在离开with后仍然存在。Python用as赋值并不会像其他语言那样创造块级作用域变量在当前函数作用域里依然绑着那个对象。刚才的Timer例子就是利用了这一点。只是对象本身的资源状态可能已经“过期”了比如文件已经关闭。3.4 多个with怎么嵌套与合并同时打开两个文件做复制的常见需求两种写法等价# 写法一嵌套 with open(source.txt, r) as src: with open(target.txt, w) as dst: dst.write(src.read()) # 写法二逗号并列3.1 支持 with open(source.txt, r) as src, open(target.txt, w) as dst: dst.write(src.read())并列写法的执行顺序是从左到右先进入src再进入dst退出顺序则相反先退出dst再退出src。这符合资源释放的直觉就像先脱外套再脱帽子脱的顺序一定和穿的顺序相反。4. contextlib把上下文管理的复杂度藏起来前面写的都是基于类的实现需要定义__enter__和__exit__两个方法。但很多时候你只是想快速地把一小段逻辑变成一个上下文管理器不值得为此定义一个类。标准库contextlib就是专门解决这个问题的模块。4.1 contextmanager 装饰器用生成器伪装上下文管理器contextlib.contextmanager是一个装饰器可以把一个生成器函数变成上下文管理器。它内部替你做了一大堆状态机和异常处理的脏活让使用者只需关心yield前后的两段逻辑。基本形态如下from contextlib import contextmanager import time contextmanager def timer(): start time.perf_counter() try: yield # yield 之前的代码相当于 __enter__ finally: elapsed time.perf_counter() - start print(f耗时: {elapsed:.4f} 秒)重点在于这几点。第一yield之前的所有代码会在进入with时执行相当于__enter__。第二yield的产出值会赋给as后面的变量。第三**yield之后确切地说是finally块里的代码在退出时执行**相当于__exit__。第三点尤其关键yield后面的代码必须以try...finally包裹这样无论with块里发生什么正常结束也好、异常也好finally里的清理逻辑保证执行。再看一个生产实践中很常用的示例切换工作目录很多人写脚本时会有这个需求contextmanager def working_directory(path): import os origin os.getcwd() os.chdir(path) try: yield finally: os.chdir(origin) with working_directory(/tmp): # 在这里所有相对路径都相对于 /tmp with open(hello.txt, w) as f: f.write(hi) # 出来之后工作目录已恢复如果不用上下文管理器每次都要手动记住“进入前在哪”异常路径上分分钟忘记恢复后果就是后续所有相对路径操作全部错乱。我见过不少自动化测试脚本因为工作目录没恢复而串了环境排查起来极为痛苦。使用contextmanager时有一个容易踩的坑如果yield之后还想返回某个值去“抑制”异常靠普通return不行需要向异常处理的语义靠拢。实际上contextmanager内部帮我们实现了异常拦截如果在yield语句处捕获到异常它会以throw()的方式重新抛入生成器如果你在生成器内写的是try/except/finally结构就可以用except分支捕获再return非None值来抑制。但为了代码阅读方便绝大多数场景建议把“是否抑制”的决策留在用上下文管理器的代码处不要在这个封装里做过多的异常吞噬。4.2 ExitStack动态管理不可预知数量的资源有些场景是“运行前根本不知道要打开多少资源”比如批量处理多个文件、按配置连接多个数据库实例。手写一个while循环自己去管理每个资源的进入和退出很容易出事。ExitStack就是为了这种场景设计的。from contextlib import ExitStack filenames [a.txt, b.txt, c.txt] with ExitStack() as stack: files [stack.enter_context(open(name, r)) for name in filenames] # 此时三个文件句柄全部打开任何一步失败都会触发已打开文件的清理 contents [f.read() for f in files] # 离开with后三个文件全部自动关闭更有价值的是ExitStack允许你根据运行时的不同情况决定要不要进入某个上下文from contextlib import ExitStack def process_with_backup(primary, backupNone): with ExitStack() as stack: main_file stack.enter_context(open(primary, r)) if backup: backup_file stack.enter_context(open(backup, r)) else: backup_file None # 处理逻辑 data main_file.read() if backup_file: data backup_file.read() return data这个灵活性是普通with嵌套做不到的因为你没法在语法层面“条件性”地打开一个上下文。4.3 其他趁手工具一览contextlib模块里还有几个高频工具这里快速盘一下suppress(*exceptions)抑制指定异常。等价于自己写一个__exit__里返回issubclass(exc_type, exceptions)的上下文管理器。redirect_stdout(stream)/redirect_stderr(stream)临时把标准输出/错误重定向到指定流。写测试断言输出时非常有用。closing(obj)专为那些没有上下文管理协议但提供了close()方法的对象设计。比如urllib里面某些响应对象。nullcontext()什么都不做但兼容with的占位符常用于“可选上下文”的默认分支让代码结构一致。一个nullcontext的典型使用姿势from contextlib import nullcontext def process_data(pathNone): # 如果传了 path 就写日志文件否则就用一个什么都不做的占位 cm open(path, a) if path else nullcontext() with cm as log: # 统一在这里处理不区分有没有日志文件 pass5. 实战应用锁、事务、临时文件等场景串讲上下文管理器在真实项目中最大的价值不是“简化语法”而是让“资源生命周期”和“业务逻辑作用域”在代码结构上完全重合。这一节用几个高频实战场景把它串起来。5.1 多线程锁某些资源只允许一个线程同时访问最常见的写法是threading.Lock配合with。锁对象本身实现了上下文管理器协议进入时获取锁退出时释放锁。代码里不需要出现一个显式的acquire和release配对因为一旦在acquire和release之间抛了异常锁就可能永远得不到释放其他线程全部卡死import threading lock threading.Lock() shared_counter 0 def increment_many(): global shared_counter for _ in range(100000): with lock: shared_counter 1 threads [threading.Thread(targetincrement_many) for _ in range(4)] for t in threads: t.start() for t in threads: t.join() print(shared_counter) # 必然是 400000关键词是“必然”。如果不用lock而用裸的shared_counter 1在GIL下虽然不一定立即出错但面对高并发时结果不可预期如果手动lock.acquire()但忘了lock.release()程序会直接卡死。用with lock同时解决两个隐患。5.2 数据库事务数据库连接和事务管理是生产代码里无可争议的上下文管理器第一应用场景。以sqlite3为例连接对象支持上下文管理器协议但有一个非常容易踩的坑旧版Python里connection的__exit__只在提交或回滚上工作得很粗糙如果业务代码里写了with conn:异常时不一定保证回滚。更可靠的模式是自己封装一层或借助周边ORM。这里给出一个通用的自定义事务上下文管理器import sqlite3 from contextlib import contextmanager contextmanager def transactional_connection(db_path): conn sqlite3.connect(db_path) try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close() with transactional_connection(data.db) as conn: cursor conn.execute(UPDATE users SET level level 1 WHERE id ?, (1,))注意设计上的一个细节commit()放在try块内、yield之后意味着只有业务代码顺利执行完才会提交有异常则直接跳进except做回滚然后raise把异常丢出去保证上层能感知到“这次操作失败已回滚”。finally负责最终关闭连接。这正是数据库事务“要么全成功、要么全回滚”的语义。如果用ORM比如SQLAlchemy它的session内部已经把这套逻辑封装好了但原理依然相同with session.begin():在正常退出时commit在异常时rollback。5.3 临时文件与目录的自动清理写测试数据、导出临时报表、处理上传文件并清理这类场景标准库tempfile提供的类本身大部分实现了上下文管理器协议比如TemporaryFile、TemporaryDirectoryfrom tempfile import TemporaryDirectory with TemporaryDirectory(prefixmyapp-) as tmpdir: # tmpdir 自动生成且前缀可标识来源 import os fpath os.path.join(tmpdir, download.zip) # 假装下载并写文件 with open(fpath, wb) as f: f.write(bfake zip content) # 离开后整个临时目录连带文件自动删除如果项目里用的不是标准库封装自己写一个基于tempfile.mkdtemp()并配合__exit__里shutil.rmtree的类也是个好办法前面第2节已经演示过。工程里最怕的是临时文件只建不删尤其跑在CI上的测试任务容器磁盘被临时文件占满的例子屡见不鲜。5.4 网络连接和HTTP会话requests库的Session对象也支持上下文管理器它保证连接池在使用完毕后被清理import requests with requests.Session() as session: resp session.get(https://httpbin.org/json, timeout5) data resp.json() # 会话已关闭连接池已释放对于自建网络服务比如socket连接拿contextmanager包一层tls握手、连接保活、关闭是常见姿态。用with把每个请求的“连接生命周期”约束住既能避免长时间占用连接又方便统计链路状态。5.5 事件钩子和临时行为开关除了资源外上下文管理器还能用来管理“临时状态”。例如在调试时临时把某段代码的warning提升为errorimport warnings from contextlib import contextmanager contextmanager def strict_warnings(): with warnings.catch_warnings(): warnings.simplefilter(error, UserWarning) yield with strict_warnings(): warnings.warn(这条警告会被当成错误)这种“临时改全局状态用完必须还原”的场景如果手动管理非常容易漏掉还原步骤。上下文管理器天生适合做这件事因为__exit__保证还原逻辑永远执行。同理还可以做环境变量临时修改、日志级别临时切换、locale临时给某段逻辑改掉等。6. 进阶玩法与翻车记录几个值得警惕的细节上下文管理器用久了会发现它不只是“with语法”更是一种设计模式。但把模式用到极致前先聊聊我踩过的几个坑和实战中总结的经验。把这些细节处理好才算真正掌握这个特性。6.1 多个 with 带来的性能开销每个with进入和退出都要多一次函数调用__enter__和__exit__如果这个上下文管理器本身逻辑很薄、又处于高频循环里性能损耗是真实存在的。比如在百万级循环里用一个timer计时from contextlib import contextmanager import time contextmanager def trivial_cm(): yield start time.perf_counter() for _ in range(5_000_000): with trivial_cm(): pass print(time.perf_counter() - start)在我本机上裸循环耗时约0.3秒套上这个空上下文管理器后变成约1.5秒多出了5倍。结论不是“不要用上下文管理器”而是要注意使用边界。高频率的原子操作、硬循环内部尽量把with提到循环外面或者用普通函数调用替代资源型场景、异常安全敏感的代码该用还得用这5倍开销换来的是内存安全和代码可维护性完全值得。6.2exit里又抛异常的后果如果__exit__本身抛了新异常那原本代码块里的异常会被这个新异常替换掉整个错误信息会非常混乱。设想一下文件关闭时磁盘满了f.close()抛了异常而你正好在__exit__里做记录日志操作日志系统又出错……这时候看到的是日志系统的错误真正的文件错误反而被覆盖了。生产中遇到这种“错误信息对不上”的问题优先检查__exit__里的清理逻辑是否健全。清理代码应该只做最必要、最低风险的操作最好用try...except Exception: pass兜底。6.3 基于生成器的上下文管理器必须用 try/finally这是contextmanager装饰器最核心的纪律。很多人第一次写contextmanager def bad_cm(): print(enter) yield print(cleanup) # 如果异常发生这一行不会执行在yield后面的代码没有在try/finally里的话一旦with块中抛异常yield点会抛出异常打断生成器后面的print(cleanup)永远不执行。清理动作放错了位置等于没有清理。正确写法应该是contextmanager def good_cm(): print(enter) try: yield finally: print(cleanup) # 无论正常退出还是异常都能执行6.4 嵌套上下文管理器的异常优先级当同时使用多个with时异常的处理顺序也值得留意。从外层到内层__exit__按“最后进入先退出”的顺序被调用任何一个__exit__返回True都能阻止异常继续向外传播但这也意味着位于更外层的__exit__没有机会“看到”这个异常。设计时如果想在内层记录错误、在外层处理错误需要确保内层__exit__返回False不拦截让异常一路传到外层。6.5 使用 as 变量时注意作用域陷阱前面提过as变量不会自动销毁。这带来一个小小的反直觉场景with open(a.txt) as f: data f.read() # 此时 f 依然存在但文件已经关闭 print(f.closed) # True print(f.read()) # ValueError: I/O operation on closed file如果后续代码不小心复用了f这个变量名可能因为无意中使用了关闭后的对象而出现难以排查的异常。最好在with块外不要依赖as变量去做资源操作把读取结果提取到普通变量后立即和资源对象“断交”。6.6 七牛八凑的自定义组合最后给一个我个人很常用的组合技巧用ExitStack把多个资源与一个“整体成功/整体失败”的判断绑定。比如批量处理一批文件我希望“要么全部写入成功要么一个都别写”这时可以先把内容都处理到内存里再用ExitStack一次性打开全部目标文件统一写入。打开任何一个文件失败时前面已打开的文件由ExitStack自动关闭from contextlib import ExitStack outputs [r1.txt, r2.txt, r3.txt] contents [hello, world, python] # 假设提前算好 try: with ExitStack() as stack: files [stack.enter_context(open(name, w)) for name in outputs] for f, content in zip(files, contents): f.write(content) except OSError as e: print(写入失败已自动清理所有已打开文件, e)这类写法让“资源管理代码”看起来像“责任声明代码”一眼看过去就知道哪个资源什么时候被释放在工程维护上极其舒服。写在最后的一个实用提醒回到文章开头那个同事的问题。他后来把全项目里所有手动open/close、手动acquire/release、手动connect/disconnect的代码全部改成了with写法运行几周后再没出现过句柄泄漏或连接爆满的问题。这就是上下文管理器最大的价值它把“人一定会犯错”这件事从资源管理路径上彻底移除了。如果你正在重构老代码建议把“所有资源操作必须走with”列为一条硬性规范配合代码评审去执行。收获的不只是bug变少更是代码阅读者心智负担的大幅下降。