你有没有为了找一个“句柄泄漏”问题,把线上脚本翻了个底朝天,最后发现就是某个文件对象没关?我有一次排查连接数暴涨,查了半天,才发现是一个爬虫任务里每次拉数据都用open()拿个文件句柄,但有几条异常分支把close()跳过去了,句柄数一路涨到系统上限。这个教训让我后来非常认真地研究了一遍 Python 的上下文管理器与 with 语句——不只是会用with open(...) as f,而是真正把它当成一套“资源收尾”的协议来用。
如果你写过with open(...) as f,其实你已经用过上下文管理器了,但大多数人只停留在“这么写能自动关文件”的层面,并不知道with背后到底发生了什么,更不清楚__exit__的返回值为什么会影响异常是否被吞掉。这篇文章就把这套机制从头到尾理清楚,内容包括协议的执行细节、标准库里那些好用的现成上下文管理器、手写自定义上下文管理器的两种方式、实战场景(数据库事务、线程锁、临时环境变量),还有我踩过的一些坑。无论你是刚入门的 Python 新手,还是想把自己的工具库做得更优雅的老手,这篇文章都值得花十分钟读完。
1. 上下文管理器解决的是“收尾”这回事
1.1 从一次文件句柄泄漏说起
先还原一下我最开始提到的那个问题。早期写脚本时,我的代码长这样:
f = open("data.txt", "r", encoding="utf-8") data = f.read() process(data) f.close()看起来没毛病,读文件、处理、关闭,顺序很清晰。但问题在于:process(data)里面如果抛了异常,f.close()就不会执行,文件对象就没有被关闭。可能你会说“脚本跑完进程退出,操作系统会回收”,话是没错,但很多脚本不是跑一次就结束的,它可能在一个长驻进程里循环执行。每跑一轮漏一个句柄,跑几千轮之后,句柄就耗尽了。这个问题在小脚本里不明显,一旦放进服务端或者定时任务里,就是定时炸弹。
后来我改用try...finally来收拾局面:
f = open("data.txt", "r", encoding="utf-8") try: data = f.read() process(data) finally: f.close()这个写法比原来强多了,无论process(data)是否抛异常,finally里的f.close()都会被调用。代码虽然多了两行,但可靠性上了一个台阶。当时我以为这已经是 Python 资源管理的“最优解”了,直到接触了 with 语句,才发现它能做更多。
1.2 with 语句的意图与优势
with语句要解决的核心问题,就是“收尾逻辑不要散落在各处,也不要因为异常而跳过”。它的写法更加紧凑:
with open("data.txt", "r", encoding="utf-8") as f: data = f.read() process(data)这两段代码效果等价。with语句在进入代码块之前,会先调用对象上的某个“准备方法”;在代码块结束之后,无论正常结束还是抛异常,都会调用另一个“清理方法”。这就是上下文管理器的基本形态。
它解决的问题是结构化的:你不用再记忆“哪里开了就要在哪里关”,只要把“使用资源”的代码放进with块里,退出时机由协议保证。这种思想还可以延伸到数据库连接、线程锁、临时环境变量、网络连接、临时文件目录等几乎所有需要成对操作的地方。
另外从代码可读性角度来说,with能够直接从视觉上划清“资源生命周期”的边界。别人看你的代码,一眼就能看出f的存活范围是从with进入到块结束,不需要在代码里到处找close()。这对团队协作尤其重要——我记得有一次 code review,看到有人在一个函数开头 open 文件、函数结尾 close,中间隔了几十行逻辑,我就觉得这段代码迟早要出事,果然后来有人在中途加了提前 return 的分支,文件就没被关掉。用with之后,这类问题基本上从源头杜绝了。
2. 协议执行原理:with 到底帮你做了什么
2.1enter与exit:协议的两个钩子
凡是能配合with使用的对象,都实现了上下文管理器协议。这个协议在 Python 文档里定义得非常精简,核心就两个方法:
__enter__(self):进入with块之前被调用,返回值会赋给as后面的变量(如果有as子句的话)。__exit__(self, exc_type, exc_value, traceback):退出with块时被调用,负责清理资源,并决定异常是否需要继续向上抛出。
你可以把__enter__理解为“开门”,把__exit__理解为“关门”,with语句则是那个“不管中间发生什么,都会帮你关门”的管家。
来看一个最朴素的自定义例子:
class Resource: def __enter__(self): print("进入 with 块:获取资源") return "resource_id" def __exit__(self, exc_type, exc_value, traceback): print("退出 with 块:释放资源") return False with Resource() as res: print("使用资源:", res)执行结果如下:
进入 with 块:获取资源 使用资源: resource_id 退出 with 块:释放资源这个例子展示了最基本的流程。__enter__的返回值"resource_id"被送到了res变量里,with块执行完后,__exit__自动被调用。注意__exit__的三个参数:exc_type是异常类型,exc_value是异常实例,traceback是堆栈对象。如果块内没有异常,这三个参数的值都是None。
2.2 异常走向:exit返回值的分水岭
很多人没注意到的是,with块内如果抛出异常,这个异常会不会传到外层,取决于__exit__的返回值。这个点非常关键,可以说整个上下文管理器协议里最容易被人忽略的就是它。
__exit__返回True时,表示“我已经处理了这个异常,你不必再向上抛了”;返回False(或者返回值不是真值)时,表示“我没处理,请你继续向上抛”。
举个具体例子:
class IgnoreDemo: def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): print("捕获到异常类型:", exc_type.__name__) return True with IgnoreDemo(): raise ValueError("这个异常被吞掉了") print("这一行能打印出来")输出:
捕获到异常类型: ValueError 这一行能打印出来因为__exit__返回了True,with块内的ValueError被“消化”掉了。如果代码块内没有异常,返回值是True还是False其实没有区别,反正__exit__只是被调用一下而已。
这就是一个分水岭:返回True就相当于一个隐形try...except,返回False则相当于把异常交给上层。在自定义上下文管理器时,绝大多数情况下你都不应该在异常发生时隐瞒异常,除非你明确知道自己想做什么(比如写一个“忽略特定异常”的工具类)。
2.3 as 子句、多上下文并列,以及作用域盲区
实际使用with时,有一些细节值得单独说。
第一,as子句可以不要。比如你用threading.Lock只需要保证加锁和释放锁的时机,并不需要拿到锁对象本身,所以可以写成:
lock = threading.Lock() with lock: # 临界区代码 pass甚至__enter__返回了对象也不用管它,只要保证__exit__按时代你执行清理。
第二,多个上下文管理器可以并列写。例如同时打开两个文件、做行对齐比较:
with open("a.txt", "r", encoding="utf-8") as f1, open("b.txt", "r", encoding="utf-8") as f2: for line1, line2 in zip(f1, f2): print(line1.strip(), line2.strip())并列多个上下文管理器时,它们会按从左到右的顺序进入,按从右到左的顺序退出。这个顺序很重要,特别是上下文之间存在依赖关系时——比如一个管理器依赖另一个管理器创建的资源,退出时就应该先销毁依赖方的资源,后销毁被依赖方的资源。Python 的嵌套退出顺序正好是从里到外,符合直觉。
第三,as后面的变量在with块结束后依然可以访问,但资源已经不在了。这是个很容易踩的坑。比如:
with open("data.txt", "r", encoding="utf-8") as f: content = f.read() print(f.closed) # Truef这个变量没被销毁,但文件已经关闭,如果你还想继续读它就会报ValueError: I/O operation on closed file。这不是 bug,是设计如此:上下文管理器负责的是“生命周期管理”,不负责“变量清理”。记住这个细节,很多“明明我在 with 块里读了文件,出来就报错”的问题就迎刃而解。
3. 标准库里的现成上下文管理器
3.1 文件与资源管理类
Python 标准库提供了许多现成的上下文管理器,不需要我们自己动手实现,直接拿来用就行。我先列几个最常见的:
open(...):文件对象本身就是一个上下文管理器,__exit__负责关闭文件。tempfile.TemporaryDirectory():创建一个临时目录,退出时自动删除整个目录及其内容。写测试代码、做临时缓存时特别方便。contextlib.closing(obj):如果某个对象只有close()方法而没有实现上下文管理器协议,可以用closing包装一下,保证退出时调用close()。
TemporaryDirectory我在做数据处理时经常用,比如解压一个压缩包,只想提取里面的内容做临时分析,分析完事就删,不给磁盘留垃圾:
import tempfile import shutil from pathlib import Path with tempfile.TemporaryDirectory() as tmpdir: tar_path = Path(tmpdir) / "data.tar.gz" shutil.copy("source.tar.gz", tar_path) # 解压、处理、分析 result = do_work(tmpdir) # 退出时 tmpdir 自动被清理如果没有上下文管理器,你得手写try/finally加shutil.rmtree,代码量会明显增加。
还有contextlib.closing,比如某些数据库连接对象或者自定义的流对象,只实现了close(),没有__enter__/__exit__,你就可以这样做:
from contextlib import closing import urllib.request with closing(urllib.request.urlopen("https://example.com")) as resp: html = resp.read(2000)虽然urlopen的结果本身也支持上下文管理,但closing这种通用包装在兼容老 API 时很管用。
3.2 锁与并发控制类
并发编程里,锁和条件变量的使用也可以用with来管理。threading.Lock实现了上下文管理器协议,__enter__加锁,__exit__释放锁,即使临界区抛异常也不会造成死锁。
对比一下两种写法:
# 手动加锁解锁 lock.acquire() try: count += 1 finally: lock.release()# 上下文管理器写法 with lock: count += 1两种写法本质一样,但后者明显少了一层缩进,阅读的时候心理负担更小。多线程代码最怕的就是某个分支忘了release(),一旦漏掉,其他线程就会永远卡在acquire()上。我在实际项目里看到过因为异常路径导致死锁的事故,原因就是手动try/finally里漏处理了一个异常类型。用with lock之后,这类事故不会再出现。
除了Lock,threading.RLock、threading.Condition支持上下文管理器,multiprocessing里的Lock也一样。信号量Semaphore其实也实现了协议,不过信号量通常不推荐在业务代码里大量使用,这里就不展开了。
3.3 环境切换与临时状态类
有些上下文管理器不是管资源,而是管“状态”。它们进入时修改环境,退出时恢复原状,这种模式在测试和临时实验里极其常见。
举个例子,decimal.localcontext可以临时修改十进制运算的精度:
import decimal from decimal import Decimal, localcontext with localcontext() as ctx: ctx.prec = 2 print(Decimal("1.23") / Decimal("3")) # 0.41 print(Decimal("1.23") / Decimal("3")) # 正常精度同样用途的还有numpy.errstate,它可以临时忽略或者捕获浮点计算中的除零警告、溢出警告,而不是全局修改warnings配置:
import numpy as np with np.errstate(divide="ignore", invalid="ignore"): result = np.array([1.0, -1.0, 0.0]) / np.array([0.0, 0.0, 0.0])还有unittest.mock.patch,它也是上下文管理器,进入时替换目标对象,退出时自动恢复原样。我们在写单元测试或者临时调试时,经常用到:
from unittest.mock import patch with patch("module_name.get_user_name", return_value="mock_user"): result = some_function()这种“临时改一下,用完恢复”的模式,如果全靠手动try/finally,每次都要记住恢复状态,非常容易遗漏。用上下文管理器之后,状态切换的边界变得非常清晰,测试代码的可靠性也提高了。
4. 自定义上下文管理器的两种主流写法
4.1 类方式:可读性高、状态可复用
当系统里没有现成的上下文管理器可以满足需求时,就该自己动手写了。第一种方式是定义一个类,实现__enter__和__exit__两个方法。
我用一个“代码块执行耗时统计”的例子来说明。假设你想临时统计某段代码跑了多长时间,不想每次都写start = time.time()和elapsed = time.time() - start这种重复代码,可以写一个耗时上下文管理器:
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}s") with Timer(): time.sleep(0.1)这个类的核心设计点是:__enter__里记录开始时间,并把self返回出去。为什么要返回self?因为外面如果想知道结束后的耗时,可以通过as拿到这个实例,然后访问它的属性:
with Timer() as t: result = expensive_function() print("还没退出,当前耗时:", t.elapsed) # 此刻 elapsed 还不存在 # 退出后可以访问 elapsed print("最终耗时:", t.elapsed)不过要注意,块内访问t.elapsed会提前抛AttributeError,因为elapsed是在__exit__里才创建的。如果你需要在块内也能看到中间耗时,可以在__enter__里也初始化一个elapsed = 0,然后手动更新。
类方式的优点很明显:状态可以保存在实例属性里,逻辑可以写得比较完整,适合那些需要跨多个方法维护状态的复杂场景。缺点也明显:代码量偏多,如果只是想包一层简单的逻辑,写一个类显得有点“重”。
4.2 contextmanager 装饰器:代码更短、思路更直
如果你不想写一个完整的类,可以用@contextlib.contextmanager装饰器,把一段普通函数变成上下文管理器。原理是利用生成器的特性:函数里yield之前的代码相当于__enter__,yield之后的代码相当于__exit__。
上面那个耗时统计,用装饰器可以写成:
import time from contextlib import contextmanager @contextmanager def timer(name="代码块"): start = time.perf_counter() try: yield finally: elapsed = time.perf_counter() - start print(f"{name}耗时:{elapsed:.4f}s") with timer("数据清洗"): time.sleep(0.2)这里有个关键点:yield前后的清理代码最好放在finally里。原因和手动资源管理一样——如果with块内部抛了异常,yield后面的代码不会自动执行,必须通过try/finally保证清理逻辑一定会跑。
装饰器写法的优势在于:代码紧凑,并且没有__enter__/__exit__那种方法拆分的割裂感——你只需要从上往下读一个函数,上半部分是准备工作,中间是挂起点,下半部分是收尾工作。这种风格在写一次性包装逻辑时非常顺手。
我在封装 Redis 连接的时候就用了这种方式:
import redis @contextmanager def redis_connection(): client = redis.Redis.from_url("redis://localhost:6379/0") try: yield client finally: client.close() with redis_connection() as r: r.set("key", "value")4.3 两种方式如何取舍
我自己心里的选择标准是这样的:
如果你需要的是一个“可以被反复初始化并复用”的类,或者状态需要在多个上下文之间共享,用类方式更合适,因为实例属性可以被外部持有。
如果你只是需要“进入时干一件事,退出时干一件事”,这种一次性逻辑用@contextmanager最省事。写测试工具、临时包装老接口时基本都用装饰器。
还有一个平衡的技巧:当逻辑复杂到装饰器函数体内要写多个内部函数时,就说明该考虑类了。我曾经把一段包装逻辑写进@contextmanager里,结果函数体达到三十多行,不仅嵌套深,连yield前后的边界都看不清。后来改成类,方法一拆,反而清晰很多。
4.4 实用案例:一个“临时切换工作目录”的上下文管理器
我觉得光说概念不够带劲,再给一个很实用的自定义案例:临时切换当前工作目录,并在退出时自动恢复原目录。这对于写脚本时批量操作文件很有用。
import os from contextlib import contextmanager @contextmanager def cd(path): old_dir = os.getcwd() os.chdir(path) try: yield finally: os.chdir(old_dir) with cd("/tmp"): # 这里所有的相对路径都基于 /tmp print(os.getcwd()) print(os.getcwd())执行的时候,你会在with块内看到/tmp,出来后回到原来的目录。这看起来简单,但如果不用上下文管理器,你会发现在脚本的多个位置都要重复os.getcwd()、os.chdir()和异常兜底,代码很容易乱。而且一旦漏了恢复,后面的步骤踩的全是坑——轻则文件写错位置,重则把临时文件生成到了错误目录。
如果你没有用过类似的功能,强烈建议收藏这个例子。以后你做文件批处理、自动化任务时,不用再为“切换目录后忘了切回来”这种低级问题买单。
5. 实战:用上下文管理器治理项目里的老代码
理论讲了很多,关键还是要落到项目里。这一部分我挑几个高频场景,讲讲怎么把上下文管理器用在真实代码中。
5.1 数据库事务封装:提交与回滚的自动化
数据库操作是最典型的“成对收尾”场景。我们用sqlite3举例,平时连数据库做写操作,最担心的就是中间某条语句失败,事务状态变成半死不活。
常规写法:
import sqlite3 conn = sqlite3.connect("app.db") try: cur = conn.cursor() cur.execute("INSERT INTO log(content) VALUES (?)", ("hello",)) conn.commit() except Exception: conn.rollback() raise finally: conn.close()这种写法写多了,人容易麻木,而且一旦在同一个函数里开了好几个事务,每个都套一层try/except/finally,代码就变得很臃肿。我更喜欢把事务提交和回滚封装成一个上下文管理器:
from contextlib import contextmanager @contextmanager def transaction(conn): try: yield conn except Exception: conn.rollback() raise else: conn.commit()用起来是这样:
conn = sqlite3.connect("app.db") with transaction(conn) as conn: conn.execute("INSERT INTO log(content) VALUES (?)", ("hello",))如果with块里的 SQL 执行正常,退出时提交;如果中途抛了异常,自动回滚,并把异常原样抛给上层。这个封装同时体现了@contextmanager和try/except/else/finally的配合,尤其是else分支的用途——只在没有异常时才commit()。这里有个细节是:事务上下文管理器只封装提交和回滚,不负责关闭连接,连接的生命周期应该由更外层管理,不要混在一起,否则你就没法在两个事务之间复用同一个连接。
5.2 锁定并发临界区:一个简单的库存扣减场景
再来看一个多线程扣减库存的小例子。假设库存数量越来越多,多个线程同时操作又没有锁,会出现典型的“读脏值更新丢失”。用threading.Lock加锁是常规解法,但锁的释放时机同样容易出问题。
import threading stock = 10 stock_lock = threading.Lock() def decrease_stock(quantity): global stock with stock_lock: if quantity > stock: raise ValueError("库存不足") stock -= quantity这里的with stock_lock保证了decrease_stock在同一时间只有一个线程能进入临界区。即使quantity > stock抛异常,锁也会自动释放,不会让其他线程卡死。
如果要兼顾性能,还可以再加一个条件变量做等待,比如库存恢复时通知线程继续扣减。threading.Condition同样实现了上下文管理器协议,with语句可以同时管理锁和条件等待的收尾,这在实现生产者消费者模型时非常好用。
5.3 临时环境变量切换:改完记得还原
平时调试人工造数据时,我经常要临时改环境变量,比如设置一个DEBUG标志或者修改语言环境。
import os from contextlib import contextmanager @contextmanager def set_env(key, value): old_value = os.environ.get(key) os.environ[key] = value try: yield finally: if old_value is None: os.environ.pop(key, None) else: os.environ[key] = old_value with set_env("DEBUG", "1"): # 这段代码里能读到 DEBUG=1 print(os.environ.get("DEBUG")) # 出了 with 块,环境变量恢复原样这段代码的逻辑虽然简单,但非常值得注意:如果之前没有这个环境变量,退出时要pop掉,否则只是把值改回去就多出一个本来不存在的变量。这种“尽可能还原现场”的意识,在运维脚本和测试代码里极其重要,否则测试之间会相互污染环境。
5.4 第三方库中的常见上下文管理器
除了标准库,很多第三方库也大量使用上下文管理器。比如requests库的Session,你可以把它当上下文管理器用,保证请求结束后会话被正确关闭:
import requests with requests.Session() as session: session.get("https://example.com")再比如pandas的ExcelWriter:
import pandas as pd with pd.ExcelWriter("output.xlsx") as writer: df1.to_excel(writer, sheet_name="Sheet1") df2.to_excel(writer, sheet_name="Sheet2")退出时自动保存并关闭文件。对比一下手动save()和close(),用with的好处是不容易因为异常中断而留下一个空文件或者未落盘的数据。做数据报表的时候,我几乎都是这么写的,很少再手动调用save()。
6. 常见误区与排查心得
看得多了,写得多了,自然也就踩了不少坑。我把跟上下文管理器有关的常见误区和排查经验整理一下,希望能帮你少走弯路。
6.1exit返回 True 把异常吞了,你还不自知
这是我最想强调的一点。很多人写自定义上下文管理器时根本没意识到__exit__的返回值会影响异常传播。如果你返回了True,那么异常就会被静默吞掉,with块之后的所有代码还能继续跑,错误可能隐藏得很深。
我举一个踩坑案例:当时我写了一个“发送埋点日志”的上下文管理器,想着“无论埋点是否成功,都不能影响主流程”,于是__exit__直接写了return True。后来某段时间埋点系统一直报错,但主流程毫无异常,排查了大半天才发现是埋点任务里的一段校验逻辑误写了raise,被__exit__吞掉了。
经验是:如果你不是刻意要实现“忽略某类异常”的语义,__exit__统一返回False,或者干脆不写返回值(隐式返回None,同样等于假值)。不要图省事。
6.2 as 子句的对象块外仍可用,但资源已关闭
前面已经提过,with open(...) as f结束后f还存在,但文件已经关闭。这个问题在写文件时尤其容易误判:
with open("out.txt", "w", encoding="utf-8") as f: f.write("hello") # 错误:继续写 f.write(" world")这段代码会抛ValueError: write operation on a closed file。排查思路很简单:怀疑文件对象被提前关闭时,先看f.closed是不是True。但要从根源上杜绝,还是要把对人、对文件的操作全部限制在with块内部。
6.3 with 后面加括号的版本兼容问题
Python 3.9 及更早版本里,with后面如果写括号,会把它解析成元组,导致莫名其妙的AttributeError。Python 3.10 引入了带括号的上下文管理器写法,允许这样写:
with ( open("a.txt", "r", encoding="utf-8") as f1, open("b.txt", "r", encoding="utf-8") as f2, ): ...这个语法在 3.10+ 是好用的,但如果你的项目要兼容 3.8 或 3.9,千万别用这种写法。我在平时维护老项目时,遇到这类跨版本兼容问题的方法是:项目里统一用逗号并列的旧写法,不写括号,这样在 2.7(如果有遗留)到 3.13 都能跑。如果你非要启用新写法,记得在 CI 里把 Python 版本固定好,免得部署环境一升级代码就炸。
6.4 异常发生后exit内的清理逻辑抛了新异常
__exit__本身也是代码,它也可能抛异常。如果with块体内的异常还没来得及处理,__exit__又抛了一个新异常,那么 Python 会用一个__context__链把两个异常关联起来,新异常冒泡到顶层,原异常作为 context 存在。
实际影响是你可能看到 traceback 里有两个异常,排查时吓一跳。比如:
class BadCleanup: def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): raise RuntimeError("清理阶段崩了") with BadCleanup(): raise ValueError("业务异常")排查这类问题,我的建议是:在写自定义上下文管理器的__exit__时,尽量不要做可能抛异常的操作。如果确实可能有异常,比如关闭网络连接时对方已经断线,可以用try/except把清理阶段的异常先记录下来,不要直接把它抛出去。
7. 进阶:ExitStack 与异步上下文管理器
如果到这里你觉得还不过瘾,那接下来这几个进阶技巧很适合你。
7.1 ExitStack:把“不确定数量”的资源交给它管
有时候你提前不知道要进入多少个上下文管理器,比如动态决定要打开几个文件、注册多少个回调。contextlib.ExitStack就是为这种“动态管理”场景设计的。
用法很简单:
from contextlib import ExitStack files_to_open = ["a.txt", "b.txt", "c.txt"] with ExitStack() as stack: files = [stack.enter_context(open(f, "r", encoding="utf-8")) for f in files_to_open] # 用 files 里的文件对象干活 for f in files: print(f.read(10)) # 退出 with 块时,所有文件自动关闭,顺序为后进先出ExitStack的enter_context会调用传入对象的__enter__,并把未来的清理工作压入栈内。退出时,栈里所有资源的__exit__会被依次调用,不用管文件数量是多少。
这个工具在做测试夹具的时候尤其好用,比如你要 mock 掉很多外部依赖,数量是由配置决定的,用ExitStack可以动态注册所有patch。我记得在写集成测试时,一个用例需要 mock 六七个服务,用ExitStack一下子清爽了。
ExitStack还有一个非常实用的场景:如果你不确定某个对象需不需要被“清理”,可以先push一个它的close方法,如果后续逻辑发现不需要,可以用pop_all()或者callback(None)把它撤掉。这种灵活性在复杂资源管理里非常稀缺。
7.2 异步上下文管理器:asyncio 场景下的 async with
Python 3.5 引入async with之后,异步上下文管理器的使用也越来越普遍了。它和同步版本对应,只是两个钩子方法变成了__aenter__和__aexit__,并且必须先await它们。
看一个最简单的例子:
import asyncio class AsyncResource: async def __aenter__(self): print("异步获取资源") return self async def __aexit__(self, exc_type, exc_value, traceback): print("异步释放资源") async def main(): async with AsyncResource(): print("使用资源") asyncio.run(main())实际开发里,aiohttp.ClientSession、异步数据库连接池都会支持这种写法。用async with可以保证异步客户端在请求结束后被正确关闭,不会因为忘关连接造成连接池耗尽。这个问题的危害和文件句柄泄漏类似,但更隐蔽——因为异步程序通常跑得更久、并发更高,连接耗尽带来的问题往往更严重。
如果你要写异步版本的上下文管理器,需要同时注意:__aenter__和__aexit__必须返回 awaitable 对象,也就是说必须定义成async def,而不能只是普通函数。同步的__enter__不会帮你在异步中自动变成__aenter__。
7.3 小而美的工具:contextlib.suppress 与 nullcontext
最后聊两个看起来不起眼、但用起来非常顺手的小工具。
contextlib.suppress可以让你优雅地忽略特定异常,相当于一个简化版的try/except:
from contextlib import suppress with suppress(FileNotFoundError): os.remove("temp_file.txt")如果文件不存在,os.remove抛的FileNotFoundError会被静默忽略。用suppress比写try/except更加“声明式”,别人一眼看出你只是不想处理这个异常,而不是想在里面写逻辑。注意它和你自己写__exit__返回True的区别:suppress是标准库专门针对“忽略指定异常”设计的,语义更清楚。
contextlib.nullcontext则是一个“什么都不做的上下文管理器”。它的用途主要是提供统一的接口占位。我在写代码时经常遇到这样的场景:某个函数需要往with块里塞一个可选的“锁对象”,如果不需要加锁,就传一个nullcontext(),这样函数的调用方可以统一对待带锁和不带锁的情况,不用在函数内部写 if-else。
from contextlib import nullcontext import threading def process(lock=None): lock = lock or nullcontext() with lock: print("工作,可能加锁也可能不加锁")原本如果我不这样设计,就要写两个分支,一个带with lock,一个不带,代码直接膨胀一倍。用nullcontext()之后,边界清晰了很多。
收尾:一点个人的使用习惯
如果用一句话总结这些经验的通用性,那就是:所有“开始做一件事、结束做一件事”的逻辑,都可以考虑用上下文管理器来框定边界。我自己在使用时的习惯是:能用标准库现成的绝不手写,需要手写时优先考虑@contextmanager,只有当状态管理复杂到需要类来承载时才回头写类。每写一个新上下文管理器,我都会在__exit__里刻意检查返回值和清理代码的健壮性,避免“吞异常”“清理阶段二次爆炸”这类坑。
最后分享一个很小但很值得养成的习惯:在写命令行小工具和数据处理脚本时,把“输出日志到文件”也封装成上下文管理器,进入时重定向sys.stdout,退出时恢复,这样看着既优雅又不容易漏掉。用熟了之后,你会发现自己写代码时越来越少担心“忘了关什么”这种事,上下文管理器把资源生命周期这点事安排得明明白白,剩下的心思可以全部放在业务逻辑上。