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实现生命周期管理
