Делал ли асинхронный контекстный менеджер на проекте

Да, неоднократно

Асинхронные контекстные менеджеры — это мощный инструмент для управления ресурсами в асинхронном коде. Я использовал их во множестве реальных проектов.

Практический пример: асинхронное подключение к БД

Один из самых частых кейсов — управление асинхронным подключением к базе данных:

from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

class DatabaseManager:
    def __init__(self, database_url: str):
        self.engine = create_async_engine(
            database_url,
            echo=False,
            pool_size=10,
            max_overflow=20
        )
        self.async_session = sessionmaker(
            self.engine,
            class_=AsyncSession,
            expire_on_commit=False
        )

    @asynccontextmanager
    async def get_session(self):
        async with self.async_session() as session:
            try:
                yield session
                await session.commit()
            except Exception:
                await session.rollback()
                raise
            finally:
                await session.close()

Так я использовал его в проекте:

db_manager = DatabaseManager("postgresql+asyncpg://user:pass@localhost/db")

async def get_user(user_id: int):
    async with db_manager.get_session() as session:
        user = await session.get(User, user_id)
        return user

Пример: управление HTTP сессией

Второй кейс — управление aiohttp сессией с гарантированным закрытием:

from contextlib import asynccontextmanager
import aiohttp

class HTTPClient:
    @asynccontextmanager
    async def session(self):
        session = aiohttp.ClientSession()
        try:
            yield session
        finally:
            await session.close()

client = HTTPClient()

async def fetch_data(url: str):
    async with client.session() as session:
        async with session.get(url) as response:
            return await response.json()

Пример: лок для асинхронных операций

Использовал асинхронные контекстные менеджеры для управления распределенными локами:

from contextlib import asynccontextmanager
import asyncio

class AsyncLock:
    def __init__(self, key: str, timeout: int = 30):
        self.key = key
        self.timeout = timeout
        self.lock = asyncio.Lock()

    @asynccontextmanager
    async def __aenter__(self):
        async with self.lock:
            yield

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        pass

async def register_user(email: str):
    async with lock:
        await check_email_exists(email)
        await create_user(email)

Пример: измерение времени выполнения

Асинхронный контекстный менеджер для логирования и мониторинга:

from contextlib import asynccontextmanager
import time
import logging

logger = logging.getLogger(__name__)

@asynccontextmanager
async def measure_time(operation_name: str):
    start = time.time()
    try:
        yield
    finally:
        duration = time.time() - start
        logger.info(f"{operation_name} completed in {duration:.2f}s")

async def process_data():
    async with measure_time("data_processing"):
        await heavy_computation()

Ключевые моменты

• @asynccontextmanager — декоратор для создания асинхронного контекстного менеджера
• async with — синтаксис для использования
• Гарантированное очищение ресурсов — блок finally срабатывает даже при исключениях
• Использование в production — применял во множестве микросервисов и data pipeline'ов

Асинхронные контекстные менеджеры — стандартный инструмент в моем наборе для надежного управления ресурсами в асинхронном коде.