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

«Как в Python создать асинхронный контекстный менеджер» — вопрос из категории Асинхронность, который задают на 10% собеседований Python Разработчик. Ниже — развёрнутый ответ с разбором ключевых моментов.

Ответ

В Python асинхронный контекстный менеджер, используемый с оператором async with, можно создать двумя основными способами.

1. С помощью класса и методов __aenter__ и __aexit__

Это фундаментальный способ. Класс должен реализовать два специальных асинхронных метода:

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

Пример:

import asyncio

class AsyncDatabaseConnection:
    def __init__(self, db_name):
        self._db_name = db_name

    async def __aenter__(self):
        print(f"Подключение к БД '{self._db_name}'...")
        await asyncio.sleep(0.5) # Имитация I/O операции
        print("Подключение установлено.")
        return self # Возвращаем сам объект для работы

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print(f"Закрытие подключения к БД '{self._db_name}'...")
        await asyncio.sleep(0.5)
        print("Подключение закрыто.")

async def main():
    async with AsyncDatabaseConnection("my_db") as conn:
        print("Выполнение запросов...")

asyncio.run(main())

2. С помощью декоратора @asynccontextmanager

Это более лаконичный способ из модуля contextlib, основанный на асинхронном генераторе. Код до yield выполняется при входе в контекст, а код после yield — при выходе.

Пример:

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def async_db_connection(db_name):
    print(f"Подключение к БД '{db_name}'...")
    await asyncio.sleep(0.5)
    try:
        yield db_name # Значение, которое получит переменная после 'as'
    finally:
        print(f"Закрытие подключения к БД '{db_name}'...")
        await asyncio.sleep(0.5)

async def main():
    async with async_db_connection("my_db") as db:
        print(f"Работа с БД: {db}")

asyncio.run(main())

Способ с декоратором часто предпочтительнее из-за его простоты и читаемости.