Async и конкурентность

← назад к разделу

FastAPI строится на асинхронности: один процесс держит тысячи одновременных запросов, пока они ждут базу или внешний сервис. Но та же асинхронность — источник незаметных проблем: один блокирующий вызов в неправильном месте останавливает весь сервис, и в логах это никак не отразится.

Разберём с нуля: что такое event loop, чем async def отличается от def, и где что использовать.

Как Python выполняет много задач в одном потоке

Обычная программа делает одно действие за другим. Если функция делает запрос к базе, она просто ждёт — никакой другой работы в это время не происходит.

Асинхронный подход устроен иначе. Есть event loop (цикл событий) — один поток, который ведёт множество задач одновременно. Когда задача начинает ждать (запрос к базе, HTTP-вызов, чтение файла), она говорит: «я жду, займись другими». Loop переключается на следующую задачу, а вернётся к первой, когда та дождётся ответа.

Ключевое слово await — это именно такая пометка: «здесь я жду, loop может переключиться».

async def get_user(user_id: int):
    user = await db.fetch_user(user_id)   # ждём БД → loop свободен для других запросов
    return user

Пока db.fetch_user выполняется на стороне базы данных, loop обрабатывает другие входящие запросы. Вернётся к этой функции, когда ответ придёт.

Две дороги: async def и def

FastAPI выполняет обработчики по-разному в зависимости от того, как они объявлены.

async def — выполняется прямо в event loop. Пока функция ждёт через await, loop свободен. Это главный путь в FastAPI.

def (обычная функция) — FastAPI сам уводит её в отдельный поток из пула, чтобы синхронный код не заблокировал loop. Это полезно, когда синхронного кода не избежать.

@router.get("/users/{user_id}")
async def get_user(user_id: int):
    # правильно: async-драйвер базы, await, loop свободен
    user = await db.execute(select(User).where(User.id == user_id))
    return user


@router.get("/report")
def generate_report():
    # правильно: синхронный тяжёлый расчёт → FastAPI уводит в threadpool
    data = heavy_sync_calculation()
    return data

Главная ловушка: блокировка event loop

Самая частая ошибка — написать async def, но внутри вызвать что-то блокирующее без await:

import time


@router.get("/bad")
async def bad():
    time.sleep(3)         # блокирует ВЕСЬ event loop на 3 секунды


@router.get("/ok")
async def ok():
    await asyncio.sleep(3)    # отдаёт управление loop — другие запросы идут

time.sleep не знает про event loop. Он останавливает поток целиком — то есть весь loop, и все параллельные запросы встают вместе с ним. Никакой ошибки в логах не будет, сервис просто начнёт медленно отвечать под нагрузкой.

Правило: в async def нельзя вызывать ничего блокирующего без await. Вы пообещали фреймворку неблокирующую функцию — он рассчитывает на это.

Async-клиенты для внешних вызовов

Если обработчик async def, все исходящие запросы тоже должны быть асинхронными. Синхронный HTTP-клиент внутри async def — та самая ловушка.

import httpx


# неправильно: requests — синхронная библиотека, блокирует loop
@router.get("/rate/bad")
async def rate_bad(currency: str):
    import requests
    response = requests.get(f"https://rates.example/{currency}")  # блокирует loop
    return response.json()


# правильно: httpx.AsyncClient с await
@router.get("/rate")
async def rate(currency: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://rates.example/{currency}")
    return response.json()

То же правило для базы данных: нужен async-драйвер и async-сессия SQLAlchemy (await session.execute(...)), а не синхронный вызов.

Параллельные запросы через asyncio.gather

Иногда нужно сделать несколько независимых async-вызовов и дождаться всех. Если делать их по очереди через await, они выполнятся последовательно. asyncio.gather запускает их одновременно:

import asyncio
import httpx


@router.get("/dashboard")
async def dashboard():
    async with httpx.AsyncClient() as client:
        user_task = client.get("https://api.example/user")
        stats_task = client.get("https://api.example/stats")

        user_resp, stats_resp = await asyncio.gather(user_task, stats_task)

    return {
        "user": user_resp.json(),
        "stats": stats_resp.json(),
    }

Оба запроса уходят почти одновременно. Если каждый занимает 200 мс, суммарное время будет около 200 мс, а не 400.

Когда блокирующего не избежать

Иногда синхронного кода не обойти: библиотека без async-версии, CPU-тяжёлый шаг, обращение к файловой системе. Если остальной обработчик async, такой вызов уводят в поток явно через asyncio.to_thread:

import asyncio


def render_pdf(body):
    # тяжёлая синхронная операция
    ...


@router.post("/render")
async def render(body: RenderRequest):
    result = await asyncio.to_thread(render_pdf, body)   # в отдельный поток, loop свободен
    return result

asyncio.to_thread выполняет синхронную функцию в пуле потоков и возвращает результат через await. Loop при этом не блокируется.

Это то же самое, что происходит автоматически при def-обработчике, — только явное, когда вы хотите оставаться в async def.

Когда что выбирать

Худший вариант — async def с синхронным блокирующим вызовом внутри: ни корректности, ни производительности.

Коротко

• Event loop — один поток, который ведёт много задач, переключаясь между ними в точках await.
• async def выполняется в event loop. В нём нельзя вызывать ничего блокирующего без await.
• def FastAPI автоматически выполняет в threadpool — синхронный код не блокирует loop.
• Блокирующий вызов в async def останавливает весь event loop и все запросы сразу.
• Для HTTP-запросов наружу используют httpx.AsyncClient, а не синхронный requests.
• Для принудительного выноса синхронного кода в поток — asyncio.to_thread.
• Для параллельного выполнения нескольких async-задач — asyncio.gather.

Что почитать дальше

• Персистентность и SQLAlchemy — async-сессии и ORM в FastAPI.
• Фоновые задачи — что делать с долгими операциями вне обработчика.
• Dependency Injection — как FastAPI управляет зависимостями.