В FastAPI можно объявить обработчик как async def, а можно как обычный def. Многие ставят async везде «потому что так быстрее» и спокойно вызывают внутри requests.get() или синхронный драйвер БД. Сервис при этом работает, тесты проходят, а на первой же реальной нагрузке всё встаёт колом, причём вместе с health‑check'ом.

В статье разберу, что на самом деле происходит с каждым вариантом, покажу замеры и дам короткую шпаргалку, какой вариант выбирать.

Как FastAPI выполняет обработчики

Правило одно, но из него следует всё остальное.

• async def выполняется прямо в event loop. Пока корутина не дошла до await, никто другой в этом процессе не работает: ни другие запросы, ни /health.

• def FastAPI сам отправляет в пул потоков через anyio.to_thread.run_sync. Event loop остаётся свободным, но размер пула ограничен: по умолчанию в anyio это 40 потоков на процесс.

То же самое касается зависимостей (Depends): синхронная зависимость уходит в threadpool, асинхронная выполняется в event loop.

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

Стенд

Чтобы не спорить на словах, я собрал небольшой стенд:

• «внешний API» — отдельный FastAPI‑сервис, который отвечает через 200 мс (await asyncio.sleep(0.2));

• тестируемый сервис с несколькими вариантами одной и той же ручки;

• нагрузочный скрипт на httpx.AsyncClient, который шлёт N одновременных запросов и считает общее время и перцентили.

Окружение: 1 vCPU, Python 3.12, FastAPI 0.141.1, Starlette 1.6.0, uvicorn 0.53.0 (один воркер), httpx 0.28.1, requests 2.33.1. Абсолютные цифры на вашей машине будут другими, но соотношения сохранятся. Каждый замер я повторял несколько раз, разброс был в пределах 10%.

Варианты ручки:

import httpx
import requests
from fastapi import FastAPI
from starlette.concurrency import run_in_threadpool

UPSTREAM = "http://127.0.0.1:9000/slow"

# 1. Ошибка: синхронный клиент внутри async def
@app.get("/async-requests")
async def async_requests():
    return requests.get(UPSTREAM).json()

# 2. Синхронная функция — FastAPI уведёт её в threadpool
@app.get("/sync-requests")
def sync_requests():
    return requests.get(UPSTREAM).json()

# 3. Синхронный вызов, явно вынесенный в поток
@app.get("/async-to-thread")
async def async_to_thread():
    r = await run_in_threadpool(requests.get, UPSTREAM)
    return r.json()

# 4. Правильно: асинхронный клиент в async def
@app.get("/async-httpx")
async def async_httpx():
    r = await app.state.client.get(UPSTREAM)  # httpx.AsyncClient из lifespan
    return r.json()

Полный код стенда — в репозитории (ссылка в конце).

Замер 1: 100 одновременных запросов

Первая строка — это не «немного медленнее». 100 запросов по 200 мс ровно за 20 секунд означают, что сервис обрабатывал их строго по одному. Каждый requests.get() держал event loop, пока ждал ответа, и остальные 99 запросов стояли в очереди. Разница с правильным вариантом — примерно в 40 раз по RPS.

Самое неприятное, что ручка с одиночным запросом отвечает за нормальные 200 мс. На локальной машине и в юнит‑тестах проблема не видна.

Остальные три варианта ведут себя нормально. Асинхронный клиент быстрее всех, потому что не тратит ресурсы на потоки, но и def, и явный run_in_threadpool дают разумный результат.

Замер 2: что происходит с health‑check

Теперь запустим 50 медленных запросов и, пока они выполняются, дёрнем самую простую ручку:

@app.get("/health")
async def health():
    return {"status": "ok"}

Десять секунд на ответ {"status": "ok"}. В Kubernetes liveness‑проба с таймаутом в пару секунд такое не переживёт: под перезапустят, нагрузка перейдёт на соседей, и те упадут по той же причине. Проблема в одной ручке превращается в падение всего сервиса.

Замер 3: CPU‑нагрузка в async def

Блокирует не только сетевой ввод‑вывод. Любые тяжёлые вычисления внутри async def делают то же самое:

def cpu_work():
    s = 0
    for i in range(3_000_000):
        s += i * i
    return s

@app.get("/async-cpu")
async def async_cpu():
    return {"s": cpu_work()}

Один такой запрос занимает около 120 мс. Пока выполняются пять таких запросов, /health отвечает за 571 мс: он ждёт, пока все пять досчитаются.

В реальных проектах такой код встречается чаще, чем кажется: хеширование паролей через bcrypt или argon2, генерация PDF и Excel, обработка изображений, pandas, сериализация огромных JSON.

Для CPU‑задач перенос в поток помогает только частично: из‑за GIL потоки не считают параллельно, хотя event loop хотя бы получает шанс переключиться. Надёжные варианты — ProcessPoolExecutor или вынос задачи в фоновый воркер (Celery, arq, Taskiq).

Замер 4: лимит threadpool

Раз def работает хорошо, можно сделать все ручки синхронными? Не совсем: у пула потоков есть предел.

Для чистоты эксперимента я взял ручку с time.sleep(0.5), которая не нагружает процессор, и отправил на неё 200 одновременных запросов. Параллельно я дёргал две быстрые ручки: синхронную /sync-health и асинхронную /health.

С 40 потоками 200 запросов по 0.5 с проходят пятью «волнами», отсюда примерно 2.5 секунды. Главное здесь — быструю синхронную ручку тоже задело: она ждала свободный поток больше двух секунд, хотя сама выполняется мгновенно. Все def‑обработчики и синхронные зависимости делят один общий пул. Асинхронный /health при этом почти не пострадал.

Лимит можно поднять в lifespan:

from contextlib import asynccontextmanager
import anyio.to_thread

@asynccontextmanager
async def lifespan(app: FastAPI):
    limiter = anyio.to_thread.current_default_thread_limiter()
    limiter.total_tokens = 100
    yield

Прежде чем крутить эту настройку, учтите два момента.

1. Каждый поток расходует память, а на малом числе ядер создание сотен потоков само по себе стоит времени. Это видно по замеру: при 200 потоках общее время всё равно не 0.5 с, а около секунды.

2. Если потоки ходят в базу через пул соединений (например, SQLAlchemy с pool_size=10), то 100 потоков будут просто стоять в очереди за 10 соединениями. Размер threadpool нужно согласовывать с размерами пулов, в которые он упирается.

Отдельно отмечу asyncio.to_thread: он использует стандартный ThreadPoolExecutor event loop, а не лимитер anyio. Это другой пул с другим размером (min(32, os.cpu_count() + 4)), и в FastAPI‑коде удобнее придерживаться run_in_threadpool из Starlette, чтобы все синхронные вызовы жили в одном понятном пуле.

Шпаргалка: что выбирать

Если сомневаетесь, а асинхронного клиента под рукой нет, def безопаснее: в худшем случае вы упрётесь в лимит пула, а не остановите весь процесс.

Типичные скрытые блокировщики

Список того, что чаще всего незаметно оказывается внутри async def:

• requests, urllib, синхронный клиент httpx.Client;

• синхронные драйверы БД: psycopg2, pymysql, синхронная сессия SQLAlchemy;

• SDK облаков и сервисов (boto3 и многие другие), даже если выглядят безобидно;

• time.sleep() вместо await asyncio.sleep();

• чтение и запись больших файлов через обычный open();

• хеширование паролей, сжатие, криптография;

• синхронный клиент Redis;

• тяжёлая работа с pandas, Pillow и генерация документов.

Как найти проблему в своём проекте

Линтер. В ruff есть набор правил ASYNC (порт flake8-async), который ловит блокирующие HTTP‑вызовы, time.sleep и open() внутри асинхронных функций:

toml

[tool.ruff.lint]
extend-select = ["ASYNC"]

Это самый дешёвый способ: проблема находится ещё до код‑ревью.

Debug‑режим asyncio. С переменной окружения PYTHONASYNCIODEBUG=1 asyncio пишет в лог предупреждения о callback'ах, которые выполнялись дольше slow_callback_duration (по умолчанию 100 мс). Для прода режим тяжеловат, но на стенде помогает быстро найти виновника.

Метрика задержки event loop. В проде я бы держал простой фоновый монитор, который замечает, что цикл «опаздывает»:

import asyncio
import logging

logger = logging.getLogger(__name__)

async def loop_lag_monitor(interval: float = 0.5, threshold: float = 0.1):
    loop = asyncio.get_running_loop()
    while True:
        start = loop.time()
        await asyncio.sleep(interval)
        lag = loop.time() - start - interval
        if lag > threshold:
            logger.warning("event loop lag: %.0f ms", lag * 1000)

Запускается как задача в lifespan, а значение lag удобно отдавать в Prometheus. Если график задержки подскакивает вместе с ростом latency, почти наверняка где‑то сидит блокирующий вызов.

py‑spy. Команда py-spy dump --pid <PID> показывает, что прямо сейчас выполняет каждый поток процесса. Если в момент подвисания главный поток стоит внутри requests или socket.recv, виновник найден.

Итоги

1. async def не делает код быстрее сам по себе. Это договорённость: внутри нет блокирующих вызовов.

2. Один синхронный вызов в async def превращает сервис в однопоточный. В замере это дало падение с ~190 до 4.9 RPS и /health, отвечающий 10 секунд.

3. def безопасен, но все синхронные обработчики и зависимости делят пул из 40 потоков, и быстрые ручки ждут медленных.

4. CPU‑нагрузку не решает ни async, ни threadpool — её нужно выносить в процессы или очередь.

5. Включите правила ASYNC в ruff и следите за задержкой event loop — это поймает большинство проблем до того, как они попадут в прод.

Код стенда: https://github.com/fonartur/fastapi‑async‑vs‑def‑benchmark.