Python21_装饰器AsyncContextManager

from contextlib import asynccontextmanager 是 Python 中用于创建异步上下文管理器的工具。

核心作用

asynccontextmanager 是一个装饰器，让你可以用异步生成器函数（async def + yield）来创建上下文管理器，而不需要编写完整的类。

对比：传统 vs 现代写法

❌ 传统写法（繁琐）

from contextlib import AbstractAsyncContextManager

class MyAsyncContext(AbstractAsyncContextManager):
    async def __enter__(self):
        await setup()
        return self
    
    async def __exit__(self, exc_type, exc, tb):
        await cleanup()

✅ 现代写法（简洁）

from contextlib import asynccontextmanager

@asynccontextmanager
async def my_async_context():
    await setup()      # 进入时执行（相当于 __enter__）
    yield resource     # 返回资源给 with 语句
    await cleanup()    # 退出时执行（相当于 __exit__）

使用示例

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def database_connection():
    print("🔗 连接数据库...")
    await asyncio.sleep(1)  # 模拟异步连接
    conn = {"status": "connected", "id": 123}
    try:
        yield conn  # 将连接对象交给 with 块使用
    finally:
        print("🔌 关闭数据库连接...")
        await asyncio.sleep(0.5)  # 模拟异步关闭

async def main():
    async with database_connection() as conn:
        print(f"✅ 使用连接: {conn}")
        # 这里可以执行数据库操作

asyncio.run(main())

输出：

🔗 连接数据库...
✅ 使用连接: {'status': 'connected', 'id': 123}
🔌 关闭数据库连接...

关键特性

异常处理示例

@asynccontextmanager
async def safe_operation():
    print("开始")
    try:
        yield
    except Exception as e:
        print(f"捕获异常: {e}")
        raise  # 可以选择重新抛出或处理
    finally:
        print("清理资源")

适用场景

• 🗄️ 异步数据库连接（aiomysql, asyncpg）
• 🌐 异步 HTTP 客户端（aiohttp, httpx）
• 🔒 异步锁/信号量
• 🚀 FastAPI lifespan 事件管理

FastAPI 实际应用

这是最常见的使用场景之一：

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时：加载模型、连接数据库
    print("🚀 应用启动...")
    yield
    # 关闭时：清理资源
    print("🛑 应用关闭...")

app = FastAPI(lifespan=lifespan)

总结

asynccontextmanager 让你用最简洁的语法实现健壮的异步资源管理，是异步 Python 编程的必备工具。