python中实现异步上下文管理器需用@asynccontextmanager装饰器或自定义类实现__aenter__和__aexit__方法，不可混用同步装饰器与异步函数。

Python 的 @contextmanager 装饰器本身不支持异步上下文管理（即不能直接用于 async with），因为它返回的是同步的生成器。要让上下文管理器支持 async with，必须使用 async def 定义，并实现 __aenter__ 和 __aexit__ 方法——也就是自定义异步上下文管理器类。

用类实现异步上下文管理器

这是最标准、最可控的方式。你需要定义一个类，手动实现两个异步方法：

• __aenter__：在 async with 进入时被 await，可执行异步初始化（如连接数据库、获取锁）
• __aexit__：在 async with 结束时被 await，可执行异步清理（如关闭连接、提交/回滚事务）

示例：

class AsyncDBConnection:     def __init__(self, url):         self.url = url         self.conn = None 
async def __aenter__(self):     self.conn = await aiomysql.connect(host=self.url)     return self.conn  async def __aexit__(self, exc_type, exc_val, exc_tb):     if self.conn:         await self.conn.close()


使用方式：
 

async with AsyncDBConnection("localhost") as conn:     result = await conn.execute("SELECT 1")

用 asynccontextmanager（推荐）
Python 3.7+ 的 contextlib 提供了 @asynccontextmanager 装饰器，语法上最接近 @contextmanager，但专为异步设计。它把异步生成器自动包装成符合 AsyncContextManager 协议的对象。
注意：需从 contextlib 导入，不是 asyncio：
 

from contextlib import asynccontextmanager 
@asynccontextmanager async def lifespan(app):
启动前：异步初始化
database = await connect_db() cache = await init_cache() try:     yield {"db": database, "cache": cache} finally:     # 关闭后：异步清理     await database.close()     await cache.shutdown()


使用方式相同：
 

async with lifespan(my_app) as resources:     await do_something(resources["db"])

不能混用 sync 和 async
以下写法是错误的：

用 @contextmanager + async def —— 会报 TypeError: @contextmanager only works with sync generators 
在 __aenter__ 中返回普通（非 awaitable）对象 —— 必须返回 awaitable，否则 async with 会失败
忘记在 __aexit__ 中处理异常或加 await —— 清理逻辑不会真正执行

常见适配场景
很多同步库（如 requests、sqlite3）没有原生异步支持。若想封装其资源为异步上下文管理器，需借助线程池或专用异步库：

对阻塞 I/O 操作（如文件读写、http 请求），可用 loop.run_in_executor 包装，再放入 @asynccontextmanager 
优先选用已有的异步生态库（如 aiohttp、aiomysql、asyncpg），它们通常自带异步上下文管理器
fastapi 的 Depends、Starlette 的 Lifespan 等都依赖 @asynccontextmanager 实现生命周期管理