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. Розгорнутий приклад

Обгортка прожектора, що вмикає LED на час виконання блоку, з коротою затримкою стабілізації для сталого освітлення перед будь-якими знімками, та вимикає LED при виході:

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__: LED вмикається, затримка await на 50 мс поступається циклу, щоб інші корутини могли просуватись тим часом, і тіло блоку починається після завершення очікування. Коли блок завершується — нормально, через виключення або через скасування — виконується __aexit__ і LED знову вимикається. Очищення виконується в кожному випадку; це гарантія, яку надає 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.