Как работает механизм асинхронных контекстных менеджеров в Python?

Асинхронные контекстные менеджеры в Python (введенные в Python 3.7 с помощью библиотеки asyncio) позволяют безопасно и удобно управлять асинхронными ресурсами, такими как асинхронные соединения, асинхронные файловые операции или любые другие объекты, требующие асинхронной инициализации и завершения.

Основной механизм заключается в использовании двух специальных методов:

• __aenter__(self): Этот метод вызывается при входе в блок async with. Он должен быть корутиной (т.е., определен с помощью async def). Он выполняет асинхронные операции настройки, такие как установление соединения или открытие файла. Он может возвращать значение, которое будет присвоено переменной, указанной после as в операторе async with. Если он вызывает исключение, блок async with не будет выполнен.
• __aexit__(self, exc_type, exc_val, exc_tb): Этот метод вызывается при выходе из блока async with, независимо от того, был ли он завершен нормально или с исключением. Он также должен быть корутиной. Он выполняет асинхронные операции завершения, такие как закрытие соединения или закрытие файла. Принимает три аргумента:
• exc_type: Тип исключения, если таковое возникло, иначе None.
• exc_val: Экземпляр исключения, если таковое возникло, иначе None.
• exc_tb: Трассировка стека исключения, если таковое возникло, иначе None.

Если __aexit__ возвращает True, исключение, если оно было, подавляется (не распространяется дальше). Если возвращается None или любое другое значение, исключение распространяется как обычно.

Пример:

import asyncio

class AsyncResource:
    async def open(self):
        print("Opening resource...")
        await asyncio.sleep(0.1)  # Simulate async operation
        print("Resource opened.")

    async def close(self):
        print("Closing resource...")
        await asyncio.sleep(0.1) # Simulate async operation
        print("Resource closed.")

class AsyncContextManager:
    def __init__(self, resource):
        self.resource = resource

    async def __aenter__(self):
        print("Entering context...")
        await self.resource.open()
        return self.resource # Возвращаем ресурс, который будет доступен внутри блока async with

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Exiting context...")
        await self.resource.close()
        print("Context exited.")

async def main():
    resource = AsyncResource()
    async with AsyncContextManager(resource) as res:
        print("Inside the async with block...")
        # Используем ресурс res (который равен resource)
        await asyncio.sleep(0.2)
        print("Doing something with the resource...")


    print("After the async with block.")

if __name__ == "__main__":
    asyncio.run(main())
  

Ключевые моменты:

• Асинхронность: Все операции внутри __aenter__ и __aexit__ должны быть асинхронными.
• Обработка исключений: __aexit__ обеспечивает надежную обработку исключений, гарантируя освобождение ресурсов даже в случае ошибок.
• Гарантированное освобождение ресурсов: Независимо от того, что происходит внутри блока async with, __aexit__ всегда будет вызван для очистки.
• Использование: Асинхронные контекстные менеджеры используются для управления ресурсами в асинхронном коде, обеспечивая более чистый и безопасный код, чем ручное управление ресурсами.