如何使用contextlib和asyncio在异步Python代码中正确管理上下文与资源清理？

嘿，我刚好也踩过类似的坑，咱们一步步来拆解怎么用asyncio和contextlib搞定异步资源的上下文管理，确保就算遇到异常或者任务被取消，资源也能被稳妥清理。

一、自定义异步上下文管理器类 #

首先，你写的AsyncResource类其实已经是个标准的异步上下文管理器了——只要实现__aenter__和__aexit__这两个异步方法就行：

• __aenter__：负责异步获取资源，完成后返回资源实例（或者你需要的对象）
• __aexit__：不管代码块正常结束、抛出异常还是任务被取消，都会执行这里的逻辑，用来异步释放资源

咱们把你的代码补全，再加上调用示例：

import asyncio

class AsyncResource:
    async def __aenter__(self):
        print("Acquiring resource")
        # 模拟资源获取的异步操作
        await asyncio.sleep(1)
        return self

    async def __aexit__(self, exc_type, exc, tb):
        print("Releasing resource")
        # 模拟资源释放的异步操作
        await asyncio.sleep(1)
        # 如果需要处理特定异常，可以在这里判断exc_type
        # 比如：if exc_type is ValueError: 做对应的异常处理

async def use_resource():
    # 用async with来管理异步资源
    async with AsyncResource() as resource:
        print("Using resource")
        # 模拟使用资源的操作，这里可以放开注释测试异常场景
        # raise ValueError("Oops, something went wrong!")
        await asyncio.sleep(2)

asyncio.run(use_resource())

你可以把注释里的异常放开试试，哪怕代码块抛出错误，__aexit__里的资源释放逻辑依然会执行。

二、用contextlib.asynccontextmanager简化实现 #

如果你觉得写类太繁琐，contextlib提供的asynccontextmanager装饰器可以让你用异步生成器的方式快速实现上下文管理器，代码会简洁很多：

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def async_resource_manager():
    # 这里是获取资源的逻辑，对应类方式的__aenter__
    print("Acquiring resource via generator")
    await asyncio.sleep(1)
    resource = "My Async Resource"  # 这里可以替换成实际的资源实例
    
    try:
        yield resource  # yield把资源传递给async with块使用
    finally:
        # 这里是清理资源的逻辑，对应类方式的__aexit__，不管有没有异常都会执行
        print("Releasing resource via generator")
        await asyncio.sleep(1)

async def use_generator_resource():
    async with async_resource_manager() as resource:
        print(f"Using {resource}")
        # 同样可以放开注释测试异常：raise RuntimeError("Test error")
        await asyncio.sleep(2)

asyncio.run(use_generator_resource())

这个方式的核心是try-finally：yield之前的代码负责获取资源，yield把资源传递给async with块，finally里的代码负责清理，不管代码块正常结束还是出问题，finally都会执行。

关键注意点 #

• 必须在异步函数里使用async with，不能在同步代码里直接调用
• 如果你的资源清理逻辑可能抛出异常，记得在__aexit__或者finally块里处理，避免清理失败
• 针对任务被取消（asyncio.CancelledError）的场景，异步上下文管理器也能正常处理——因为取消本质也是一种异常，__aexit__或者finally依然会执行

两种方式怎么选？ #

• 如果你需要复杂的资源管理逻辑（比如资源的状态跟踪、多个子资源的管理），用类的方式更灵活
• 如果只是简单的“获取-使用-清理”流程，用asynccontextmanager装饰器的生成器方式更简洁

备注：内容来源于stack exchange，提问作者Russell