async with要求__aenter__必须用async def定义,返回值自动await;__aexit__需完整声明三参数;禁止阻塞操作和重试循环,仅做确定性准备与清理。

async with 语句要求 __aenter__ 必须是协程函数
直接在类里写 def __aenter__(self) 是无效的——即使里面用了 await,Python 也会报 TypeError: object X can't be used in 'await' expression。关键点只有一个:必须用 async def 声明,返回值会被自动 await。
常见错误是把同步逻辑硬塞进异步方法里,比如调用 time.sleep() 或阻塞式 I/O(如 requests.get()),这会卡住整个事件循环。正确做法是只做真正可挂起的操作:
-
await asyncio.sleep()替代time.sleep() - 用
aiohttp/aiomysql等异步库替代同步库 - 资源初始化本身不耗时?直接
return self即可,无需await
__aenter__ 和 __aexit__ 的返回值与参数必须匹配
__aenter__ 的返回值会绑定到 as 后的变量;__aexit__ 的三个参数(exc_type, exc_value, traceback)必须完整声明,哪怕你只打算忽略异常。漏掉任意一个都会导致 async with 执行失败。
典型写法:
class AsyncResource:
async def __aenter__(self):
await self._connect() # 比如建立 aiohttp session
return self # 或者 return 某个具体对象,如 self.db_conn
<pre class="brush:php;toolbar:false;">async def __aexit__(self, exc_type, exc_value, traceback):
await self._close() # 清理连接、释放锁等
# 不需要显式 return,除非你想抑制异常(return True)
注意:__aexit__ 里 return True 会吞掉异常,和同步上下文管理器行为一致,但容易掩盖问题,慎用。
别在 __aenter__ 里做重试或长轮询逻辑
异步上下文管理器不是启动后台任务的入口。如果 __aenter__ 里写了死循环 while True: await asyncio.sleep(1); try: ... except: continue,会导致 async with 永远无法进入主体代码块。
真正需要重试的场景,应该封装成独立的异步方法(比如 async def connect_with_retry(self)),在 __aenter__ 中调用它一次即可。上下文管理器的核心职责是“确定性地准备和清理”,不是“兜底运行”。
- 连接失败应抛出异常,由外层决定是否重试
- 想自动重连?考虑用
asyncio.wait_for()包裹连接逻辑,设超时 - 需要后台心跳?启动
asyncio.create_task(),但务必在__aexit__中取消它
测试 __aenter__ 是否生效,最简单的办法是加日志 + await 断点
光看语法没错没用。实际运行中常遇到:协程没被调度、事件循环已关闭、或 async with 被套在普通函数里(没用 async def 包裹)。验证方式很直接:
- 在
__aenter__开头加print("entering...")或logging.debug - 用
asyncio.run()启动最简测试,别用 Jupyter 或某些框架的伪事件循环 - 确保调用链全是异步的:外层函数必须是
async def,且用await或asyncio.run()驱动
最容易被忽略的一点:如果你在类实例化后立刻 await obj.__aenter__()(绕过 async with),那只是手动调用,不代表上下文管理器机制跑通了——必须走 async with obj as x: 这条路径才算数。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











