Асинхронность в Python: asyncio, aiohttp и best practices

Введение в асинхронность в Python

Асинхронная разработка программ — это подход к написанию кода, при котором выполнение операций не блокируется на ожидание внешних событий (например, сетевых запросов или ввода-вывода). В отличие от многопоточности, где создаётся несколько потоков для параллельного выполнения задач, асинхронность использует один поток и переключение между задачами по мере готовности ресурсов. Это позволяет достичь высокой производительности без избыточных накладных расходов.

В Python асинхронные возможности были представлены начиная с версии 3.4 через модуль asyncio, а в более поздних версиях (начиная с 3.5) — расширены поддержкой синтаксиса async/await. Эти инструменты позволяют эффективно обрабатывать I/O-ограниченные операции, такие как HTTP-запросы к API или работа с базами данных.

В данной статье рассматриваются основы асинхронности в Python: принципы работы event loop’а, использование библиотеки aiohttp для асинхронных HTTP-вызовов и лучшие практики применения асинхронного программирования. Особое внимание уделено вопросам конкурентности, избеганию блокирующих операций и эффективной интеграции с синхронными компонентами.

Основы asyncio

Модуль asyncio — центральный элемент Python для реализации асинхронной логики. Он предоставляет механизм event loop’а (цикла событий), который управляет выполнением корутин — специальных объектов, представляющих асинхронные функции.

Корутина — это функция, определённая с ключевым словом async, которая может быть приостановлена и возобновлена позже:

import asyncio

def simple_coroutine():
    print('Начало')
    await asyncio.sleep(1)
    print('Продолжение')

async def main():
    task = asyncio.create_task(simple_coroutine())
    # Можно выполнять другие задачи одновременно
    await task

if __name__ == '__main__':
    asyncio.run(main())

Здесь simple_coroutine — корута, а create_task позволяет запустить её асинхронно. Функция await asyncio.sleep(1) не блокирует поток полностью: event loop продолжает обрабатывать другие задачи.

Event Loop и планирование задач

Event loop’ы в Python могут быть реализованы разными способами (через asyncio.run, loop.run_until_complete или вручную), но в большинстве случаев используется стандартный цикл из модуля asyncio. Он отвечает за:

• Запуск корутин;

• Переключение между ними при ожидании I/O;

• Обработку исключений;

• Управление таймерами (через ensure_future, create_task).

Важно понимать, что асинхронность работает только на операциях ввода-вывода. CPU-bound задачи (например, тяжёлые вычисления) не ускоряются — они всё ещё выполняются в одном потоке и могут стать узким местом.

Пример сравнения синхронного и асинхронного подхода при выполнении 10 HTTP-запросов:

import time
import aiohttp
import requests
from concurrent.futures import ThreadPoolExecutor

# Синхронный способ (блокирующий)
def sync_requests():
    start = time.time()
    with ThreadPoolExecutor(5) as executor:
        futures = [executor.submit(requests.get, 'https://httpbin.org/delay/1') for _ in range(10)]
        results = [f.result() for f in futures]
    print(f'Синхронно: {time.time() - start:.2f} сек')

# Асинхронный способ (неблокирующий)
async def async_requests():
    start = time.time()
    async with aiohttp.ClientSession() as session:
        tasks = [session.get('https://httpbin.org/delay/1') for _ in range(10)]
        responses = await asyncio.gather(*tasks)
    print(f'Асинхронно: {time  .time() - start:.2f} сек')

if __name__ == '__main__':
    sync_requests()
    asyncio.run(async_requests())

Результат обычно показывает, что асинхронный подход работает быстрее при большом количестве I/O-операций — поскольку не создаёт много потоков и эффективно использует один.

aiohttp: асинхронные HTTP-клиенты

Библиотека aiohttp — стандартное решение для работы с HTTP в асинхронном режиме. Она предоставляет классы, аналогичные requests, но с поддержкой асинхронных вызовов (ClientSession, Request, Response).

Пример использования:

import asyncio
from aiohttp import ClientSession, ClientTimeout

async def fetch_data(url):
    async with ClientSession(timeout=ClientTimeout(total=5)) as session:
        try:
            async with session.get(url) as response:
                if response.status == 200:
                    data = await response.json()
                    return data
                else:
                    raise Exception(f'Ошибка: {response.status}')
        except asyncio.TimeoutError:
            print('Таймаут запроса')
            return None

async def main():
    tasks = [fetch_data('https://httpbin.org/delay/2') for _ in range(5)]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    successful = [r for r in results if not isinstance(r, Exception)]
    print(f'Успешно: {len(successful)}')

if __name__ == '__main__':
    asyncio.run(main())

Особенности работы с aiohttp

• ClientSession должен быть создан один раз и переиспользован для нескольких запросов — это повышает производительность.

• Необходимо обрабатывать исключения (TimeoutError, CancelledError) явно, так как асинхронные ошибки не всегда видны в синхронном коде.

• Для повторных попыток (retry) можно использовать сторонние библиотеки вроде tenacity:

from tenacity import retry, stop_after_attempt, wait_fixed

def fetch_with_retry(url):
    @retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
    async def inner():
        return await fetch_data(url)
    return asyncio.run(inner())

Работа с базами данных: асинхронные драйверы

Асинхронность особенно полезна при работе с БД, где запросы к серверу — I/O-ограниченные операции. Современные Python-драйверы для СУБД поддерживают асинхронный режим.

asyncpg (PostgreSQL)

Для PostgreSQL используется asyncpg:

import asyncio
import asyncpg

connection = None

def get_connection():
    global connection
    if not connection:
        loop = asyncio.get_event_loop()
        connection = await loop.run_until_complete(asyncpg.connect(
            user='user', password='pass', database='db', host='localhost'
        ))
    return connection

async def query_db():
    conn = await get_connection()
    result = await conn.fetch('SELECT * FROM users WHERE id = $1', 1)
    print(result)
    await conn.close()

if __name__ == '__main__':
    asyncio.run(query_db())

Примечание: asyncpg не поддерживает многопользовательский доступ к одной сессии — каждая задача должна создавать свою сессию.

aiomysql / asyncmy (MySQL)

Для MySQL можно использовать либо aiomysql, либо более современную обёртку asyncmy, которая совместима со стандартным API драйверов (pymysql), но работает асинхронно:

import asyncio
from asyncmy import create_pool, connect

current_pool = None

def get_db_connection():
    global current_pool
    if not current_pool:
        loop = asyncio.get_event_loop()
        pool = await loop.run_until_complete(
            create_pool(user='user', password='pass', db='db', host='localhost')
        )
        current_pool = pool
    return current_pool

async def read_users():
    conn = await get_db_connection()
    cursor = await conn.execute('SELECT * FROM users')
    result = await cursor.fetchall()
    print(result)
    if cursor:
        await cursor.close()

SQLAlchemy с async режимом (asyncio + asyncio-sqlite / asyncpg)

SQLAlchemy 2.0 поддерживает асинхронные движки через create_engine с параметром 'execution_options': {'isolation_level': None}, но чаще используется вместе с asyncmy:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
import asyncpg

global ENGINE
ENGINE = None

def get_engine():
    global ENGINE
    if not ENGINE:
        ENGINE = create_async_engine(
            'postgresql+asyncpg://user:pass@localhost/db', echo=False
        )
    return ENGINE

# Создание сессии для работы
async def do_query():
    async with AsyncSession(get_engine()) as session:
        result = await session.execute('SELECT name FROM users')
        print(result.scalar())

Лучшие практики асинхронности

Для надёжной и эффективной асинхронной разработки важно соблюдать ряд правил.

1. Избегайте блокирующих вызовов внутри корутин

Любые await на синхронные функции (например, requests, subprocess.Popen) приводят к потере преимуществ асинхронности:

# Плохо: смешивание синхроники и асинхроники в одной coroutine
async def bad_example():
    async with aiohttp.ClientSession() as session:
        response = await session.get('https://api.example.com')  # Ok
        data = requests.get(url='https://slow-site.com').json()  # Блокирует!

Вместо этого — либо полностью асинхронно, либо явно использовать пул потоков (ThreadPoolExecutor).

2. Используйте asyncio.gather или asyncio.wait для параллельного выполнения задач

Это позволяет запускать несколько запросов одновременно:

tasks = [fetch_url(url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)

3. Управляйте временем жизни сессий и соединений

• ClientSession в aiohttp, соединения к БД — не должны создаваться при каждом запросе.

• Используйте пул соединений (asyncpg, aiomysql, SQLAlchemy).

4. Обрабатывайте исключения асинхронно

Асинхронные ошибки могут быть скрытыми:

try:
    await task
except asyncio.CancelledError:
    print('Задача отменена')
except Exception as e:
    log_error(e)

5. Не создавайте слишком много задач за раз

Event loop может перегружаться большим количеством Task. Лучше группировать по логическим блокам.

Интеграция с синхронным кодом

Часто асинхронные сервисы должны взаимодействовать со синхронными компонентами (например, legacy-системы). Здесь важно не «синхронизировать» всё — а адаптироваться под нужды системы.

Подход: run_in_executor

Можно запускать асинхронные функции внутри синхронного контекста:

import asyncio
from concurrent.futures import ThreadPoolExecutor

def sync_call():
    loop = asyncio.new_event_loop()  # Не рекомендуется без необходимости
    return loop.run_until_complete(async_function())

Но правильнее — использовать ThreadPoolExecutor для запуска асинхронных функций из синхронного кода:

from concurrent.futures import ThreadPoolExecutor
import asyncio

def sync_wrapper():
    with ThreadPoolExecutor(max_workers=3) as executor:
        future = executor.submit(asyncio.run, async_function())
        return future.result()  # Блокирует до завершения

Альтернатива: run_sync с asyncio-to-threadpool (через сторонние инструменты)

Библиотека nest-asyncio позволяет запускать асинхронный loop внутри другого (например, Jupyter или GUI), но не рекомендуется для production-систем — только в тестах и прототипах.

Заключение: когда использовать асинхронность?

Асинхронность даёт выигрыш:

• При большом количестве I/O-запросов (HTTP, БД);

• В микросервисных архитектурах с высокой нагрузкой;

• Когда важно минимизировать потребление памяти и число потоков.

Однако она не подходит для:

• CPU-интенсивных вычислений — используйте multiprocessing или concurrent.futures.ProcessPoolExecutor;

• Сложной логики, где нужна синхронная последовательность (например, обработка событий в реальном времени без блокировок).

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

Дополнительные материалы и инструменты

• Официальная документация asyncio

• aiohttp docs — https://docs.aiohttp.org/en/stable/

• SQLAlchemy 2.0 Async — https://www.sqlalchemy.org/tutorials/modern/async-sqlengine

• Библиотеки для retry: tenacity, [aioresponses] (для моков)