Python 中的 @asynccontextmanager 裝飾器，它是異步編程中用于快速定義異步上下文管理器的工具，基于 contextlib 模塊實(shí)現(xiàn)，核心作用是讓你用異步生成器（async yield） 替代手動(dòng)定義異步上下文管理器的兩個(gè)魔法方法，大幅簡(jiǎn)化代碼。

先鋪墊：異步上下文管理器的基礎(chǔ)

在異步代碼中，我們會(huì)用 async with 語(yǔ)句管理資源（比如異步連接數(shù)據(jù)庫(kù)、異步打開(kāi)文件、異步網(wǎng)絡(luò)會(huì)話(huà)），而能被 async with 使用的對(duì)象，就是異步上下文管理器。

手動(dòng)定義一個(gè)異步上下文管理器，需要實(shí)現(xiàn)兩個(gè)異步魔法方法：

1. __aenter__()：進(jìn)入 async with 時(shí)執(zhí)行，返回要操作的資源（比如數(shù)據(jù)庫(kù)連接）；
2. __aexit__()：退出 async with 時(shí)執(zhí)行，負(fù)責(zé)釋放資源（比如關(guān)閉連接、處理異常）。

手動(dòng)實(shí)現(xiàn)的缺點(diǎn)是代碼繁瑣，而 @asynccontextmanager 就是為了解決這個(gè)問(wèn)題而生的。

@asynccontextmanager 核心作用

通過(guò)這個(gè)裝飾器，你可以用一個(gè)異步生成器函數(shù)，直接定義出異步上下文管理器的「進(jìn)入邏輯」和「退出邏輯」，無(wú)需手動(dòng)實(shí)現(xiàn) __aenter__ 和 __aexit__：

• 生成器中 yield 之前的代碼 → 對(duì)應(yīng) __aenter__() 的邏輯（進(jìn)入上下文，初始化/獲取資源）；
• 生成器中 yield 之后的代碼 → 對(duì)應(yīng) __aexit__() 的邏輯（退出上下文，釋放/清理資源）；
• yield 后面的值 → 就是 async with ... as 變量 中「變量」接收到的資源對(duì)象。

基本使用語(yǔ)法

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

? 關(guān)鍵：必須用 try...finally 包裹 yield，確保即使上下文內(nèi)代碼拋出異常，資源清理邏輯也能執(zhí)行。

實(shí)際示例：異步文件管理器

用 @asynccontextmanager 實(shí)現(xiàn)一個(gè)簡(jiǎn)單的異步文件操作上下文管理器（模擬異步文件IO，實(shí)際Python3.7+有原生asyncio.open）：

from contextlib import asynccontextmanager
import asyncio
# 定義異步上下文管理器：異步打開(kāi)/關(guān)閉文件
@asynccontextmanager
async def async_open(file_path, mode="r"):
    print("進(jìn)入上下文：異步打開(kāi)文件")
    # 模擬異步打開(kāi)文件（實(shí)際是同步，僅作演示）
    file = open(file_path, mode, encoding="utf-8")
    try:
        yield file  # 返回文件對(duì)象，供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)  # 模擬異步寫(xiě)操作
        f.write("Hello @asynccontextmanager!")
    print("文件操作完成，資源已釋放")
# 運(yùn)行異步代碼
asyncio.run(read_file())

執(zhí)行結(jié)果：

進(jìn)入上下文：異步打開(kāi)文件
退出上下文：異步關(guān)閉文件
文件操作完成，資源已釋放

可以看到：進(jìn)入async with時(shí)執(zhí)行yield前的代碼，退出時(shí)執(zhí)行yield后的清理代碼。

進(jìn)階示例：異步數(shù)據(jù)庫(kù)連接（模擬）

實(shí)際開(kāi)發(fā)中，@asynccontextmanager 最常用在異步數(shù)據(jù)庫(kù)、異步Redis、異步網(wǎng)絡(luò)請(qǐng)求等場(chǎng)景，比如模擬異步MySQL連接：

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

執(zhí)行結(jié)果：

建立異步數(shù)據(jù)庫(kù)連接
執(zhí)行異步SQL：SELECT * FROM user
查詢(xún)結(jié)果：[{'id': 1, 'name': 'test'}]
關(guān)閉異步數(shù)據(jù)庫(kù)連接

無(wú)論query中是否拋出異常，close都會(huì)執(zhí)行，確保數(shù)據(jù)庫(kù)連接被釋放。

關(guān)鍵注意點(diǎn)

1. 裝飾的必須是異步生成器：函數(shù)必須加 async def，且內(nèi)部有 yield（普通生成器不行，普通生成器用@contextmanager）；
2. 異常處理：yield 處可能拋出上下文內(nèi)的異常，必須用try...finally，否則異常會(huì)跳過(guò)清理邏輯；
3. 與同步版的區(qū)別：同步上下文管理器用 @contextmanager（裝飾普通生成器），異步用 @asynccontextmanager（裝飾異步生成器），前者配合with，后者配合async with；
4. 返回值：yield 資源 是可選的，如果上下文不需要返回資源，直接yield即可（async with 管理器: 不帶as）。

對(duì)比：手動(dòng)實(shí)現(xiàn) vs @asynccontextmanager

用手動(dòng)實(shí)現(xiàn)異步上下文管理器的方式重寫(xiě)上面的「異步文件管理器」，對(duì)比代碼量：

import asyncio
# 手動(dòng)實(shí)現(xiàn)異步上下文管理器（需實(shí)現(xiàn)__aenter__/__aexit__）
class AsyncOpen:
    def __init__(self, file_path, mode="r"):
        self.file_path = file_path
        self.mode = mode
        self.file = None
    # 異步進(jìn)入方法
    async def __aenter__(self):
        print("進(jìn)入上下文：異步打開(kāi)文件")
        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("手動(dòng)實(shí)現(xiàn)異步上下文管理器")
asyncio.run(read_file())

可以明顯看到：手動(dòng)實(shí)現(xiàn)需要寫(xiě)大量模板代碼，而@asynccontextmanager用一個(gè)異步生成器就搞定，代碼簡(jiǎn)潔數(shù)倍。

總結(jié)

1. @asynccontextmanager 是 Python contextlib 模塊提供的異步上下文管理器裝飾器，專(zhuān)為異步編程設(shè)計(jì)；
2. 核心是將異步生成器函數(shù)轉(zhuǎn)為異步上下文管理器，yield 前是「進(jìn)入邏輯」，yield 后是「退出清理邏輯」；
3. 必須配合 async with 使用，且生成器內(nèi)要加 try...finally 保證資源必被清理；
4. 對(duì)比手動(dòng)實(shí)現(xiàn) __aenter__/__aexit__，大幅簡(jiǎn)化異步資源管理的代碼，是異步開(kāi)發(fā)中管理臨時(shí)資源（連接、文件、會(huì)話(huà)）的最佳實(shí)踐之一；
5. 同步場(chǎng)景用它的兄弟裝飾器 @contextmanager（配合with，裝飾普通生成器），異步場(chǎng)景用@asynccontextmanager（配合async with，裝飾異步生成器）。

到此這篇關(guān)于Python 中的異步上下文生成器 @asynccontextmanager的文章就介紹到這了,更多相關(guān)Python 異步上下文生成器 @asynccontextmanager內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論