Python異步之上下文管理器怎么使用

本文講解"Python異步之上下文管理器如何使用"，希望能夠解決相關問題。

正文

上下文管理器是一種 Python 構造，它提供了一個類似 try-finally 的環境，具有一致的接口和方便的語法，例如通過“with”表達。

它通常與資源一起使用，確保在我們完成資源后始終關閉或釋放資源，無論資源的使用是成功還是因異常而失敗。

Asyncio 允許我們開發異步上下文管理器。

我們可以通過定義一個將 aenter() 和 aexit() 方法實現為協程的對象來在 asyncio 程序中創建和使用異步上下文管理器。

1. 什么是異步上下文管理器

異步上下文管理器是一個實現了 aenter() 和 aexit() 方法的 Python 對象。

在我們深入了解異步上下文管理器的細節之前，讓我們回顧一下經典的上下文管理器。

1.1. Context Manager

上下文管理器是一個 Python 對象，它實現了 enter() 和 exit() 方法。

• enter() 方法定義了塊開頭發生的事情，例如打開或準備資源，如文件、套接字或線程池。

• exit() 方法定義退出塊時發生的情況，例如關閉準備好的資源。

通過“with”表達式使用上下文管理器。通常，上下文管理器對象是在“with”表達式的開頭創建的，并且會自動調用 enter() 方法。內容的主體通過命名的上下文管理器對象使用資源，然后 aexit() 方法在塊退出時自動調用，通?；蛲ㄟ^異常。

...
# open a context manager
with ContextManager() as manager:
	# ...
# closed automatically

這反映了 try-finally 表達式。

...
# create the object
manager = ContextManager()
try:
	manager.__enter__()
	# ...
finally:
	manager.__exit__()

1.2. Asynchronous Context Manager

“PEP 492 – Coroutines with async and await syntax”引入了異步上下文管理器。

它們提供了一個上下文管理器，可以在進入和退出時掛起。

aenter 和 aexit 方法被定義為協同程序，由調用者等待。這是使用“async with”表達式實現的。

因此，異步上下文管理器只能在 asyncio 程序中使用，例如在調用協程中。

• 什么是“async with”

“async with”表達式用于創建和使用異步上下文管理器。它是“with”表達式的擴展，用于異步程序中的協程。

“async with”表達式就像用于上下文管理器的“with”表達式，除了它允許在協同程序中使用異步上下文管理器。

為了更好地理解“async with”，讓我們仔細看看異步上下文管理器。async with 表達式允許協程創建和使用上下文管理器的異步版本。

...
# create and use an asynchronous context manager
async with AsyncContextManager() as manager:
	# ...

這相當于：

...
# create or enter the async context manager
manager = await AsyncContextManager()
try:
	# ...
finally:
	# close or exit the context manager
	await manager.close()

請注意，我們正在實現與傳統上下文管理器大致相同的模式，只是創建和關閉上下文管理器涉及等待協程。

這會暫停當前協程的執行，調度一個新的協程并等待它完成。因此，異步上下文管理器必須實現必須通過 async def 表達式定義的 aenter() 和 aexit() 方法。這使得它們自己協程也可能等待。

2. 如何使用異步上下文管理器

在本節中，我們將探討如何在我們的 asyncio 程序中定義、創建和使用異步上下文管理器。

2.1. 定義

我們可以將異步上下文管理器定義為實現 aenter() 和 aexit() 方法的 Python 對象。

重要的是，這兩種方法都必須使用“async def”定義為協程，因此必須返回可等待對象。

# define an asynchronous context manager
class AsyncContextManager:
    # enter the async context manager
    async def __aenter__(self):
        # report a message
        print('>entering the context manager')
    # exit the async context manager
    async def __aexit__(self, exc_type, exc, tb):
        # report a message
        print('>exiting the context manager')

因為每個方法都是協程，所以它們本身可能等待協程或任務。

# define an asynchronous context manager
class AsyncContextManager:
    # enter the async context manager
    async def __aenter__(self):
        # report a message
        print('>entering the context manager')
        # block for a moment
        await asyncio.sleep(0.5)
    # exit the async context manager
    async def __aexit__(self, exc_type, exc, tb):
        # report a message
        print('>exiting the context manager')
        # block for a moment
        await asyncio.sleep(0.5)

2.2. 使用

通過“async with”表達式使用異步上下文管理器。這將自動等待進入和退出協程，根據需要暫停調用協程。

...
# use an asynchronous context manager
async with AsyncContextManager() as manager:
	# ...

因此，“async with”表達式和異步上下文管理器更普遍地只能在 asyncio 程序中使用，例如在協程中。

現在我們知道如何使用異步上下文管理器，讓我們看一個有效的例子。

3. 異步上下文管理器和“異步”示例

我們可以探索如何通過“async with”表達式使用異步上下文管理器。

在這個例子中，我們將更新上面的例子，以正常方式使用上下文管理器。

我們將使用“async with”表達式，并在一行中創建并進入上下文管理器。這將自動等待 enter 方法。

然后我們可以在內部塊中使用管理器。在這種情況下，我們將只報告一條消息。

退出內部塊將自動等待上下文管理器的退出方法。將這個例子與前面的例子進行對比，可以看出“async with”表達式在 asyncio 程序中為我們做了多少繁重的工作。

# SuperFastPython.com
# example of an asynchronous context manager via async with
import asyncio
# define an asynchronous context manager
class AsyncContextManager:
    # enter the async context manager
    async def __aenter__(self):
        # report a message
        print('>entering the context manager')
        # block for a moment
        await asyncio.sleep(0.5)
    # exit the async context manager
    async def __aexit__(self, exc_type, exc, tb):
        # report a message
        print('>exiting the context manager')
        # block for a moment
        await asyncio.sleep(0.5)
# define a simple coroutine
async def custom_coroutine():
    # create and use the asynchronous context manager
    async with AsyncContextManager() as manager:
        # report the result
        print(f'within the manager')
# start the asyncio program
asyncio.run(custom_coroutine())

運行示例首先創建 main() 協程并將其用作 asyncio 程序的入口點。

main() 協程運行并在“async with”表達式中創建我們的 AsyncContextManager 類的實例。

該表達式自動調用 enter 方法并等待協程。報告一條消息，協程暫時阻塞。

main() 協程恢復并執行上下文管理器的主體，打印一條消息。

塊退出，自動等待上下文管理器的退出方法，報告消息并休眠片刻。

這突出了 asyncio 程序中異步上下文管理器的正常使用模式。

>entering the context manager
within the manager
>exiting the context manager

關于 "Python異步之上下文管理器如何使用" 就介紹到此。希望多多支持碩編程。