Python 如何实现一个支持异步上下文管理的类

13次阅读

要创建可与async with配合的异步上下文管理器,需实现__aenter__和__aexit__两个异步方法,二者均须用async def定义;建议继承asyncio.AsyncContextManager以增强类型安全;必须在协程中使用async with调用,且__aexit__返回值控制异常传播。

Python 如何实现一个支持异步上下文管理的类

如果您希望在 python 中创建一个能够与 async with 语句配合使用的上下文管理器,则必须实现特定的异步协议方法。以下是实现此功能的步骤:

一、实现 __aenter__ 和 __aexit__ 方法

异步上下文管理器需定义两个特殊方法:__aenter__ 返回进入上下文时的资源,__aexit__ 处理退出逻辑(包括异常传播与清理)。这两个方法都必须声明为 async def,并返回协程对象

1、定义一个类,例如 AsyncResource

2、在类中使用 async def 声明 __aenter__ 方法,在其中执行初始化操作(如打开连接、获取锁等),并返回所需资源或 self。

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

3、在类中使用 async def 声明 __aexit__ 方法,接收 exc_type、exc_value、traceback 三个参数;该方法应在内部完成清理工作(如关闭连接、释放锁),并可选择返回 True 以抑制异常。

二、使用 asyncio.sleep 模拟异步初始化与清理

为体现异步特性,可在 __aenter__ 和 __aexit__ 中嵌入真正的异步调用,避免阻塞事件循环。这确保类在高并发场景下仍保持非阻塞性质。

1、在 __aenter__ 方法体中插入 await asyncio.sleep(0.1),模拟异步资源获取延迟。

2、在 __aexit__ 方法体中插入 await asyncio.sleep(0.1),模拟异步资源释放延迟。

3、确保类所在模块已 import asyncio,且调用方在 asyncio.run() 或事件循环中执行 async with 语句。

三、继承 asyncio.AsyncContextManager 抽象基类

从 Python 3.7 开始,标准库提供 asyncio.AsyncContextManager 作为类型提示和结构检查的抽象基类。显式继承它有助于静态分析工具识别合法性,并在未实现必要方法时触发 TypeError。

1、在类定义中添加对 asyncio.AsyncContextManager 的继承,如 class AsyncResource(asyncio.AsyncContextManager):。

2、保持 __aenter__ 和 __aexit__ 的异步签名不变,否则继承将失败。

3、注意:AsyncContextManager 不提供默认实现,仅用于类型标注与运行时协议检查

四、在 async with 中正确调用实例

异步上下文管理器只能在协程函数内通过 async with 语法使用,不可在普通函数或同步代码块中直接调用。Python 解释器会自动调度 __aenter__ 和 __aexit__ 协程。

1、定义一个 async def main() 函数。

2、在该函数中创建类的实例,例如 resource = AsyncResource()。

3、使用 async with resource as r: 结构包裹业务逻辑,其中 r 是 __aenter__ 的返回值。

4、若在非协程环境中使用 async with,将触发 SyntaxError

五、处理异常并控制传播行为

__aexit__ 方法的返回值决定异常是否被抑制。返回 False 或 None 表示不抑制,异常继续向上抛出;返回 True 则阻止异常传播,等效于在 finally 块中加了 pass。

1、在 __aexit__ 方法末尾添加 return False 显式声明不抑制异常。

2、若需忽略特定异常(如 CancelledError),可在方法体内判断 exc_type 并有条件地返回 True。

3、务必在 __aexit__ 中完成所有清理动作,即使发生异常也应确保执行

text=ZqhQzanResources