如何实现一个支持异步上下文管理的类(aenter/aexit)

Python 3.7+ 异步上下文管理器需实现 async def aenter 和 async def aexit 两个方法,遵循协议而非继承;__aenter__ 返回资源,__aexit__ 接收异常三元组并可选择抑制异常。

Python 3.7+ 支持异步上下文管理器,只需在类中定义 __aenter____aexit__ 两个异步方法,并用 async with 调用即可。

必须实现的两个异步方法

异步上下文管理器不是靠继承某个基类,而是靠协议(protocol):只要对象有可等待的 __aenter____aexit__ 方法,它就是合法的异步上下文管理器。

  • __aenter__ 必须是 async def,返回进入上下文时的值(如连接对象、资源句柄)
  • __aexit__ 必须是 async def,接收异常三元组 (exc_type, exc_val, exc_tb);返回 True 可抑制异常,NoneFalse 表示不处理

一个典型例子:异步数据库连接

模拟一个需要异步初始化和清理的资源:

import asynci

o

class AsyncDBConnection: def init(self, url): self.url = url self._conn = None

async def __aenter__(self):
    # 模拟异步连接建立
    await asyncio.sleep(0.1)
    self._conn = f"conn_to_{self.url}"
    return self._conn  # 这个值会绑定到 as 后面的变量

async def __aexit__(self, exc_type, exc_val, exc_tb):
    # 模拟异步关闭连接
    await asyncio.sleep(0.05)
    self._conn = None
    # 若发生异常,这里可以做清理或记录,但默认不抑制
    return False  # 显式返回 False(也可省略,等价于 None)

使用方式

async def main(): async with AsyncDBConnection("postgresql://...") as conn: print("Connected:", conn)

do something with conn

注意异常传播与资源安全

__aexit__ 中即使发生异常,也不会自动阻止原始异常传播 —— 它本身若抛出异常,会替代原始异常(除非你手动处理)。为保障资源释放,建议:

  • __aexit__ 内部用 try/except 包裹清理逻辑,避免清理失败中断原始错误
  • 不要在 __aexit__raise 新异常,除非有意替换错误上下文
  • 如果清理本身需重试或超时控制,应在该方法内完成,而不是依赖外部逻辑

兼容同步与异步的写法(可选进阶)

若想让一个类同时支持 withasync with,可同时实现两套协议(同步的 __enter__/__exit__ 和异步的 __aenter__/__aexit__),但需注意语义一致。例如:

  • 同步版本应做同步阻塞初始化/清理(如普通 socket.connect)
  • 异步版本对应非阻塞等价操作(如 asyncio.open_connection
  • 两者不应混用;解释器会根据语法(with vs async with)自动选择对应协议