Python 上下文管理器在异步中的等价物

2次阅读

async with 是异步上下文管理器的唯一语法,要求对象实现 __aenter__ 和 __aexit__(非同步版),常见错误包括漏写 a 前缀、未用 await、参数缺失或返回值错误;第三方库如 asynccontextmanager 仅封装协议,不绕过实现;需注意嵌套退出顺序、取消风险、性能开销及测试须用异步函数。

Python 上下文管理器在异步中的等价物

async with 是 async context manager 的唯一入口

pythonasync with 就是异步上下文管理器的语法糖,没有别的替代写法。它要求对象实现 __aenter____aexit__,而不是同步的 __enter__/__exit__。漏掉 a 前缀是新手最常犯的错——写了 __enter__ 却用 async with,直接报 TypeError: Object does not support async context manager protocol

常见错误现象:RuntimeWarning: coroutine 'MyClass.__aenter__' was never awaited,说明你返回了协程但没在 __aenter__ 里用 await;或者 __aexit__ 返回了非布尔值(比如忘了加 return False)导致异常吞没。

  • 必须用 async def __aenter__(self),不能是普通 def
  • __aexit__ 必须带三个参数:exc_type, exc_value, traceback,且需显式返回 True 才抑制异常
  • 类不能同时靠 __enter__ 支撑 with__aenter__ 支撑 async with——得各自实现

aiostream、asyncstdlib 等第三方库不解决协议问题

有人想用 asyncstdlib.contextlibaiostream.stream 绕过手写协议,但这些只是工具集,不是魔法。它们内部照样依赖 __aenter__/__aexit__。比如 asynccontextmanager 装饰器,本质就是帮你生成符合协议的类实例,底层仍是那两个方法。

使用场景:快速包装一个异步初始化+清理逻辑,比如打开异步数据库连接再关闭。

立即学习Python免费学习笔记(深入)”;

  • @asynccontextmanager 要求被装饰函数是 async def,且 yield 前后都得是 awaitable
  • yield 后的代码在 __aexit__ 中执行,如果这里抛异常,会覆盖原异常,除非显式处理
  • 别在 asynccontextmanager 里调用同步阻塞操作(如 time.sleep),否则整个 Event loop 被卡住

嵌套 async with 的顺序和取消风险

多个 async with 嵌套时,退出顺序是倒序——最后进入的最先退出。这点和同步 with 一致,但异步下容易被忽略的是:如果某个 __aexit__ 内部 await 耗时长,或被 cancel,会影响外层退出时机。

性能影响:每个 async with 块都会增加一次协程调度开销,高频短生命周期资源(比如每次 http 请求都新建一个 session)不如复用对象 + 显式 await session.close()

  • 避免在 __aexit__ 里做重试或网络请求,失败就失败,别拖慢整体退出
  • 若资源关闭本身可能超时(如远端服务无响应),应在 __aexit__ 中加 asyncio.wait_for 并捕获 TimeoutError
  • 不要指望 async with 自动处理 task cancellation——__aexit__ 被中断时,不会自动回滚已做的清理

测试 async context manager 得用 async test 方法

pytestunittest 测异步上下文管理器时,测试函数本身必须是 async def,且要用 await 进入上下文。写成普通函数直接调 async with 会报 SyntaxError: 'async with' outside async function

可给出简短示例:

async def test_db_connection():     async with DatabasePool() as db:         result = await db.fetch("SELECT 1")         assert result == [(1,)]
  • pytest 需安装 pytest-asyncio 插件,并标记测试函数为 @pytest.mark.asyncio
  • 别在测试里用 loop.run_until_complete 手动驱动——现代测试框架已封装好 event loop 生命周期
  • mock 异步方法时,用 AsyncMock(Python 3.8+)或 asynctest,普通 Mock 不会 await

事情说清了就结束。真正难的不是写对协议,而是判断哪些资源值得做成 async context manager——比如一个只做一次 await 的连接池,其实没必要;而涉及多次 IO、状态切换、错误恢复的,才值得投入。

text=ZqhQzanResources