Ответ
В 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())Способ с декоратором часто предпочтительнее из-за его простоты и читаемости.
Ответ 18+ 🔞
Слушай, а вот эти ваши асинхронные контекстные менеджеры в Питоне — это ж, блядь, просто песня! Как будто в баню зашёл, всё по полочкам разложили, а потом ещё и убрали за собой, не оставив ни хуя мусора. Вообще красота, ебать мои старые костыли!
Ну, способов-то их создать, по сути, два. И оба — охуенные, но для разных случаев.
1. Классика жанра: через класс, как взрослые дяди
Тут надо, чтобы твой класс был не просто так, а с двумя специальными асинхронными методами. Представь, что ты — швейцар в дорогом отеле.
async def __aenter__(self): Это когда гость (твой код) подъезжает. Ты ему: «Добро пожаловать, ёпта!», открываешь дверь (await), помогаешь с багажом и говоришь: «Вот ваш ключ-кабинет (возвращаешь объект)».async def __aexit__(self, exc_type, exc_val, exc_tb): А это когда гость выезжает. Даже если он в номере люстру разбил или в сортире потоп устроил (исключение), ты всё равно заходишь, убираешь всё за ним (finally), сдаёшь номер и вежливо провожаешь.
Смотри, как это выглядит в коде:
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) # Прикидываемся, что ходим в сеть, а не просто спим
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 — это твой __aenter__. Всё, что после — это __aexit__. А сам yield — это то, что ты суёшь в переменную после as. И самое охуенное — finally тут неявно, всё само закроется, даже если посередине пиздец случится.
Вот, смотри, та же история, но в два раза короче:
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 # Вот эту хуйню отдаём во внешний мир
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())Честно? В 90% случаев берут именно этот способ. Меньше писанины, проще читать, и не нужно городить целый класс, если тебе просто на пару строк кода обёртку сделать.
В общем, выбирай, что душе угодно. Хочешь фундаментально — класс, хочешь быстро и без геморроя — декоратор. Главное — ресурсы закрывать, а то получится как в том анекдоте про Герасима и Муму, только с сокетами.