from contextlib import asynccontextmanager
import asyncio
# 用@asynccontextmanager裝飾異步生成器函數(shù)
@asynccontextmanager
async def 異步上下文管理器名(參數(shù)):
    # 1. 進入邏輯：初始化/獲取異步資源（對應(yīng)__aenter__）
    資源 = 等待異步操作獲取資源()
    try:
        yield 資源  # 把資源返回給async with ... as 變量，暫停執(zhí)行
    finally:
        # 2. 退出邏輯：釋放/清理資源（對應(yīng)__aexit__，無論是否異常都會執(zhí)行）
        等待異步操作釋放資源(資源)
# 調(diào)用：用async with語句
async def main():
    async with 異步上下文管理器名(參數(shù)) as 資源變量:
        # 操作資源
        await 資源變量.異步方法()
# 運行異步主函數(shù)
asyncio.run(main())

from contextlib import asynccontextmanager
import asyncio
# 定義異步上下文管理器：異步打開/關(guān)閉文件
@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("退出上下文：異步關(guān)閉文件")
        file.close()  # 清理資源：關(guān)閉文件
# 調(diào)用異步上下文管理器
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
# 模擬異步數(shù)據(jù)庫客戶端
class AsyncMySQLClient:
    async def connect(self):
        print("建立異步數(shù)據(jù)庫連接")
        return self
    async def close(self):
        print("關(guān)閉異步數(shù)據(jù)庫連接")
    async def query(self, sql):
        print(f"執(zhí)行異步SQL：{sql}")
        return [{"id": 1, "name": "test"}]
# 定義異步數(shù)據(jù)庫連接的上下文管理器
@asynccontextmanager
async def get_mysql_conn():
    # 1. 進入邏輯：建立異步連接
    client = AsyncMySQLClient()
    conn = await client.connect()
    try:
        yield conn  # 返回連接對象，供業(yè)務(wù)代碼使用
    finally:
        # 2. 退出邏輯：關(guān)閉連接，釋放資源
        await conn.close()
# 業(yè)務(wù)代碼：使用數(shù)據(jù)庫連接
async def query_data():
    async with get_mysql_conn() as conn:
        result = await conn.query("SELECT * FROM user")
        print(f"查詢結(jié)果：{result}")
# 運行
asyncio.run(query_data())

import asyncio
# 手動實現(xiàn)異步上下文管理器（需實現(xiàn)__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("退出上下文：異步關(guān)閉文件")
        if self.file:
            self.file.close()
# 調(diào)用方式不變
async def read_file():
    async with AsyncOpen("test.txt", "w") as f:
        f.write("手動實現(xiàn)異步上下文管理器")
asyncio.run(read_file())