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

Слушай, а вот эти ваши асинхронные контекстные менеджеры в Питоне — это ж, блядь, просто песня! Как будто в баню зашёл, всё по полочкам разложили, а потом ещё и убрали за собой, не оставив ни хуя мусора. Вообще красота, ебать мои старые костыли!

Ну, способов-то их создать, по сути, два. И оба — охуенные, но для разных случаев.

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% случаев берут именно этот способ. Меньше писанины, проще читать, и не нужно городить целый класс, если тебе просто на пару строк кода обёртку сделать.

В общем, выбирай, что душе угодно. Хочешь фундаментально — класс, хочешь быстро и без геморроя — декоратор. Главное — ресурсы закрывать, а то получится как в том анекдоте про Герасима и Муму, только с сокетами.