Python 中的异步上下文生成器 @asynccontextmanager详解_Python

python 中的 @asynccontextmanager 装饰器，它是异步编程中用于快速定义异步上下文管理器的工具，基于 contextlib 模块实现，核心作用是让你用异步生成器（async yield） 替代手动定义异步上下文管理器的两个魔法方法，大幅简化代码。

先铺垫：异步上下文管理器的基础

在异步代码中，我们会用 async with 语句管理资源（比如异步连接数据库、异步打开文件、异步网络会话），而能被 async with 使用的对象，就是异步上下文管理器。

手动定义一个异步上下文管理器，需要实现两个异步魔法方法：

__aenter__()：进入 async with 时执行，返回要操作的资源（比如数据库连接）；
__aexit__()：退出 async with 时执行，负责释放资源（比如关闭连接、处理异常）。

手动实现的缺点是代码繁琐，而 @asynccontextmanager 就是为了解决这个问题而生的。

@asynccontextmanager 核心作用

通过这个装饰器，你可以用一个异步生成器函数，直接定义出异步上下文管理器的「进入逻辑」和「退出逻辑」，无需手动实现 __aenter__ 和 __aexit__：

生成器中 yield 之前的代码 → 对应 __aenter__() 的逻辑（进入上下文，初始化/获取资源）；
生成器中 yield 之后的代码 → 对应 __aexit__() 的逻辑（退出上下文，释放/清理资源）；
yield 后面的值 → 就是 async with ... as 变量 中「变量」接收到的资源对象。

基本使用语法

from contextlib import asynccontextmanager
import asyncio
# 用@asynccontextmanager装饰异步生成器函数
@asynccontextmanager
async def 异步上下文管理器名(参数):
    # 1. 进入逻辑：初始化/获取异步资源（对应__aenter__）
    资源 = 等待异步操作获取资源()
    try:
        yield 资源  # 把资源返回给async with ... as 变量，暂停执行
    finally:
        # 2. 退出逻辑：释放/清理资源（对应__aexit__，无论是否异常都会执行）
        等待异步操作释放资源(资源)
# 调用：用async with语句
async def main():
    async with 异步上下文管理器名(参数) as 资源变量:
        # 操作资源
        await 资源变量.异步方法()
# 运行异步主函数
asyncio.run(main())

✅ 关键：必须用 try...finally 包裹 yield，确保即使上下文内代码抛出异常，资源清理逻辑也能执行。

实际示例：异步文件管理器

用 @asynccontextmanager 实现一个简单的异步文件操作上下文管理器（模拟异步文件io，实际python3.7+有原生asyncio.open）：

from contextlib import asynccontextmanager
import asyncio
# 定义异步上下文管理器：异步打开/关闭文件
@asynccontextmanager
async def async_open(file_path, mode="r"):
    print("进入上下文：异步打开文件")
    # 模拟异步打开文件（实际是同步，仅作演示）
    file = open(file_path, mode, encoding="utf-8")
    try:
        yield file  # 返回文件对象，供async with使用
    finally:
        print("退出上下文：异步关闭文件")
        file.close()  # 清理资源：关闭文件
# 调用异步上下文管理器
async def read_file():
    async with async_open("test.txt", "w") as f:
        await asyncio.sleep(0.1)  # 模拟异步写操作
        f.write("hello @asynccontextmanager!")
    print("文件操作完成，资源已释放")
# 运行异步代码
asyncio.run(read_file())

执行结果：

进入上下文：异步打开文件
退出上下文：异步关闭文件
文件操作完成，资源已释放

可以看到：进入async with时执行yield前的代码，退出时执行yield后的清理代码。

进阶示例：异步数据库连接（模拟）

实际开发中，@asynccontextmanager 最常用在异步数据库、异步redis、异步网络请求等场景，比如模拟异步mysql连接：

from contextlib import asynccontextmanager
import asyncio
# 模拟异步数据库客户端
class asyncmysqlclient:
    async def connect(self):
        print("建立异步数据库连接")
        return self
    async def close(self):
        print("关闭异步数据库连接")
    async def query(self, sql):
        print(f"执行异步sql：{sql}")
        return [{"id": 1, "name": "test"}]
# 定义异步数据库连接的上下文管理器
@asynccontextmanager
async def get_mysql_conn():
    # 1. 进入逻辑：建立异步连接
    client = asyncmysqlclient()
    conn = await client.connect()
    try:
        yield conn  # 返回连接对象，供业务代码使用
    finally:
        # 2. 退出逻辑：关闭连接，释放资源
        await conn.close()
# 业务代码：使用数据库连接
async def query_data():
    async with get_mysql_conn() as conn:
        result = await conn.query("select * from user")
        print(f"查询结果：{result}")
# 运行
asyncio.run(query_data())

执行结果：

建立异步数据库连接
执行异步sql：select * from user
查询结果：[{'id': 1, 'name': 'test'}]
关闭异步数据库连接

无论query中是否抛出异常，close都会执行，确保数据库连接被释放。

关键注意点

装饰的必须是异步生成器：函数必须加 async def，且内部有 yield（普通生成器不行，普通生成器用@contextmanager）；
异常处理：yield 处可能抛出上下文内的异常，必须用try...finally，否则异常会跳过清理逻辑；
与同步版的区别：同步上下文管理器用 @contextmanager（装饰普通生成器），异步用 @asynccontextmanager（装饰异步生成器），前者配合with，后者配合async with；
返回值：yield 资源 是可选的，如果上下文不需要返回资源，直接yield即可（async with 管理器: 不带as）。

对比：手动实现 vs @asynccontextmanager

用手动实现异步上下文管理器的方式重写上面的「异步文件管理器」，对比代码量：

import asyncio
# 手动实现异步上下文管理器（需实现__aenter__/__aexit__）
class asyncopen:
    def __init__(self, file_path, mode="r"):
        self.file_path = file_path
        self.mode = mode
        self.file = none
    # 异步进入方法
    async def __aenter__(self):
        print("进入上下文：异步打开文件")
        self.file = open(self.file_path, self.mode, encoding="utf-8")
        return self.file
    # 异步退出方法（exc_type/val/tb 接收异常信息）
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("退出上下文：异步关闭文件")
        if self.file:
            self.file.close()
# 调用方式不变
async def read_file():
    async with asyncopen("test.txt", "w") as f:
        f.write("手动实现异步上下文管理器")
asyncio.run(read_file())

可以明显看到：手动实现需要写大量模板代码，而@asynccontextmanager用一个异步生成器就搞定，代码简洁数倍。

总结

@asynccontextmanager 是 python contextlib 模块提供的异步上下文管理器装饰器，专为异步编程设计；
核心是将异步生成器函数转为异步上下文管理器，yield 前是「进入逻辑」，yield 后是「退出清理逻辑」；
必须配合 async with 使用，且生成器内要加 try...finally 保证资源必被清理；
对比手动实现 __aenter__/__aexit__，大幅简化异步资源管理的代码，是异步开发中管理临时资源（连接、文件、会话）的最佳实践之一；
同步场景用它的兄弟装饰器 @contextmanager（配合with，装饰普通生成器），异步场景用@asynccontextmanager（配合async with，装饰异步生成器）。