8.12. Асинхронные менеджеры контекста

Обзор Python познакомил с менеджерами контекста – объектами, которыми управляют блоки with, где __enter__ выполняется на входе, а __exit__ – на выходе, независимо от того, как завершился блок. Асинхронные менеджеры контекста – это та же идея, перенесённая в asyncio: __aenter__ и __aexit__ являются сопрограммами, поэтому настройка или очистка могут что-то await. Блоком, который ими управляет, является async with.

Мы уже использовали один из них. asyncio.Lock является асинхронным менеджером контекста; страница блокировки использовала его как async with bus_lock: .... Эта страница посвящена написанию такого менеджера для собственного ресурса приложения.

8.12.1. Когда писать свой

Когда у приложения есть ресурс, требующий парных настройки и завершения, и хотя бы одна из этих сторон должна что-то await. Сетевые соединения, датчики, которым нужна задержка на установление после конфигурации, всё, что блокирует что-то на входе и разблокирует на выходе. Обычная синхронная форма with не подходит, потому что её __enter__ и __exit__ не могут быть сопрограммами.

8.12.2. Два метода

  • async def __aenter__(self) выполняется при входе в блок. То, что он возвращает, и есть то, к чему привязывается необязательное предложение as name конструкции async with. Возврат self – самая распространённая форма, но подойдёт любое значение.

  • async def __aexit__(self, exc_type, exc, tb) выполняется при выходе из блока. exc_type равен None при нормальном выходе; при исключении (или отмене) это класс исключения, а exc – его экземпляр. Возврат истинного значения сообщает Python, что исключение было обработано и не должно распространяться. Возврат None (обычный случай) позволяет исключению продолжить движение вверх по цепочке вызовов после выполнения очистки.

8.12.3. Проработанный пример

Обёртка-подсветка, которая включает светодиод на время выполнения тела блока с короткой задержкой на установление, чтобы освещение было стабильным до начала любого захвата, и снова выключает светодиод на выходе:

import asyncio
from machine import LED


class Spotlight:
    def __init__(self, led):
        self._led = led

    async def __aenter__(self):
        self._led.on()
        await asyncio.sleep_ms(50)
        return self

    async def __aexit__(self, exc_type, exc, tb):
        self._led.off()

async def main():
    led = LED("LED_WHITE")

    async with Spotlight(led):
        # work that benefits from steady illumination
        await asyncio.sleep_ms(200)

asyncio.run(main())

При входе в блок выполняется __aenter__: светодиод включается, await 50-миллисекундной паузы на установление уступает управление циклу, чтобы другие сопрограммы тем временем могли продвигаться, и тело блока начинается, как только ожидание завершается. Когда блок заканчивается – при нормальном выходе, при исключении или при отмене – выполняется __aexit__, и светодиод снова выключается. Очистка выполняется в каждом случае; это гарантия, которую предоставляет async with.

The frame capture page shows how to make csi.CSI.snapshot() await-friendly; once that wrapper is in hand, the body of an async with Spotlight(led): block would typically be a capture loop running under the steady illumination.