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

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())

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())

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())

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())