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

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

正文

上下文管理器是一種 Python 構(gòu)造，它提供了一個(gè)類似 try-finally 的環(huán)境，具有一致的接口和方便的語(yǔ)法，例如通過(guò)“with”表達(dá)。

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

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

我們可以通過(guò)定義一個(gè)將 aenter() 和 aexit() 方法實(shí)現(xiàn)為協(xié)程的對(duì)象來(lái)在 asyncio 程序中創(chuàng)建和使用異步上下文管理器。

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

異步上下文管理器是一個(gè)實(shí)現(xiàn)了 aenter() 和 aexit() 方法的 Python 對(duì)象。

在我們深入了解異步上下文管理器的細(xì)節(jié)之前，讓我們回顧一下經(jīng)典的上下文管理器。

1.1. Context Manager

上下文管理器是一個(gè) Python 對(duì)象，它實(shí)現(xiàn)了 enter() 和 exit() 方法。

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

• exit() 方法定義退出塊時(shí)發(fā)生的情況，例如關(guān)閉準(zhǔn)備好的資源。

通過(guò)“with”表達(dá)式使用上下文管理器。通常，上下文管理器對(duì)象是在“with”表達(dá)式的開頭創(chuàng)建的，并且會(huì)自動(dòng)調(diào)用 enter() 方法。內(nèi)容的主體通過(guò)命名的上下文管理器對(duì)象使用資源，然后 aexit() 方法在塊退出時(shí)自動(dòng)調(diào)用，通?；蛲ㄟ^(guò)異常。

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

這反映了 try-finally 表達(dá)式。

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

1.2. Asynchronous Context Manager

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

它們提供了一個(gè)上下文管理器，可以在進(jìn)入和退出時(shí)掛起。

aenter 和 aexit 方法被定義為協(xié)同程序，由調(diào)用者等待。這是使用“async with”表達(dá)式實(shí)現(xiàn)的。

因此，異步上下文管理器只能在 asyncio 程序中使用，例如在調(diào)用協(xié)程中。

• 什么是“async with”

“async with”表達(dá)式用于創(chuàng)建和使用異步上下文管理器。它是“with”表達(dá)式的擴(kuò)展，用于異步程序中的協(xié)程。

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

為了更好地理解“async with”，讓我們仔細(xì)看看異步上下文管理器。async with 表達(dá)式允許協(xié)程創(chuàng)建和使用上下文管理器的異步版本。

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

這相當(dāng)于：

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

請(qǐng)注意，我們正在實(shí)現(xiàn)與傳統(tǒng)上下文管理器大致相同的模式，只是創(chuàng)建和關(guān)閉上下文管理器涉及等待協(xié)程。

這會(huì)暫停當(dāng)前協(xié)程的執(zhí)行，調(diào)度一個(gè)新的協(xié)程并等待它完成。因此，異步上下文管理器必須實(shí)現(xiàn)必須通過(guò) async def 表達(dá)式定義的 aenter() 和 aexit() 方法。這使得它們自己協(xié)程也可能等待。

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

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

2.1. 定義

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

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

# 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')

因?yàn)槊總€(gè)方法都是協(xié)程，所以它們本身可能等待協(xié)程或任務(wù)。

# 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. 使用

通過(guò)“async with”表達(dá)式使用異步上下文管理器。這將自動(dòng)等待進(jìn)入和退出協(xié)程，根據(jù)需要暫停調(diào)用協(xié)程。

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

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

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

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

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

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

我們將使用“async with”表達(dá)式，并在一行中創(chuàng)建并進(jìn)入上下文管理器。這將自動(dòng)等待 enter 方法。

然后我們可以在內(nèi)部塊中使用管理器。在這種情況下，我們將只報(bào)告一條消息。

退出內(nèi)部塊將自動(dòng)等待上下文管理器的退出方法。將這個(gè)例子與前面的例子進(jìn)行對(duì)比，可以看出“async with”表達(dá)式在 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())

運(yùn)行示例首先創(chuàng)建 main() 協(xié)程并將其用作 asyncio 程序的入口點(diǎn)。

main() 協(xié)程運(yùn)行并在“async with”表達(dá)式中創(chuàng)建我們的 AsyncContextManager 類的實(shí)例。

該表達(dá)式自動(dòng)調(diào)用 enter 方法并等待協(xié)程。報(bào)告一條消息，協(xié)程暫時(shí)阻塞。

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

塊退出，自動(dòng)等待上下文管理器的退出方法，報(bào)告消息并休眠片刻。

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

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

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