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

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

• Python, асинхронное программирование, контекстные менеджеры
• Реализовать методы aenter и aexit
• aenter — асинхронно выполняет вход в контекст (поддерживает await)
• aexit — асинхронно завершает работу (поддерживает await)
• Эти методы должны быть реализованы классом или объектом
• Менеджер предназначен для работы с ресурсами через async/await
• Подходит для безопасного асинхронного управления файлами, сетевыми соединениями и другими ресурсами

Развернутый ответ

Основной ответ

В Python асинхронный контекстный менеджер реализуется через методы __aenter__ и __aexit__, отвечающие соответственно за вход в асинхронный контекст и выход из него. В отличие от стандартной пары методов (__enter__/__exit__), они представляют собой корутины и поэтому поддерживают await. Такой объект используется вместе с оператором async with, благодаря чему асинхронные операции можно выполнять при входе в блок и после его завершения.

Основные моменты

• Реализация методов: __aenter__(self) — асинхронно выполняется при входе в контекст и обычно подготавливает необходимые ресурсы; __aexit__(self, exc_type, exc, tb) — асинхронно вызывается при выходе, освобождает ресурсы и может обрабатывать исключения.
• Использование: Такой подход применяют для асинхронной работы с ресурсами: например, подключения к базе данных (в asyncpg), файлами или сетевыми сессиями.
• Поддержка: Асинхронные контекстные менеджеры доступны в Python 3.5+, а в стандартной библиотеке в качестве примеров используются asyncio.Lock и aiofiles.

Практический пример

Распространенный сценарий — асинхронно установить сетевое соединение, а затем закрыть его, дождавшись выполнения операций внутри блока async with. Это делает код понятнее и помогает управлять ресурсами, не блокируя поток.

class AsyncContextManager:
async def __aenter__(self):
print("Подключение...")
await asyncio.sleep(1)  # эмуляция async действия
print("Подключено")
return self

async def __aexit__(self, exc_type, exc, tb):
print("Отключение...")
await asyncio.sleep(1)
print("Отключено")

# Использование:
# async with AsyncContextManager() as manager:
#     # асинхронная работа

Этот способ характерен для современных приложений на asyncio: он позволяет сохранять отзывчивость event loop и корректно контролировать жизненный цикл используемых ресурсов.