asyncio и асинхронное программирование в Python: паттерны, футуры, примеры, работа с БД

import asyncio
from asyncio import Queue
 
async def producer(queue: Queue, producer_id: int):
    """Генерирует задачи"""
    for i in range(5):
        item = f"задача_{producer_id}_{i}"
        await queue.put(item)
        print(f"Производитель {producer_id} добавил {item}")
        await asyncio.sleep(0.5)
    
async def consumer(queue: Queue, consumer_id: int):
    """Обрабатывает задачи"""
    while True:
        item = await queue.get()
        if item is None:  # Сигнал завершения
            queue.task_done()
            break
        print(f"Потребитель {consumer_id} обрабатывает {item}")
        await asyncio.sleep(1)  # Имитация работы
        queue.task_done()
 
async def main():
    queue = Queue(maxsize=10)
    
    # Запускаем производителей и потребителей
    producers = [asyncio.create_task(producer(queue, i)) for i in range(2)]
    consumers = [asyncio.create_task(consumer(queue, i)) for i in range(3)]
    
    # Ждем завершения производителей
    await asyncio.gather(*producers)
    
    # Отправляем сигналы завершения потребителям
    for _ in consumers:
        await queue.put(None)
    
    # Ждем завершения потребителей
    await asyncio.gather(*consumers)
 
asyncio.run(main())

async def rate_limited_request(url: str, semaphore: asyncio.Semaphore):
    """HTTP запрос с ограничением конкурентности"""
    async with semaphore:  # Захватываем разрешение
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                return await response.text()
 
async def fetch_many_urls(urls: list[str]):
    semaphore = asyncio.Semaphore(10)  # Максимум 10 одновременно
    tasks = [rate_limited_request(url, semaphore) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

async def fetch_with_retry(url: str, max_retries: int = 3):
    """Запрос с повторными попытками"""
    for attempt in range(max_retries):
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(url, timeout=10) as response:
                    response.raise_for_status()
                    return await response.json()
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt == max_retries - 1:
                raise  # Последняя попытка - пробрасываем ошибку
            
            delay = 2 ** attempt  # Экспоненциальная задержка
            print(f"Попытка {attempt + 1} не удалась, жду {delay}с")
            await asyncio.sleep(delay)

class DatabaseConnection:
    """Асинхронный контекстный менеджер для соединения с БД"""
    
    async def __aenter__(self):
        self.conn = await asyncpg.connect("postgresql://localhost/db")
        return self.conn
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.conn.close()
        # Возвращаем False, чтобы исключение пробрасывалось дальше
        return False
 
async def use_database():
    async with DatabaseConnection() as conn:
        result = await conn.fetch("SELECT * FROM users")
        return result
    # Соединение автоматически закрыто

class BatchedWriter:
    def __init__(self, batch_size: int = 100):
        self.batch_size = batch_size
        self.buffer = []
        self.conn = None
    
    async def write(self, item):
        self.buffer.append(item)
        if len(self.buffer) >= self.batch_size:
            await self.flush()
    
    async def flush(self):
        if not self.buffer:
            return
        
        await self.conn.executemany(
            "INSERT INTO data VALUES ($1, $2)",
            self.buffer
        )
        self.buffer.clear()

import aiohttp
import asyncio
 
async def fetch_url(session, url):
    """Загружает содержимое одного URL"""
    async with session.get(url) as response:
        return await response.text()
 
async def fetch_all(urls):
    """Параллельная загрузка множества URL"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    return results
 
# Использование
urls = [
    "https://api.example.com/data1",
    "https://api.example.com/data2",
    "https://api.example.com/data3",
]
results = asyncio.run(fetch_all(urls))

async def fetch_with_limit(session, url, semaphore):
    """Запрос с контролем конкурентности"""
    async with semaphore:
        async with session.get(url) as response:
            return await response.json()
 
async def fetch_many_controlled(urls, limit=10):
    """Загрузка с ограничением одновременных запросов"""
    semaphore = asyncio.Semaphore(limit)
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_with_limit(session, url, semaphore) 
            for url in urls
        ]
        return await asyncio.gather(*tasks, return_exceptions=True)

async def robust_fetch(urls):
    """Надежная загрузка с обработкой ошибок"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # Разделяем успешные и неудачные
    successful = []
    failed = []
    
    for url, result in zip(urls, results):
        if isinstance(result, Exception):
            failed.append((url, str(result)))
        else:
            successful.append((url, result))
    
    return successful, failed

async def fetch_with_timeout(url, timeout_seconds=5):
    """Запрос с общим таймаутом"""
    timeout = aiohttp.ClientTimeout(total=timeout_seconds)
    try:
        async with aiohttp.ClientSession(timeout=timeout) as session:
            async with session.get(url) as response:
                return await response.text()
    except asyncio.TimeoutError:
        return None  # Или пробросить исключение

async def fetch_paginated_api(base_url):
    """Загрузка API с пагинацией"""
    all_data = []
    page = 1
    
    async with aiohttp.ClientSession() as session:
        while True:
            url = f"{base_url}?page={page}"
            async with session.get(url) as response:
                data = await response.json()
                
                if not data.get("items"):
                    break  # Достигли конца
                
                all_data.extend(data["items"])
                page += 1
                
                if page > data.get("total_pages", page):
                    break
    
    return all_data

async def process_as_ready(urls):
    """Обработка результатов по мере поступления"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        
        for coro in asyncio.as_completed(tasks):
            try:
                result = await coro
                # Обрабатываем результат немедленно
                process_data(result)
            except Exception as e:
                handle_error(e)

async def optimized_fetcher(urls):
    """Оптимизированная загрузка с настройкой пула"""
    connector = aiohttp.TCPConnector(
        limit=100,           # Максимум 100 соединений всего
        limit_per_host=20,   # Максимум 20 на один хост
        ttl_dns_cache=300    # Кеш DNS на 5 минут
    )
    
    async with aiohttp.ClientSession(connector=connector) as session:
        semaphore = asyncio.Semaphore(50)
        tasks = [
            fetch_with_limit(session, url, semaphore) 
            for url in urls
        ]
        return await asyncio.gather(*tasks, return_exceptions=True)

async def fetch_with_retry(session, url, max_attempts=3):
    """Запрос с повторными попытками"""
    for attempt in range(max_attempts):
        try:
            async with session.get(url) as response:
                response.raise_for_status()
                return await response.json()
        except aiohttp.ClientError as e:
            if attempt == max_attempts - 1:
                raise  # Последняя попытка - сдаемся
            
            # Экспоненциальная задержка: 1s, 2s, 4s
            delay = 2 ** attempt
            await asyncio.sleep(delay)

async def safe_fetch(url):
    """Запрос с базовой обработкой ошибок"""
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url, timeout=5) as response:
                return await response.json()
    except aiohttp.ClientError as e:
        print(f"Ошибка сети: {e}")
        return None
    except asyncio.TimeoutError:
        print(f"Таймаут для {url}")
        return None
    except Exception as e:
        print(f"Неожиданная ошибка: {e}")
        raise

async def fetch_multiple_fail_fast(urls):
    """При ошибке прерываем все"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        try:
            results = await asyncio.gather(*tasks)
            return results
        except aiohttp.ClientError as e:
            print(f"Одна из задач упала: {e}")
            return []

async def fetch_multiple_tolerant(urls):
    """Собираем все результаты, включая ошибки"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # Разделяем успешные и неудачные
    successful = []
    failed = []
    
    for url, result in zip(urls, results):
        if isinstance(result, Exception):
            failed.append((url, result))
        else:
            successful.append((url, result))
    
    return successful, failed

async def robust_operation(url):
    """Каждая операция с собственной обработкой ошибок"""
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                data = await response.json()
                return {"status": "success", "data": data}
    except aiohttp.ClientError as e:
        return {"status": "network_error", "error": str(e)}
    except asyncio.TimeoutError:
        return {"status": "timeout", "error": "Request timed out"}
    except json.JSONDecodeError as e:
        return {"status": "parse_error", "error": str(e)}
 
async def process_many(urls):
    """Обработка с детальным статусом каждой операции"""
    tasks = [robust_operation(url) for url in urls]
    results = await asyncio.gather(*tasks)
    
    # Теперь можем анализировать статусы
    success_count = sum(1 for r in results if r["status"] == "success")
    print(f"Успешно: {success_count}/{len(results)}")
    
    return results

async def background_worker():
    """Фоновая задача с обработкой ошибок"""
    try:
        while True:
            await process_queue()
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("Задача отменена, очищаем ресурсы")
        await cleanup()
        raise  # Важно пробросить CancelledError
    except Exception as e:
        print(f"Фоновая задача упала: {e}")
        # Логируем, уведомляем, но не роняем приложение
        raise
 
async def main():
    task = asyncio.create_task(background_worker())
    
    try:
        await asyncio.sleep(10)
    finally:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass  # Ожидаемая отмена

async def failing_tasks():
    """Несколько задач с разными ошибками"""
    async def task_a():
        await asyncio.sleep(0.1)
        raise ValueError("Ошибка в A")
    
    async def task_b():
        await asyncio.sleep(0.2)
        raise TypeError("Ошибка в B")
    
    async def task_c():
        await asyncio.sleep(0.15)
        raise KeyError("Ошибка в C")
    
    results = await asyncio.gather(
        task_a(), task_b(), task_c(),
        return_exceptions=True
    )
    
    # Собираем исключения в группу
    exceptions = [r for r in results if isinstance(r, Exception)]
    if exceptions:
        raise ExceptionGroup("Множественные ошибки", exceptions)
 
async def handle_grouped_errors():
    """Обработка группы исключений"""
    try:
        await failing_tasks()
    except* ValueError as eg:
        print(f"Поймали ValueError: {eg.exceptions}")
    except* TypeError as eg:
        print(f"Поймали TypeError: {eg.exceptions}")
    except* KeyError as eg:
        print(f"Поймали KeyError: {eg.exceptions}")

async def monitored_background_task(task_name: str):
    """Фоновая задача с мониторингом"""
    last_heartbeat = time.time()
    
    try:
        while True:
            await do_work()
            last_heartbeat = time.time()
            await report_heartbeat(task_name, last_heartbeat)
            await asyncio.sleep(1)
    except Exception as e:
        await send_alert(f"Task {task_name} failed: {e}")
        await log_exception(task_name, e)
        raise

import asyncpg
import asyncio
 
async def fetch_users():
    """Асинхронное подключение и запрос"""
    conn = await asyncpg.connect(
        host='localhost',
        database='mydb',
        user='postgres',
        password='secret'
    )
    
    try:
        rows = await conn.fetch('SELECT id, name FROM users WHERE active = $1', True)
        return [dict(row) for row in rows]
    finally:
        await conn.close()

async def setup_database():
    """Создание пула соединений"""
    pool = await asyncpg.create_pool(
        host='localhost',
        database='mydb',
        user='postgres',
        password='secret',
        min_size=5,    # Минимум соединений
        max_size=20    # Максимум соединений
    )
    return pool
 
async def get_user_by_id(pool, user_id: int):
    """Запрос через пул"""
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            'SELECT * FROM users WHERE id = $1',
            user_id
        )
        return dict(row) if row else None
 
async def main():
    pool = await setup_database()
    
    try:
        # Параллельные запросы к базе
        tasks = [get_user_by_id(pool, i) for i in range(1, 11)]
        users = await asyncio.gather(*tasks)
        print(f"Загружено пользователей: {len([u for u in users if u])}")
    finally:
        await pool.close()
 
asyncio.run(main())

async def transfer_money(pool, from_id: int, to_id: int, amount: float):
    """Перевод денег с обработкой транзакции"""
    async with pool.acquire() as conn:
        async with conn.transaction():
            # Проверяем баланс
            balance = await conn.fetchval(
                'SELECT balance FROM accounts WHERE id = $1',
                from_id
            )
            
            if balance < amount:
                raise ValueError("Недостаточно средств")
            
            # Списываем у отправителя
            await conn.execute(
                'UPDATE accounts SET balance = balance - $1 WHERE id = $2',
                amount, from_id
            )
            
            # Начисляем получателю
            await conn.execute(
                'UPDATE accounts SET balance = balance + $1 WHERE id = $2',
                amount, to_id
            )
            
            # При выходе из блока транзакция автоматически commit'ится
            # При исключении - rollback

async def bulk_insert_users(pool, users: list):
    """Массовая вставка пользователей"""
    async with pool.acquire() as conn:
        await conn.executemany(
            'INSERT INTO users (name, email, created_at) VALUES ($1, $2, $3)',
            [(u['name'], u['email'], datetime.now()) for u in users]
        )
 
# Вместо цикла с N запросами:
# for user in users:
#     await conn.execute('INSERT INTO users ...')

async def query_with_prepared_statement(pool):
    """Использование prepared statement"""
    async with pool.acquire() as conn:
        # Готовим statement
        stmt = await conn.prepare('SELECT * FROM users WHERE age > $1 AND country = $2')
        
        # Выполняем с разными параметрами
        young_users = await stmt.fetch(18, 'RU')
        middle_aged = await stmt.fetch(35, 'RU')
        
        return young_users, middle_aged

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
 
engine = create_async_engine(
    'postgresql+asyncpg://user:pass@localhost/db',
    echo=True
)
 
async_session = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)
 
async def get_users_orm():
    """Запрос через SQLAlchemy ORM"""
    async with async_session() as session:
        result = await session.execute(
            select(User).where(User.active == True)
        )
        users = result.scalars().all()
        return users

async def monitor_pool(pool):
    """Периодический мониторинг состояния пула"""
    while True:
        print(f"Размер пула: {pool.get_size()}")
        print(f"Свободных: {pool.get_idle_size()}")
        print(f"Используется: {pool.get_size() - pool.get_idle_size()}")
        await asyncio.sleep(10)

import asyncio
 
async def process_line(line: str) -> dict:
    """Обработка одной строки данных"""
    # Имитация тяжелой обработки
    await asyncio.sleep(0.1)
    return {"data": line.strip(), "length": len(line)}
 
async def stream_file_lines(filepath: str):
    """Генератор для потокового чтения файла"""
    async with aiofiles.open(filepath, mode='r') as file:
        async for line in file:
            yield line
 
async def process_file_stream(filepath: str, batch_size: int = 10):
    """Обработка файла порциями"""
    batch = []
    
    async for line in stream_file_lines(filepath):
        batch.append(line)
        
        if len(batch) >= batch_size:
            # Обрабатываем порцию параллельно
            results = await asyncio.gather(
                *[process_line(line) for line in batch]
            )
            
            # Сохраняем результаты или отправляем дальше
            for result in results:
                await save_result(result)
            
            batch.clear()
    
    # Обрабатываем остаток
    if batch:
        results = await asyncio.gather(
            *[process_line(line) for line in batch]
        )

async def download_and_process_stream(url: str, chunk_size: int = 8192):
    """Потоковая загрузка и обработка"""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            total_size = 0
            
            async for chunk in response.content.iter_chunked(chunk_size):
                # Обрабатываем chunk немедленно
                await process_chunk(chunk)
                total_size += len(chunk)
                
                # Показываем прогресс
                print(f"Обработано: {total_size / 1024:.2f} КБ")
    
    return total_size

async def stream_producer(queue: asyncio.Queue, source):
    """Читает данные и помещает в очередь"""
    async for item in source:
        await queue.put(item)
    
    # Сигнал завершения
    await queue.put(None)
 
async def stream_consumer(queue: asyncio.Queue, consumer_id: int):
    """Обрабатывает элементы из очереди"""
    while True:
        item = await queue.get()
        
        if item is None:
            queue.task_done()
            break
        
        try:
            result = await process_item(item)
            await send_to_output(result)
        except Exception as e:
            print(f"Consumer {consumer_id} error: {e}")
        finally:
            queue.task_done()
 
async def stream_pipeline(source, num_consumers: int = 3):
    """Конвейер обработки потока"""
    queue = asyncio.Queue(maxsize=50)  # Буфер на 50 элементов
    
    # Запускаем producer и consumers
    producer = asyncio.create_task(stream_producer(queue, source))
    consumers = [
        asyncio.create_task(stream_consumer(queue, i))
        for i in range(num_consumers)
    ]
    
    # Ждем завершения producer
    await producer
    
    # Отправляем сигналы завершения всем consumers
    for _ in consumers:
        await queue.put(None)
    
    # Ждем завершения всех consumers
    await asyncio.gather(*consumers)

async def fetch_paginated_stream(api_url: str):
    """Потоковая загрузка данных с пагинацией"""
    page = 1
    
    async with aiohttp.ClientSession() as session:
        while True:
            url = f"{api_url}?page={page}"
            async with session.get(url) as response:
                data = await response.json()
                
                if not data['items']:
                    break
                
                # Отдаем элементы по одному
                for item in data['items']:
                    yield item
                
                page += 1
 
async def process_api_stream():
    """Обработка потока из API"""
    async for item in fetch_paginated_stream("https://api.example.com/data"):
        await process_item(item)

async def controlled_stream_producer(queue: asyncio.Queue):
    """Producer с учетом backpressure"""
    for i in range(1000):
        item = await fetch_data(i)
        
        # Если очередь полна, put() будет ждать
        await queue.put(item)
        
        # Producer автоматически замедляется

import asyncio
 
# Создание Future вручную
loop = asyncio.get_event_loop()
future = loop.create_future()
 
print(f"Состояние: done={future.done()}, cancelled={future.cancelled()}")
 
# Установка результата
future.set_result(42)
print(f"После set_result: done={future.done()}")
print(f"Результат: {future.result()}")

def legacy_async_operation(callback):
    """Старая функция с callback"""
    def delayed_work():
        import time
        time.sleep(1)
        callback("результат работы")
    
    import threading
    threading.Thread(target=delayed_work).start()
 
async def wrap_legacy_operation():
    """Обертка для использования с await"""
    loop = asyncio.get_running_loop()
    future = loop.create_future()
    
    def callback(result):
        # Устанавливаем результат в Future
        loop.call_soon_threadsafe(future.set_result, result)
    
    legacy_async_operation(callback)
    return await future
 
# Теперь можно использовать с await
result = await wrap_legacy_operation()

def sync_callback():
    print("Callback выполнен")
 
loop = asyncio.get_running_loop()
 
# Выполнится на следующей итерации loop
loop.call_soon(sync_callback)
 
# Выполнится через 2 секунды
loop.call_later(2.0, sync_callback)

import concurrent.futures
 
def blocking_operation(n):
    """CPU-intensive или блокирующая операция"""
    result = sum(i**2 for i in range(n))
    return result
 
async def async_wrapper():
    loop = asyncio.get_running_loop()
    
    # Выполняем в thread pool
    with concurrent.futures.ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(
            pool, 
            blocking_operation, 
            1_000_000
        )
    
    return result

async def some_coroutine():
    await asyncio.sleep(1)
    return "готово"
 
# Эти два вызова эквивалентны
task1 = asyncio.create_task(some_coroutine())
task2 = asyncio.ensure_future(some_coroutine())

import socket
 
async def raw_socket_client():
    loop = asyncio.get_running_loop()
    
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setblocking(False)
    
    await loop.sock_connect(sock, ('example.com', 80))
    
    request = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n"
    await loop.sock_sendall(sock, request)
    
    response = await loop.sock_recv(sock, 4096)
    
    sock.close()
    return response.decode()

import asyncio
 
async def async_range(start, stop):
    """Асинхронный аналог range()"""
    current = start
    while current < stop:
        yield current
        current += 1
        await asyncio.sleep(0.1)  # Имитация асинхронной работы
 
async def consume_async_generator():
    """Потребление значений из асинхронного генератора"""
    async for value in async_range(0, 5):
        print(f"Получено значение: {value}")
 
asyncio.run(consume_async_generator())

async def stream_database_records(pool, batch_size=1000):
    """Потоковое чтение из базы"""
    offset = 0
    
    while True:
        async with pool.acquire() as conn:
            rows = await conn.fetch(
                'SELECT * FROM large_table LIMIT $1 OFFSET $2',
                batch_size, offset
            )
            
            if not rows:
                break
            
            for row in rows:
                yield dict(row)
            
            offset += batch_size
            await asyncio.sleep(0)  # Отдаем управление
 
async def process_large_dataset(pool):
    """Обработка большого датасета без загрузки в память"""
    count = 0
    
    async for record in stream_database_records(pool):
        await process_record(record)
        count += 1
        
        if count % 10000 == 0:
            print(f"Обработано записей: {count}")

async def accumulator_generator():
    """Генератор-аккумулятор"""
    total = 0
    
    while True:
        value = yield total
        if value is None:
            break
        total += value
 
async def use_accumulator():
    """Использование генератора с отправкой значений"""
    gen = accumulator_generator()
    
    # Инициализация генератора
    await gen.asend(None)
    
    # Отправляем значения
    result = await gen.asend(10)  # total = 10
    print(f"Текущая сумма: {result}")
    
    result = await gen.asend(20)  # total = 30
    print(f"Текущая сумма: {result}")
    
    await gen.aclose()  # Закрываем генератор

class AsyncRangeIterator:
    """Кастомный асинхронный итератор"""
    
    def __init__(self, start, stop, delay=0.1):
        self.current = start
        self.stop = stop
        self.delay = delay
    
    def __aiter__(self):
        return self
    
    async def __anext__(self):
        if self.current >= self.stop:
            raise StopAsyncIteration
        
        await asyncio.sleep(self.delay)
        value = self.current
        self.current += 1
        return value
 
async def use_custom_iterator():
    """Использование кастомного итератора"""
    async for value in AsyncRangeIterator(0, 5):
        print(f"Значение: {value}")

async def async_square_numbers():
    """Возведение в квадрат асинхронно"""
    async def async_range(n):
        for i in range(n):
            await asyncio.sleep(0.1)
            yield i
    
    # Асинхронный list comprehension
    squares = [x**2 async for x in async_range(10)]
    return squares
 
async def filter_async_data():
    """Фильтрация с условием"""
    async def fetch_numbers():
        for i in range(20):
            await asyncio.sleep(0.05)
            yield i
    
    # Comprehension с условием
    evens = [x async for x in fetch_numbers() if x % 2 == 0]
    return evens

# Плохо - псевдоасинхронный генератор
async def fake_async_gen():
    for i in range(100):
        yield i  # Нет await - блокирует loop
 
# Хорошо - настоящий асинхронный генератор
async def real_async_gen():
    for i in range(100):
        yield i
        await asyncio.sleep(0)  # Отдаем управление

async def safe_generator_usage():
    """Безопасное использование генератора"""
    gen = stream_database_records(pool)
    
    try:
        async for record in gen:
            if should_stop(record):
                break
            await process_record(record)
    finally:
        await gen.aclose()  # Гарантированное закрытие

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor
 
def blocking_operation(n):
    """Синхронная блокирующая функция"""
    time.sleep(2)  # Блокирует поток
    return sum(i**2 for i in range(n))
 
async def async_wrapper(n):
    """Обертка для вызова из асинхронного кода"""
    loop = asyncio.get_running_loop()
    
    # Выполняем в thread pool
    result = await loop.run_in_executor(None, blocking_operation, n)
    return result
 
async def main():
    # Запускаем несколько операций параллельно
    tasks = [async_wrapper(1_000_000) for _ in range(3)]
    results = await asyncio.gather(*tasks)
    print(f"Результаты: {results}")
 
asyncio.run(main())

from concurrent.futures import ProcessPoolExecutor
 
async def cpu_intensive_task(data):
    """CPU-intensive вычисления"""
    loop = asyncio.get_running_loop()
    
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(
            pool,
            heavy_computation,
            data
        )
    
    return result

# Плохо - объект сессии не сериализуется
async def broken_approach(session):
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(
            pool,
            process_with_session,
            session  # Ошибка при сериализации
        )
 
# Хорошо - передаем параметры для создания сессии
async def correct_approach(db_url):
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(
            pool,
            process_with_url,
            db_url  # Строка сериализуется без проблем
        )

async def wrap_callback_api(param):
    """Обертка для callback-based библиотеки"""
    loop = asyncio.get_running_loop()
    future = loop.create_future()
    
    def callback(result, error=None):
        if error:
            loop.call_soon_threadsafe(future.set_exception, error)
        else:
            loop.call_soon_threadsafe(future.set_result, result)
    
    # Запуск синхронной операции с колбеком
    legacy_library.async_operation(param, callback)
    
    return await future

class AsyncService:
    """Сервис с переиспользуемым executor'ом"""
    
    def __init__(self):
        self.executor = ThreadPoolExecutor(max_workers=10)
    
    async def blocking_call(self, data):
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(
            self.executor,
            sync_operation,
            data
        )
    
    async def cleanup(self):
        """Обязательная очистка ресурсов"""
        self.executor.shutdown(wait=True)

# Плохо - блокирует event loop
async def bad_approach():
    data = requests.get("https://api.example.com")  # Блокирует
    return data.json()
 
# Хорошо - через executor
async def good_approach():
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(
        None,
        lambda: requests.get("https://api.example.com").json()
    )
 
# Еще лучше - асинхронная библиотека
async def best_approach():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com") as response:
            return await response.json()