拓冰网站定制 
1. 为什么需要上下文管理器手动管理资源的滋味不好受1.1 从一段最朴素的代码说起先看一段几乎所有Python初学者都写过的代码file open(data.txt, r) data file.read() file.close()这段代码的问题在哪儿如果file.read()抛了异常close()根本不会执行文件句柄就一直开着。Python的垃圾回收虽然最后会把对象回收掉但回收时机不可控尤其在高并发的服务里文件描述符是有限资源一旦耗尽后续所有的open()都会报OSError: [Errno 24] Too many open files整个进程基本就废了。于是你可能会改成这样file open(data.txt, r) try: data file.read() finally: file.close()这样确实安全了finally保证了无论中间发生什么文件都会被关闭。但问题来了——这样的代码写多了之后try/finally的结构噪音特别大而且一旦涉及多个资源嵌套管理代码层级会迅速膨胀可读性直线下降。1.2 with语句到底解决了一个什么问题with语句解决的核心问题就是把资源获取和资源释放这两件事绑定在一起并且保证释放逻辑一定会执行。它的设计目标很明确让你写出既安全又简洁的代码不用每次手动记住用完要释放。拿最经典的文件操作来说with open(data.txt, r) as f: data f.read()两行半代码替代了上面五行的try/finally结构而且无论是正常执行还是抛出异常文件都会被正确关闭。这不是魔法而是open()返回的文件对象实现了上下文管理协议——也就是定义了__enter__和__exit__这两个方法。写到这里很多人可能会觉得我平时用with只是用来打开文件、开数据库连接好像也没必要深入理解原理吧。这种想法我建议尽早扔掉。上下文管理器远不止资源释放这一个用途光是contextlib标准库里面就有大量能提升代码质量的好东西更不用说很多第三方库比如requests、threading里的锁对象、unittest里的断言上下文都深度依赖这个协议。如果说装饰器是Python函数级别的语法糖那上下文管理器就是代码块级别的语法糖。理解它的原理你的Python水平会明显往上走一个台阶。2. with语句的执行序__enter__和__exit__到底做了什么2.1 协议定义与执行流程拆解上下文管理协议只有两个方法class MyContext: def __enter__(self): # 进入with代码块前调用 # 返回值会绑定给as后面的变量 return something def __exit__(self, exc_type, exc_val, exc_tb): # 离开with代码块时调用无论是否抛异常 # 返回True表示异常已被处理返回False/None表示继续传播异常 ...当你在代码里写with MyContext() as obj:时Python解释器实际做的是调用MyContext()得到上下文对象。执行ctx.__enter__()把返回值赋给obj。运行with代码块主体。代码块结束后调用ctx.__exit__(exc_type, exc_val, exc_tb)。要注意的是如果__enter__本身抛了异常__exit__不会执行因为上下文对象根本还没有进入已启动状态。如果__exit__自己抛异常那这个新异常会替代原来的异常继续往外抛。这里有一个大家容易忽略的细节as后面的变量绑定的是__enter__的返回值而不是上下文对象本身。对于文件对象来说open()返回的文件对象同时实现了__enter__和__exit__并且__enter__返回self所以你拿到的是同一个文件对象。但有些类设计上会让__enter__返回一个跟自身不同的对象比如数据库连接用with语句时你拿到的可能是游标或事务对象而不是连接本身。所以写代码前最好确认一下__enter__到底返回了什么。2.2 __exit__的返回值为什么是True和False__exit__的返回值是整个协议里最容易被误解的部分。它的规则只有一条返回True时with代码块内抛出的异常会被吞掉不再往外传播返回False、None或任何假值时异常会照常向上抛。看看这段代码class IgnoreZeroDivision: def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): return True with IgnoreZeroDivision(): 1 / 0 # 不会抛异常被静默吞掉了 print(程序继续执行)运行结果会直接打印程序继续执行——因为__exit__返回了True解释器认为异常已经处理完毕于是不再传播。这个设计本意是给那些我已经在__exit__里处理掉异常了不需要再往上抛的场景用的。但实际开发中我见过很多人盲目返回True结果把异常全部吞了线上排查问题时日志一片空白连错误都不知道从哪冒出来的。除非你非常清楚自己在干什么否则__exit__应该返回False或None。大多数情况下处理异常的逻辑应该写在with代码块内或者交给外层去处理。__exit__里更适合做的是清理工作比如提交事务、关闭连接、释放锁而不是把异常拦下来消化掉。2.3 多个上下文管理器逗号连写的执行顺序Python的with语句支持一次管理多个上下文管理器with open(a.txt) as f1, open(b.txt) as f2: data1 f1.read() data2 f2.read()这个写法在Python 2.7和3.1之后都支持。但有几个执行顺序的细节值得记一下f1和f2的初始化按从左到右的顺序执行。f1.__enter__()先执行f2.__enter__()后执行。with代码块退出时f2.__exit__()先执行f1.__exit__()后执行——退出顺序和进入顺序相反这其实和嵌套with的语义是一致的。所以如果是资源之间有依赖关系比如先开数据库连接、再开游标退出时游标先关连接后关正好符合直觉。不过这种逗号连写格式在行数变长之后不太好看我更推荐用嵌套写法或者用contextlib.ExitStack来管理——这个后面专门讲。3. contextlib大礼包不写类也能定义上下文管理器3.1 contextmanager装饰器的原理与模板每次实现上下文管理器都要硬写一个类定义__enter__和__exit__时间长了你会觉得这种样板代码很烦人。好在标准库contextlib提供了一个叫contextmanager的装饰器可以直接用生成器函数来定义上下文管理器from contextlib import contextmanager contextmanager def managed_file(path, moder): f open(path, mode) try: yield f finally: f.close() with managed_file(data.txt) as f: data f.read()这个写法的执行逻辑是这样的函数执行到yield之前的部分相当于__enter__里面的逻辑。yield后面的值会绑定给as后面的变量。yield之后的代码相当于__exit__里面的逻辑不管with代码块是否抛异常finally都能保证执行。关键一点是如果with代码块内抛了异常这个异常会在yield处重新抛出所以你可以用try/except捕获它做定制化处理contextmanager def suppress_exception(exc_type): try: yield except exc_type: print(f捕获到异常{exc_type.__name__}但选择忽略) with suppress_exception(ValueError): int(不是数字)这个写法比写一个完整的类简洁得多也是我个人日常用得最多的方式。需要提醒的是yield前如果抛了异常finally依然会执行所以把资源释放写在finally里面是最稳妥的。3.2 closing、suppress、redirect_stdout等现成工具contextlib里还有几个不需要自己动手就能直接用的上下文管理器。contextlib.closing用于那些实现了close()方法但没实现上下文协议的对象from contextlib import closing from urllib.request import urlopen with closing(urlopen(https://example.com)) as page: data page.read()没有closing的话你得自己写try/finally去调close()有它之后代码会干净很多。contextlib.suppress可以按异常类型静默特定异常比自己在except里写pass更明确from contextlib import suppress import os with suppress(FileNotFoundError): os.remove(temp_cache.json)这个用法在处理删除一个可能不存在的文件关闭一个可能已经断开的连接这类场景时特别好用代码语义也比try/except pass更清晰。contextlib.redirect_stdout和redirect_stderr可以把print输出重定向到文件或缓冲区适合测试或者封装日志from contextlib import redirect_stdout import io buffer io.StringIO() with redirect_stdout(buffer): print(这行不会打印到终端而是写到buffer里)3.3 ExitStack动态管理不确定数量的资源ExitStack是contextlib里最强大的工具之一它让你在运行时动态地、按需地注册多个清理回调退出时按逆序统一执行。看这个例子from contextlib import ExitStack def open_many_files(paths): stack ExitStack() files [] try: for p in paths: f open(p) stack.callback(f.close) # 或者 stack.push(f.close) files.append(f) return files, stack except Exception: stack.close() # 出问题就立刻清理已打开的文件 raise调用方用完以后只要执行stack.close()之前注册的所有close回调就会逆序执行。你也可以用stack.enter_context()把普通上下文管理器交给ExitStack托管from contextlib import ExitStack with ExitStack() as stack: f1 stack.enter_context(open(a.txt)) f2 stack.enter_context(open(b.txt)) session stack.enter_context(create_db_session()) # 退出with块时f2、f1、session会按逆序自动清理这种写法本质上是动态版的多层with嵌套非常适合那些资源数量不确定、需要条件式注册清理逻辑的场景。4. 实战自己动手实现几个常用上下文管理器4.1 带超时的代码块计时器很多性能分析工具都能做代码计时但用上下文管理器自己写一个更灵活可以精确控制哪个代码块要计、要不要打印日志、超时没超时。import time from contextlib import contextmanager contextmanager def timed_block(name, timeout1.0): start time.perf_counter() try: yield finally: elapsed time.perf_counter() - start if elapsed timeout: print(f[警告] {name} 执行耗时 {elapsed:.3f}s超过阈值 {timeout}s) else: print(f[信息] {name} 执行耗时 {elapsed:.3f}s) with timed_block(数据归一化, timeout0.5): # 模拟耗时操作 time.sleep(0.6)这个上下文管理器在生产环境做性能埋点时很实用。你不需要侵入业务代码结构只需要在被观测的代码块外面套一层with就行比手动记录time.time()前后差值要整洁得多。4.2 临时目录自动清理器写测试用例或数据处理脚本时经常需要创建临时目录放中间文件跑完以后想自动清理。下面的上下文管理器专门干这个import os import shutil import tempfile from contextlib import contextmanager contextmanager def temporary_dir(prefixtmp_): path tempfile.mkdtemp(prefixprefix) try: yield path finally: if os.path.isdir(path): shutil.rmtree(path) with temporary_dir() as work_dir: # 在work_dir里随便写文件 with open(os.path.join(work_dir, output.csv), w) as f: f.write(a,b,c\n1,2,3) # with块结束目录自动被删除不留垃圾这里用了tempfile.mkdtemp创建真实物理目录用shutil.rmtree递归删除。如果没有这个上下文管理器你每次都需要在测试结束后手动清理临时目录一旦中间断言失败抛异常临时目录就变成磁盘上的残留垃圾。有了它不管测试通过还是失败目录都会被清干净。4.3 可重试的数据库连接数据库操作是上下文管理器最典型的应用场景之一核心逻辑在于连接要关、事务要提交或回滚。实际写业务代码时我通常这样组织from contextlib import contextmanager contextmanager def db_transaction(connection): 将连接传入自动处理提交/回滚/关闭 try: yield connection connection.commit() except Exception: connection.rollback() raise finally: connection.close() # 使用示例 with db_transaction(conn) as cursor: cursor.execute(UPDATE users SET balance balance - 100 WHERE id 1) cursor.execute(UPDATE users SET balance balance 100 WHERE id 2)这里的关键在于commit的位置——它放在yield后面的try主体里也就是只有当with代码块正常执行完才会提交事务如果代码块内抛了异常rollback回滚后重新把异常抛出去避免脏数据落库。更进一步数据库连接经常因为网络抖动而断开你还可以在上下文管理器里写重试逻辑import time from contextlib import contextmanager contextmanager def db_retry_connect(connection_factory, retries3, delay0.5): last_exc None for attempt in range(retries): conn connection_factory() try: yield conn return except ConnectionError as exc: last_exc exc conn.close() time.sleep(delay * (attempt 1)) except Exception: conn.close() raise raise last_exc这个例子展示了上下文管理器不只是资源释放工具它还能承载策略逻辑比如重试、超时、降级。调用方写业务代码时完全不需要关心重试次数和等待时间只需要把真正的业务逻辑放在with块里。4.4 自定义锁与条件变量的管理threading.Lock本身实现了上下文管理协议可以直接用with lock:来加锁/释放但有时你需要对锁行为做定制。比如给锁加一个超时控制避免线程死等import threading from contextlib import contextmanager contextmanager def timed_lock(lock, timeout5.0): acquired lock.acquire(timeouttimeout) if not acquired: raise TimeoutError(f在 {timeout} 秒内未能获取锁) try: yield lock finally: lock.release() lock threading.Lock() def worker(): try: with timed_lock(lock, timeout3.0) as lk: # 临界区代码 print(拿到锁开始工作) except TimeoutError: print(锁等待超时放弃) threads [threading.Thread(targetworker) for _ in range(5)] for t in threads: t.start() for t in threads: t.join()它的底层原理并不复杂acquire(timeout)返回布尔值拿不到锁就主动抛出超时异常而不是让线程无限期等下去。在微服务架构里等锁超时比无限阻塞要安全得多因为超时可以被监控、被快速失败。5. 踩坑记录上下文管理器实践中的几个隐蔽问题5.1 异常被静默吞掉的后果前面讲过__exit__返回True会吞异常但这里再展开说一个真实场景。我曾经在一个内部工具里见过这样的代码from contextlib import contextmanager contextmanager def api_session(): session create_session() try: yield session except Exception: ... # 这里打印了日志但忘了重新raise问题在于except捕获异常后既没有raise也没有在finally里做额外处理contextmanager会把函数内吞掉的异常视为已处理调用方的try/except就永远接不到错误。业务代码里一旦发生这种情况API调用失败但调用方毫无感知数据不同步的问题往往要延迟很久才暴露。排查这类问题的最快方法是在__exit__或except里打日志时顺便把exc_info完整输出并确认是否重新抛出。一个更稳妥的习惯是上下文管理器里捕获异常后默认加上raise除非你有充分理由确定不要传播。5.2 生成器上下文管理器中的yield陷阱contextmanager装饰的生成器有一个容易踩的坑如果函数在yield之前就抛了异常with块根本不会执行但你可能以为资源已经被正确清理了。看这个例子contextmanager def risky(): resource acquire_resource() # 这里抛异常 try: yield resource finally: release_resource() with risky() as res: ...如果acquire_resource()抛异常risky()函数根本进不到try所以不会执行finally。这个行为其实是对的——都没成功获取资源当然没有释放可言。但很多人写代码时会把获取资源和使用资源的边界搞混把本应放在yield之后的安全清理逻辑错误地放在了try之外导致异常发生时资源泄漏。我的建议是所有需要释放的资源获取动作都放进try里面。也就是先resource None然后用try/finally包住yield这样即使获取过程中出错finally里也能安全地判断并清理。5.3 装饰器和上下文管理器组合的顺序问题Python里装饰器和上下文管理器经常搭配使用但顺序错了会让代码行为和预期完全相反。比如你在FastAPI或Flask里可能见过这样的写法这里用Flask风格示意from functools import wraps def with_transaction(): def decorator(func): wraps(func) def wrapper(*args, **kwargs): with db_transaction(conn): return func(*args, **kwargs) return wrapper return decorator with_transaction() def update_balance(user_id, amount): ...这种组合本身没问题但如果你在装饰器内部用了contextmanager又同时在函数外部也用with就要特别注意执行顺序。比如下面这段代码的运行顺序是外层with先进入装饰器包装的函数体内部再进入第二个with退出时内层先退、外层后退。这个顺序如果搞反事务的提交时机就错了。**我的经验法则是先理解一层上下文管理器的生命周期再叠加多层涉及多层时从最外层往里推进入顺序从最内层往外推退出顺序。**这样无论组合多复杂都能理清逻辑。5.4 ExitStack使用时的常见误操作ExitStack确实强大但刚上手时容易出问题。最经典的一个误操作是把return或yield放在ExitStack的with块外导致资源提前释放。看这段有问题的代码from contextlib import ExitStack def get_files(): stack ExitStack() with stack: f1 stack.enter_context(open(a.txt)) f2 stack.enter_context(open(b.txt)) return f1, f2 # 这里with块已退出文件都关了调用方拿到的f1、f2实际上是已经关闭的文件对象读数据必然报错。正确做法是让stack的生命周期覆盖整个文件使用期通常让get_files也变成一个生成器或上下文管理器from contextlib import ExitStack, contextmanager contextmanager def open_many(paths): with ExitStack() as stack: files [stack.enter_context(open(p)) for p in paths] yield files另外一个容易忽略的点是pop_all()方法可以把当前已注册的回调整体搬走这在将资源所有权从一个上下文管理器转移给另一个时很有用但用不好会让清理逻辑彻底失序。新手阶段尽量别折腾pop_all等把ExitStack的基本用法吃透了再进阶。6. 把上下文管理器用在业务代码里的几个设计建议6.1 用上下文管理器封装事务边界而不是散落try/except业务系统里最常见的反模式是每个函数里都写try/except做事务提交回滚代码重复不说一旦新增一个数据库操作经常忘记补事务边界或者提交时机搞错。用上下文管理器封装事务边界之后业务函数的职责就变得很纯粹——只关心数据怎么处理不用关心事务怎么开、怎么提交、怎么回滚。这个思维方式的转变对代码可维护性的提升非常明显。6.2 把耗时操作的埋点做成可复用的上下文管理器除了性能分析上下文管理器还特别适合做日志链路追踪。比如你可以写一个trace_span(name)的上下文管理器import uuid from contextlib import contextmanager contextmanager def trace_span(name): request_id uuid.uuid4().hex[:8] print(f[trace] 开始 {name}request_id{request_id}) try: yield request_id finally: print(f[trace] 结束 {name}request_id{request_id})这个设计思路完全可以接入现有的日志系统或链路追踪平台而且不需要改动业务函数的签名对老代码的侵入性极低。6.3 小心上下文管理器对象与上下文管理器本身的生命周期这是最后一个我特别想强调的点。定义上下文管理器之后要分清两种使用方式# 方式一把上下文管理器对象赋给变量在外部管理 ctx managed_resource() with ctx: pass # 方式二直接使用 with managed_resource(): pass方式一里有隐藏风险ctx对象的__enter__和__exit__可以多次调用吗可以重入吗如果实现不支持重复进入那么第二次with ctx就会出问题。多数自实现上下文管理器不处理重入语义所以除非明确知道它是可重入的否则不要复用同一个上下文管理器对象。这个坑在写测试时会特别常见——有人为了减少重复代码在setUp里创建一个上下文管理器对象然后在多个测试方法里复用结果第二个测试就会出诡异的问题。正确的做法是让__enter__每次都返回全新的状态或者干脆在with语句内部创建新的上下文管理器实例。7. 我的实测体会上下文管理器是我在使用Python时受益最大的语言特性之一它看似简单背后却凝聚了资源安全和代码结构两个维度的思考。写这个主题的文章我最大的体会是普通开发者用with语句是在使用一个语法进阶开发者用with语句是在设计一段代码的执行边界。从文件操作到数据库事务从锁管理到临时资源生命周期上下文管理器把进入和退出这两个隐形的边界显性化了。它让我在写业务代码时更敢放心地分配资源因为知道释放逻辑一定会被可靠执行哪怕发生异常。最后分享两个我自己的习惯一是能用contextmanager装饰器解决的就尽量不写原生类代码量少、读起来直观二是凡是涉及数据库连接、文件句柄、网络连接这些重量级资源的上下文管理器__exit__里一定写上完整的状态日志并且严格保持raise继续传播绝不静默吞异常。如果你之前只用过with open建议找个休息日下午把contextlib的文档从头到尾翻一遍再顺手把这些实战例子在自己环境里跑一遍。相信我下次你写代码时看待with的眼神会完全不一样。 Python上下文管理器实战：with语句、contextlib与资源安全释放
1. 为什么需要上下文管理器手动管理资源的滋味不好受1.1 从一段最朴素的代码说起先看一段几乎所有Python初学者都写过的代码file open(data.txt, r) data file.read() file.close()这段代码的问题在哪儿如果file.read()抛了异常close()根本不会执行文件句柄就一直开着。Python的垃圾回收虽然最后会把对象回收掉但回收时机不可控尤其在高并发的服务里文件描述符是有限资源一旦耗尽后续所有的open()都会报OSError: [Errno 24] Too many open files整个进程基本就废了。于是你可能会改成这样file open(data.txt, r) try: data file.read() finally: file.close()这样确实安全了finally保证了无论中间发生什么文件都会被关闭。但问题来了——这样的代码写多了之后try/finally的结构噪音特别大而且一旦涉及多个资源嵌套管理代码层级会迅速膨胀可读性直线下降。1.2 with语句到底解决了一个什么问题with语句解决的核心问题就是把资源获取和资源释放这两件事绑定在一起并且保证释放逻辑一定会执行。它的设计目标很明确让你写出既安全又简洁的代码不用每次手动记住用完要释放。拿最经典的文件操作来说with open(data.txt, r) as f: data f.read()两行半代码替代了上面五行的try/finally结构而且无论是正常执行还是抛出异常文件都会被正确关闭。这不是魔法而是open()返回的文件对象实现了上下文管理协议——也就是定义了__enter__和__exit__这两个方法。写到这里很多人可能会觉得我平时用with只是用来打开文件、开数据库连接好像也没必要深入理解原理吧。这种想法我建议尽早扔掉。上下文管理器远不止资源释放这一个用途光是contextlib标准库里面就有大量能提升代码质量的好东西更不用说很多第三方库比如requests、threading里的锁对象、unittest里的断言上下文都深度依赖这个协议。如果说装饰器是Python函数级别的语法糖那上下文管理器就是代码块级别的语法糖。理解它的原理你的Python水平会明显往上走一个台阶。2. with语句的执行序__enter__和__exit__到底做了什么2.1 协议定义与执行流程拆解上下文管理协议只有两个方法class MyContext: def __enter__(self): # 进入with代码块前调用 # 返回值会绑定给as后面的变量 return something def __exit__(self, exc_type, exc_val, exc_tb): # 离开with代码块时调用无论是否抛异常 # 返回True表示异常已被处理返回False/None表示继续传播异常 ...当你在代码里写with MyContext() as obj:时Python解释器实际做的是调用MyContext()得到上下文对象。执行ctx.__enter__()把返回值赋给obj。运行with代码块主体。代码块结束后调用ctx.__exit__(exc_type, exc_val, exc_tb)。要注意的是如果__enter__本身抛了异常__exit__不会执行因为上下文对象根本还没有进入已启动状态。如果__exit__自己抛异常那这个新异常会替代原来的异常继续往外抛。这里有一个大家容易忽略的细节as后面的变量绑定的是__enter__的返回值而不是上下文对象本身。对于文件对象来说open()返回的文件对象同时实现了__enter__和__exit__并且__enter__返回self所以你拿到的是同一个文件对象。但有些类设计上会让__enter__返回一个跟自身不同的对象比如数据库连接用with语句时你拿到的可能是游标或事务对象而不是连接本身。所以写代码前最好确认一下__enter__到底返回了什么。2.2 __exit__的返回值为什么是True和False__exit__的返回值是整个协议里最容易被误解的部分。它的规则只有一条返回True时with代码块内抛出的异常会被吞掉不再往外传播返回False、None或任何假值时异常会照常向上抛。看看这段代码class IgnoreZeroDivision: def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): return True with IgnoreZeroDivision(): 1 / 0 # 不会抛异常被静默吞掉了 print(程序继续执行)运行结果会直接打印程序继续执行——因为__exit__返回了True解释器认为异常已经处理完毕于是不再传播。这个设计本意是给那些我已经在__exit__里处理掉异常了不需要再往上抛的场景用的。但实际开发中我见过很多人盲目返回True结果把异常全部吞了线上排查问题时日志一片空白连错误都不知道从哪冒出来的。除非你非常清楚自己在干什么否则__exit__应该返回False或None。大多数情况下处理异常的逻辑应该写在with代码块内或者交给外层去处理。__exit__里更适合做的是清理工作比如提交事务、关闭连接、释放锁而不是把异常拦下来消化掉。2.3 多个上下文管理器逗号连写的执行顺序Python的with语句支持一次管理多个上下文管理器with open(a.txt) as f1, open(b.txt) as f2: data1 f1.read() data2 f2.read()这个写法在Python 2.7和3.1之后都支持。但有几个执行顺序的细节值得记一下f1和f2的初始化按从左到右的顺序执行。f1.__enter__()先执行f2.__enter__()后执行。with代码块退出时f2.__exit__()先执行f1.__exit__()后执行——退出顺序和进入顺序相反这其实和嵌套with的语义是一致的。所以如果是资源之间有依赖关系比如先开数据库连接、再开游标退出时游标先关连接后关正好符合直觉。不过这种逗号连写格式在行数变长之后不太好看我更推荐用嵌套写法或者用contextlib.ExitStack来管理——这个后面专门讲。3. contextlib大礼包不写类也能定义上下文管理器3.1 contextmanager装饰器的原理与模板每次实现上下文管理器都要硬写一个类定义__enter__和__exit__时间长了你会觉得这种样板代码很烦人。好在标准库contextlib提供了一个叫contextmanager的装饰器可以直接用生成器函数来定义上下文管理器from contextlib import contextmanager contextmanager def managed_file(path, moder): f open(path, mode) try: yield f finally: f.close() with managed_file(data.txt) as f: data f.read()这个写法的执行逻辑是这样的函数执行到yield之前的部分相当于__enter__里面的逻辑。yield后面的值会绑定给as后面的变量。yield之后的代码相当于__exit__里面的逻辑不管with代码块是否抛异常finally都能保证执行。关键一点是如果with代码块内抛了异常这个异常会在yield处重新抛出所以你可以用try/except捕获它做定制化处理contextmanager def suppress_exception(exc_type): try: yield except exc_type: print(f捕获到异常{exc_type.__name__}但选择忽略) with suppress_exception(ValueError): int(不是数字)这个写法比写一个完整的类简洁得多也是我个人日常用得最多的方式。需要提醒的是yield前如果抛了异常finally依然会执行所以把资源释放写在finally里面是最稳妥的。3.2 closing、suppress、redirect_stdout等现成工具contextlib里还有几个不需要自己动手就能直接用的上下文管理器。contextlib.closing用于那些实现了close()方法但没实现上下文协议的对象from contextlib import closing from urllib.request import urlopen with closing(urlopen(https://example.com)) as page: data page.read()没有closing的话你得自己写try/finally去调close()有它之后代码会干净很多。contextlib.suppress可以按异常类型静默特定异常比自己在except里写pass更明确from contextlib import suppress import os with suppress(FileNotFoundError): os.remove(temp_cache.json)这个用法在处理删除一个可能不存在的文件关闭一个可能已经断开的连接这类场景时特别好用代码语义也比try/except pass更清晰。contextlib.redirect_stdout和redirect_stderr可以把print输出重定向到文件或缓冲区适合测试或者封装日志from contextlib import redirect_stdout import io buffer io.StringIO() with redirect_stdout(buffer): print(这行不会打印到终端而是写到buffer里)3.3 ExitStack动态管理不确定数量的资源ExitStack是contextlib里最强大的工具之一它让你在运行时动态地、按需地注册多个清理回调退出时按逆序统一执行。看这个例子from contextlib import ExitStack def open_many_files(paths): stack ExitStack() files [] try: for p in paths: f open(p) stack.callback(f.close) # 或者 stack.push(f.close) files.append(f) return files, stack except Exception: stack.close() # 出问题就立刻清理已打开的文件 raise调用方用完以后只要执行stack.close()之前注册的所有close回调就会逆序执行。你也可以用stack.enter_context()把普通上下文管理器交给ExitStack托管from contextlib import ExitStack with ExitStack() as stack: f1 stack.enter_context(open(a.txt)) f2 stack.enter_context(open(b.txt)) session stack.enter_context(create_db_session()) # 退出with块时f2、f1、session会按逆序自动清理这种写法本质上是动态版的多层with嵌套非常适合那些资源数量不确定、需要条件式注册清理逻辑的场景。4. 实战自己动手实现几个常用上下文管理器4.1 带超时的代码块计时器很多性能分析工具都能做代码计时但用上下文管理器自己写一个更灵活可以精确控制哪个代码块要计、要不要打印日志、超时没超时。import time from contextlib import contextmanager contextmanager def timed_block(name, timeout1.0): start time.perf_counter() try: yield finally: elapsed time.perf_counter() - start if elapsed timeout: print(f[警告] {name} 执行耗时 {elapsed:.3f}s超过阈值 {timeout}s) else: print(f[信息] {name} 执行耗时 {elapsed:.3f}s) with timed_block(数据归一化, timeout0.5): # 模拟耗时操作 time.sleep(0.6)这个上下文管理器在生产环境做性能埋点时很实用。你不需要侵入业务代码结构只需要在被观测的代码块外面套一层with就行比手动记录time.time()前后差值要整洁得多。4.2 临时目录自动清理器写测试用例或数据处理脚本时经常需要创建临时目录放中间文件跑完以后想自动清理。下面的上下文管理器专门干这个import os import shutil import tempfile from contextlib import contextmanager contextmanager def temporary_dir(prefixtmp_): path tempfile.mkdtemp(prefixprefix) try: yield path finally: if os.path.isdir(path): shutil.rmtree(path) with temporary_dir() as work_dir: # 在work_dir里随便写文件 with open(os.path.join(work_dir, output.csv), w) as f: f.write(a,b,c\n1,2,3) # with块结束目录自动被删除不留垃圾这里用了tempfile.mkdtemp创建真实物理目录用shutil.rmtree递归删除。如果没有这个上下文管理器你每次都需要在测试结束后手动清理临时目录一旦中间断言失败抛异常临时目录就变成磁盘上的残留垃圾。有了它不管测试通过还是失败目录都会被清干净。4.3 可重试的数据库连接数据库操作是上下文管理器最典型的应用场景之一核心逻辑在于连接要关、事务要提交或回滚。实际写业务代码时我通常这样组织from contextlib import contextmanager contextmanager def db_transaction(connection): 将连接传入自动处理提交/回滚/关闭 try: yield connection connection.commit() except Exception: connection.rollback() raise finally: connection.close() # 使用示例 with db_transaction(conn) as cursor: cursor.execute(UPDATE users SET balance balance - 100 WHERE id 1) cursor.execute(UPDATE users SET balance balance 100 WHERE id 2)这里的关键在于commit的位置——它放在yield后面的try主体里也就是只有当with代码块正常执行完才会提交事务如果代码块内抛了异常rollback回滚后重新把异常抛出去避免脏数据落库。更进一步数据库连接经常因为网络抖动而断开你还可以在上下文管理器里写重试逻辑import time from contextlib import contextmanager contextmanager def db_retry_connect(connection_factory, retries3, delay0.5): last_exc None for attempt in range(retries): conn connection_factory() try: yield conn return except ConnectionError as exc: last_exc exc conn.close() time.sleep(delay * (attempt 1)) except Exception: conn.close() raise raise last_exc这个例子展示了上下文管理器不只是资源释放工具它还能承载策略逻辑比如重试、超时、降级。调用方写业务代码时完全不需要关心重试次数和等待时间只需要把真正的业务逻辑放在with块里。4.4 自定义锁与条件变量的管理threading.Lock本身实现了上下文管理协议可以直接用with lock:来加锁/释放但有时你需要对锁行为做定制。比如给锁加一个超时控制避免线程死等import threading from contextlib import contextmanager contextmanager def timed_lock(lock, timeout5.0): acquired lock.acquire(timeouttimeout) if not acquired: raise TimeoutError(f在 {timeout} 秒内未能获取锁) try: yield lock finally: lock.release() lock threading.Lock() def worker(): try: with timed_lock(lock, timeout3.0) as lk: # 临界区代码 print(拿到锁开始工作) except TimeoutError: print(锁等待超时放弃) threads [threading.Thread(targetworker) for _ in range(5)] for t in threads: t.start() for t in threads: t.join()它的底层原理并不复杂acquire(timeout)返回布尔值拿不到锁就主动抛出超时异常而不是让线程无限期等下去。在微服务架构里等锁超时比无限阻塞要安全得多因为超时可以被监控、被快速失败。5. 踩坑记录上下文管理器实践中的几个隐蔽问题5.1 异常被静默吞掉的后果前面讲过__exit__返回True会吞异常但这里再展开说一个真实场景。我曾经在一个内部工具里见过这样的代码from contextlib import contextmanager contextmanager def api_session(): session create_session() try: yield session except Exception: ... # 这里打印了日志但忘了重新raise问题在于except捕获异常后既没有raise也没有在finally里做额外处理contextmanager会把函数内吞掉的异常视为已处理调用方的try/except就永远接不到错误。业务代码里一旦发生这种情况API调用失败但调用方毫无感知数据不同步的问题往往要延迟很久才暴露。排查这类问题的最快方法是在__exit__或except里打日志时顺便把exc_info完整输出并确认是否重新抛出。一个更稳妥的习惯是上下文管理器里捕获异常后默认加上raise除非你有充分理由确定不要传播。5.2 生成器上下文管理器中的yield陷阱contextmanager装饰的生成器有一个容易踩的坑如果函数在yield之前就抛了异常with块根本不会执行但你可能以为资源已经被正确清理了。看这个例子contextmanager def risky(): resource acquire_resource() # 这里抛异常 try: yield resource finally: release_resource() with risky() as res: ...如果acquire_resource()抛异常risky()函数根本进不到try所以不会执行finally。这个行为其实是对的——都没成功获取资源当然没有释放可言。但很多人写代码时会把获取资源和使用资源的边界搞混把本应放在yield之后的安全清理逻辑错误地放在了try之外导致异常发生时资源泄漏。我的建议是所有需要释放的资源获取动作都放进try里面。也就是先resource None然后用try/finally包住yield这样即使获取过程中出错finally里也能安全地判断并清理。5.3 装饰器和上下文管理器组合的顺序问题Python里装饰器和上下文管理器经常搭配使用但顺序错了会让代码行为和预期完全相反。比如你在FastAPI或Flask里可能见过这样的写法这里用Flask风格示意from functools import wraps def with_transaction(): def decorator(func): wraps(func) def wrapper(*args, **kwargs): with db_transaction(conn): return func(*args, **kwargs) return wrapper return decorator with_transaction() def update_balance(user_id, amount): ...这种组合本身没问题但如果你在装饰器内部用了contextmanager又同时在函数外部也用with就要特别注意执行顺序。比如下面这段代码的运行顺序是外层with先进入装饰器包装的函数体内部再进入第二个with退出时内层先退、外层后退。这个顺序如果搞反事务的提交时机就错了。**我的经验法则是先理解一层上下文管理器的生命周期再叠加多层涉及多层时从最外层往里推进入顺序从最内层往外推退出顺序。**这样无论组合多复杂都能理清逻辑。5.4 ExitStack使用时的常见误操作ExitStack确实强大但刚上手时容易出问题。最经典的一个误操作是把return或yield放在ExitStack的with块外导致资源提前释放。看这段有问题的代码from contextlib import ExitStack def get_files(): stack ExitStack() with stack: f1 stack.enter_context(open(a.txt)) f2 stack.enter_context(open(b.txt)) return f1, f2 # 这里with块已退出文件都关了调用方拿到的f1、f2实际上是已经关闭的文件对象读数据必然报错。正确做法是让stack的生命周期覆盖整个文件使用期通常让get_files也变成一个生成器或上下文管理器from contextlib import ExitStack, contextmanager contextmanager def open_many(paths): with ExitStack() as stack: files [stack.enter_context(open(p)) for p in paths] yield files另外一个容易忽略的点是pop_all()方法可以把当前已注册的回调整体搬走这在将资源所有权从一个上下文管理器转移给另一个时很有用但用不好会让清理逻辑彻底失序。新手阶段尽量别折腾pop_all等把ExitStack的基本用法吃透了再进阶。6. 把上下文管理器用在业务代码里的几个设计建议6.1 用上下文管理器封装事务边界而不是散落try/except业务系统里最常见的反模式是每个函数里都写try/except做事务提交回滚代码重复不说一旦新增一个数据库操作经常忘记补事务边界或者提交时机搞错。用上下文管理器封装事务边界之后业务函数的职责就变得很纯粹——只关心数据怎么处理不用关心事务怎么开、怎么提交、怎么回滚。这个思维方式的转变对代码可维护性的提升非常明显。6.2 把耗时操作的埋点做成可复用的上下文管理器除了性能分析上下文管理器还特别适合做日志链路追踪。比如你可以写一个trace_span(name)的上下文管理器import uuid from contextlib import contextmanager contextmanager def trace_span(name): request_id uuid.uuid4().hex[:8] print(f[trace] 开始 {name}request_id{request_id}) try: yield request_id finally: print(f[trace] 结束 {name}request_id{request_id})这个设计思路完全可以接入现有的日志系统或链路追踪平台而且不需要改动业务函数的签名对老代码的侵入性极低。6.3 小心上下文管理器对象与上下文管理器本身的生命周期这是最后一个我特别想强调的点。定义上下文管理器之后要分清两种使用方式# 方式一把上下文管理器对象赋给变量在外部管理 ctx managed_resource() with ctx: pass # 方式二直接使用 with managed_resource(): pass方式一里有隐藏风险ctx对象的__enter__和__exit__可以多次调用吗可以重入吗如果实现不支持重复进入那么第二次with ctx就会出问题。多数自实现上下文管理器不处理重入语义所以除非明确知道它是可重入的否则不要复用同一个上下文管理器对象。这个坑在写测试时会特别常见——有人为了减少重复代码在setUp里创建一个上下文管理器对象然后在多个测试方法里复用结果第二个测试就会出诡异的问题。正确的做法是让__enter__每次都返回全新的状态或者干脆在with语句内部创建新的上下文管理器实例。7. 我的实测体会上下文管理器是我在使用Python时受益最大的语言特性之一它看似简单背后却凝聚了资源安全和代码结构两个维度的思考。写这个主题的文章我最大的体会是普通开发者用with语句是在使用一个语法进阶开发者用with语句是在设计一段代码的执行边界。从文件操作到数据库事务从锁管理到临时资源生命周期上下文管理器把进入和退出这两个隐形的边界显性化了。它让我在写业务代码时更敢放心地分配资源因为知道释放逻辑一定会被可靠执行哪怕发生异常。最后分享两个我自己的习惯一是能用contextmanager装饰器解决的就尽量不写原生类代码量少、读起来直观二是凡是涉及数据库连接、文件句柄、网络连接这些重量级资源的上下文管理器__exit__里一定写上完整的状态日志并且严格保持raise继续传播绝不静默吞异常。如果你之前只用过with open建议找个休息日下午把contextlib的文档从头到尾翻一遍再顺手把这些实战例子在自己环境里跑一遍。相信我下次你写代码时看待with的眼神会完全不一样。