Ответ
В 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())
Способ с декоратором часто предпочтительнее из-за его простоты и читаемости.