python之asynccontextmanager学习-编程实验室

一、asynccontextmanager的作用

asynccontextmanager是 Python 标准库contextlib模块提供的异步上下文管理器装饰器，专门用于简化异步场景下上下文管理器的实现。

它的核心特性：

装饰异步生成器函数（使用async def定义，包含yield关键字）
自动封装异步上下文管理器的__aenter__()和__aexit__()方法（无需手动实现这两个魔法方法）
配合async with语句使用，实现异步资源的自动获取与释放

二、适用场景

当你在异步代码中需要管理「需要先获取、使用后必须释放 / 清理」的资源时，asynccontextmanager是最优选择，典型场景包括：

异步网络连接：异步数据库连接（如asyncmy操作 MySQL）、异步 Redis 连接、异步 HTTP 客户端连接，自动建立 / 关闭连接
异步文件操作：异步读写文件（如aiofiles），自动打开 / 关闭文件句柄
异步锁 / 信号量：异步编程中的asyncio.Lock，自动获取 / 释放锁，避免死锁
异步资源初始化 / 销毁：如 FastAPI 的异步应用生命周期管理（启动时初始化资源，关闭时清理资源）

简单来说：异步场景 + 资源需要自动管理（获取 + 释放），就用asynccontextmanager。

三、示例

示例中会用到异步文件操作库aiofiles（演示异步资源管理），先安装：

pip install aiofiles

完整代码

# 1. 导入必要模块importasyncioimportaiofilesfromcontextlibimportasynccontextmanager# 2. 使用@asynccontextmanager装饰异步生成器，实现异步上下文管理器@asynccontextmanagerasyncdefasync_file_manager(file_path:str,mode:str="r",encoding:str="utf-8"):""" 异步文件管理器：自动打开/关闭文件，封装异步文件操作的资源管理 :param file_path: 文件路径 :param mode: 文件打开模式（r/w/a等） :param encoding: 文件编码 """# -------- 对应__aenter__()方法：获取/初始化资源（支持await异步操作） --------file_handle=Nonetry:# 异步打开文件（aiofiles提供异步文件操作，需要await）file_handle=awaitaiofiles.open(file_path,mode=mode,encoding=encoding)print(f"✅ 异步打开文件：{file_path}（模式：{mode}）")# 关键：yield资源给async with语句使用（类似return，但会保留函数状态）yieldfile_handle# -------- 对应__aexit__()方法：释放/清理资源（无论是否发生异常，都会执行） --------finally:iffile_handle:# 异步关闭文件（需要await）awaitfile_handle.close()print(f"❌ 异步关闭文件：{file_path}")# 3. 使用异步上下文管理器（配合async with语句）asyncdefmain():# 示例1：异步读取文件（先创建一个测试文件test.txt，写入一些内容）test_file_path="test.txt"# 第一步：先异步写入测试内容（使用async_file_manager）asyncwithasync_file_manager(test_file_path,mode="w")asf:awaitf.write("Hello, asynccontextmanager!\n")awaitf.write("这是异步文件写入测试\n")# 第二步：异步读取文件内容（再次使用async_file_manager）asyncwithasync_file_manager(test_file_path,mode="r")asf:content=awaitf.read()print("\n📄 文件内容：")print(content)# 4. 运行异步主函数if__name__=="__main__":asyncio.run(main())

四、关键细节解析

异步生成器函数的结构

@asynccontextmanager装饰的函数必须满足：

用async def定义（异步函数）
包含且仅包含一个yield关键字（分割「资源获取」和「资源释放」逻辑）
yield之前的代码：在进入async with时执行（对应__aenter__），负责获取 / 初始化资源
yield之后的代码：在离开async with时执行（对应__aexit__），通常放在finally块中，确保无论是否发生异常（如文件读写报错），都能释放资源

async with的使用语法

asyncwith异步上下文管理器(参数)as资源变量:# 操作资源（支持await异步操作）await资源变量.xxx()

异常处理（可选扩展）

@asynccontextmanagerasyncdefasync_file_manager(file_path:str,mode:str="r",encoding:str="utf-8"):file_handle=Nonetry:file_handle=awaitaiofiles.open(file_path,mode=mode,encoding=encoding)print(f"✅ 异步打开文件：{file_path}（模式：{mode}）")yieldfile_handle# 此处抛出的异常会被传递到async with代码块exceptExceptionase:print(f"❌ 操作文件出错：{e}")raise# 可选：重新抛出异常，让调用方处理finally:iffile_handle:awaitfile_handle.close()print(f"❌ 异步关闭文件：{file_path}")

五、扩展场景：FastAPI 异步生命周期管理

asynccontextmanager在 Web 框架中应用广泛，例如 FastAPI 中用于管理应用的异步生命周期：

fromfastapiimportFastAPIfromcontextlibimportasynccontextmanagerimportasyncio# 定义异步生命周期管理器@asynccontextmanagerasyncdefapp_lifespan(app:FastAPI):# 应用启动时：初始化异步资源（如异步数据库连接池）print("🚀 FastAPI应用启动，初始化异步资源...")awaitasyncio.sleep(1)# 模拟异步初始化操作yield# 应用运行中，传递控制权# 应用关闭时：清理异步资源（如关闭数据库连接池）print("🛑 FastAPI应用关闭，清理异步资源...")awaitasyncio.sleep(1)# 模拟异步清理操作# 传入生命周期管理器创建FastAPI应用app=FastAPI(lifespan=app_lifespan)# 定义接口@app.get("/")asyncdefroot():return{"message":"Hello, FastAPI + asynccontextmanager!"}