Перейти к содержимому
Подготовка к собеседованию

asyncio и FastAPI

Как написать HTTP-сервис на FastAPI, как asyncio переключает корутины в одном потоке и чем это отличается от async/await в C#, где выполняются эндпоинты и как выбрать между потоками, процессами и asyncio. На собеседовании по Python-бэкенду это ядро практических вопросов.

Перед этой темой: Язык Python, Асинхронность в .NET.

Junior+

Как написать простой REST-сервис на FastAPI?

Заголовок раздела «Как написать простой REST-сервис на FastAPI?»
Ответ

Что это

FastAPI — фреймворк для HTTP API на Python. Эндпоинт — функция с декоратором, который задаёт метод и путь: @app.get("/items/{item_id}").

Как пользоваться

  1. Параметр с именем из пути (item_id) — path-параметр.
  2. Параметр простого типа, которого нет в пути, — query: ?currency=USD. Значение по умолчанию делает его необязательным.
  3. Параметр-модель Pydantic — JSON-тело запроса.
  4. Код ответа задают в декораторе (status_code=201), ошибку — исключением HTTPException(status_code=404).

Пакет ставят командой pip install "fastapi[standard]", запускают fastapi dev main.py, документация открывается на /docs.

Частая ошибка

Проверять типы вручную. Если объявлено item_id: int, а пришло abc, FastAPI сам ответит 422 с описанием ошибки.

Как устроено

  1. При старте FastAPI читает сигнатуры функций и строит для каждого маршрута модель параметров.
  2. На запрос он достаёт значения из пути, строки запроса, заголовков и тела и проверяет их через Pydantic.
  3. Ошибки валидации собираются в ответ 422: где поле (loc) и что не так (msg).
  4. Результат сериализуется в JSON. Если указан возвращаемый тип или response_model, ответ проверяется и фильтруется по этой модели.

Откуда берётся параметр

Объявление Источник
имя есть в шаблоне пути путь
int, str, bool и другие простые типы строка запроса
модель Pydantic JSON-тело
Query(), Header(), Body() в Annotated указан явно

Ловушки

  • Списки. list[str] без Query() считается телом, а не query-параметром.
  • Мягкая валидация. По умолчанию Pydantic приводит типы: строка "2" станет числом 2. Нужна строгость — Field(strict=True) или ConfigDict(strict=True).
  • Content-Type. С FastAPI 0.132 JSON-тело без заголовка Content-Type: application/json отклоняется с 422.

Правило

Контракт описывайте типами и моделями, а не ручными проверками: тогда валидация, документация и код не расходятся.

Под капотом

  • Класс FastAPI наследует Starlette: маршрутизация, middleware и работа с HTTP — оттуда. FastAPI добавляет разбор параметров, зависимости и OpenAPI.
  • Схемы параметров и ответов строятся один раз при старте. На запросе работает валидация в pydantic-core — ядре Pydantic на Rust.
  • С FastAPI 0.130 ответ с Pydantic-моделью или объявленным возвращаемым типом сериализуется в JSON прямо в Rust-ядре Pydantic.
  • Документ OpenAPI 3.1 отдаётся на /openapi.json, интерфейсы — на /docs (Swagger UI) и /redoc.

Цена

  • Разбор и валидация тела выполняются на каждом запросе. Огромный JSON заметно нагружает CPU процесса.
  • Модель ответа — вторая проверка: данные валидируются на выходе ещё раз.

Компромиссы

  • Возвращаемый тип — защита от утечки: поля, которых нет в модели ответа, например password_hash, в JSON не попадут. Но та же фильтрация молча теряет поля, если модель забыли обновить.
  • Отдельные модели входа и выхода (ItemCreate, ItemOut) — то же, что разные DTO запроса и ответа в ASP.NET Core.
  • Публичную документацию в проде часто отключают: FastAPI(docs_url=None, redoc_url=None).

Как проверить в проде

  • Контракт фиксируют тестом: TestClient получает /openapi.json и сравнивает со снимком, поэтому поломка API видна в CI.
  • Всплеск 422 в метриках — клиенты шлют не то, что ожидает схема.

Итог: FastAPI превращает сигнатуру в контракт API; платят за это валидацией на каждом запросе и дисциплиной в моделях ответа.

Пример
# pip install "fastapi[standard]"; запуск: fastapi dev main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field


app = FastAPI()


class Item(BaseModel):                 # модель тела запроса
    name: str
    price: float = Field(gt=0)         # проверка: цена больше нуля


items: dict[int, Item] = {}


@app.post("/items", status_code=201)
async def create_item(item: Item) -> dict:       # тело — модель Pydantic
    item_id = len(items) + 1
    items[item_id] = item
    return {"id": item_id, **item.model_dump()}


@app.get("/items/{item_id}")
async def read_item(item_id: int, currency: str = "UAH") -> dict:  # path и query
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"name": items[item_id].name, "currency": currency}


# POST /items {"name": "Tea", "price": 42} → 201 {"id": 1, "name": "Tea", "price": 42.0}
# GET /items/1?currency=USD → 200 {"name": "Tea", "currency": "USD"}
# GET /items/abc → 422: item_id не число;  GET /items/2 → 404 {"detail": "Item not found"}
Источники
Middle+

Как работает asyncio и чем он отличается от async/await в C#?

Заголовок раздела «Как работает asyncio и чем он отличается от async/await в C#?»
Ответ

Что это

  • Корутина — функция async def. Её вызов не выполняет код, а создаёт объект, который нужно дождаться через await.
  • Цикл событий — диспетчер в одном потоке: пока одна корутина ждёт ответа сети, он выполняет другую.
  • asyncio.run(main()) создаёт цикл, выполняет main и закрывает цикл.

Как пользоваться

  1. await пишут только внутри async def.
  2. Чтобы операции шли одновременно, запускают их вместе: await asyncio.gather(a(), b()) или через TaskGroup.
  3. Пауза — await asyncio.sleep(1), а не time.sleep(1).

Частая ошибка

Вызвать блокирующую функцию (time.sleep, requests.get) внутри async def. Поток один, поэтому встанут все корутины.

Как устроено

  1. asyncio.create_task(coro) ставит корутину в очередь цикла. Без задачи или await корутина не начнёт работу.
  2. Задача выполняется до await на незавершённой операции, затем цикл берёт следующую готовую.
  3. Когда ввод-вывод завершился, цикл возобновляет ждущую задачу с того же места.

Сравнение с C#

Признак asyncio C#
Где выполняется один поток цикла продолжения в пуле потоков
Запуск корутина ленивая Task стартует при вызове
Ждать несколько gather, TaskGroup Task.WhenAll
Таймаут asyncio.timeout() CancelAfter, WaitAsync
Отмена task.cancel() CancellationToken
Блокирующий вызов стоит весь цикл занят один поток пула

Ловушки

  • Потерянная задача. Цикл держит на задачи только слабые ссылки: задачу без сохранённой ссылки сборщик мусора может удалить до завершения.
  • gather при ошибке отдаёт первое исключение, а остальные задачи продолжают работать. TaskGroup отменяет оставшиеся и собирает ошибки в ExceptionGroup, который ловят через except*.
  • Проглоченная отмена. CancelledError наследует BaseException. Поймали его ради очистки — пробросьте дальше через raise.

Правило

Конкурентность — через TaskGroup и asyncio.timeout (3.11+), блокирующий код — через asyncio.to_thread.

Под капотом

  • Задача продвигает корутину методом send(). Дойдя до незавершённого Future, корутина отдаёт его задаче, та подписывается на завершение и возвращает управление циклу.
  • Кадр корутины хранит локальные переменные сам. Компилятор не генерирует отдельный класс, как state machine в C#.
  • Итерация цикла: выполнить готовые колбэки, спросить ОС о готовом вводе-выводе (epoll в Linux, IOCP через ProactorEventLoop в Windows), запланировать продолжения.
  • cancel() не прерывает поток, а бросает CancelledError при следующем возобновлении. Счётчик cancelling() и метод uncancel() (3.11) помогают TaskGroup и timeout отличить свою отмену от внешней.

Цена

  • Задача — объект в куче без собственного стека, поэтому десятки тысяч задач — норма.
  • Код между await выполняется монопольно: разбор JSON на 200 мс задерживает все запросы процесса на 200 мс.

Компромиссы

  • Гонок внутри участка без await нет, но состояние, которое читают до await и меняют после, требует asyncio.Lock.
  • Синхронный мир — драйверы, SDK — подключается через to_thread, то есть через обычные потоки со всеми их ограничениями.

Как проверить в проде

  • Режим отладки (PYTHONASYNCIODEBUG=1 или asyncio.run(main(), debug=True)) логирует колбэки дольше 100 мс — так находят блокирующие вызовы.
  • С Python 3.14 python -m asyncio ps PID и pstree PID показывают задачи живого процесса и кто кого ждёт.
  • Task exception was never retrieved в логе — задачу запустили и не дождались.

Итог: asyncio — один поток с явными точками переключения; его надёжность держится на структурной конкурентности, честной отмене и отсутствии блокирующих вызовов.

Пример
import asyncio
import time


async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)              # отдаёт управление циклу событий
    return name


async def main() -> None:
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:   # Python 3.11+
        a = tg.create_task(fetch("a", 1))
        b = tg.create_task(fetch("b", 1))
    print(a.result(), b.result(), round(time.perf_counter() - start, 1))  # a b 1.0


    try:
        async with asyncio.timeout(0.5):    # Python 3.11+
            await fetch("slow", 2)
    except TimeoutError:
        print("timeout")                    # timeout


    start = time.perf_counter()
    await asyncio.gather(asyncio.to_thread(time.sleep, 1), fetch("c", 1))
    print(round(time.perf_counter() - start, 1))  # 1.0: блокирующий sleep ушёл в поток


asyncio.run(main())

Если вызвать time.sleep(1) прямо в корутине, gather из последнего блока займёт 2 секунды: цикл стоит, пока поток спит.

Отмена задачи
import asyncio


async def worker() -> None:
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("cleanup")            # освобождаем ресурсы
        raise                       # и обязательно пробрасываем дальше


async def main() -> None:
    task = asyncio.create_task(worker())
    await asyncio.sleep(0.1)        # даём задаче начать работу
    task.cancel()                   # CancelledError возникнет в точке await
    try:
        await task
    except asyncio.CancelledError:
        print(task.cancelled())     # True


asyncio.run(main())
Источники
Middle+

Как устроен FastAPI?

Заголовок раздела «Как устроен FastAPI?»
Ответ

Что это

Приложение FastAPI запускает ASGI-сервер Uvicorn. Сервер держит соединения, FastAPI обрабатывает запросы.

Как пользоваться

  • Зависимости. Общий код — сессию БД, текущего пользователя — пишут функцией и подключают параметром Annotated[Session, Depends(get_db)]. FastAPI вызовет её на каждый запрос.
  • async def или def. Внутри только await-вызовы — async def. Синхронная библиотека (requests, драйвер БД без async) — def, FastAPI выполнит его в пуле потоков.
  • Запуск. fastapi dev main.py — разработка с перезагрузкой, fastapi run main.py — прод.
  • Ошибки. raise HTTPException(status_code=404, detail="...") превращается в JSON {"detail": "..."}.

Частая ошибка

requests.get внутри async def. Пока идёт запрос, сервер не отвечает никому.

Как устроено

  1. Uvicorn принимает запрос и вызывает ASGI-приложение в своём цикле событий.
  2. FastAPI находит маршрут и решает зависимости. Каждая вызывается один раз на запрос, даже если нужна нескольким.
  3. Параметры проходят валидацию Pydantic v2; ошибка — 422.
  4. Эндпоинт и зависимости async def вызываются в цикле, def — в пуле потоков AnyIO.
  5. После отправки ответа выполняется код после yield в зависимостях.

Сравнение с ASP.NET Core

ASP.NET Core FastAPI
регистрация в builder.Services нет: зависимость — функция в сигнатуре
Scoped с Dispose зависимость с yield и finally
Singleton, IHostedService объект из lifespan
ConfigureTestServices в тестах app.dependency_overrides
исключения и ProblemDetails HTTPException, @app.exception_handler

Ловушки

  • Пул потоков не резиновый. def-эндпоинты и зависимости делят 40 потоков: если все ждут медленный сервис, новые запросы стоят в очереди.
  • Lifespan в каждом воркере. --workers 4 — четыре пула соединений и четыре копии модели в памяти.
  • except без raise в зависимости с yield прячет исключение эндпоинта от FastAPI.

Правило

async def — когда все вызовы асинхронные, def — когда есть блокирующие, тяжёлые вычисления — вне процесса веб-сервера.

Под капотом

  • Маршрут — APIRoute: при старте он строит дерево зависимостей и модели Pydantic для параметров и ответа.
  • Функции def уходят в run_in_threadpool Starlette, то есть в anyio.to_thread.run_sync с общим лимитом 40 потоков на процесс.
  • uvicorn --workers N или fastapi run --workers N — менеджер процессов: каждый воркер импортирует приложение, выполняет lifespan и слушает общий порт.

Цена

  • def — переход в поток и обратно на каждый вызов. Документация FastAPI прямо советует тривиальные функции без блокирующего ввода-вывода объявлять async def.
  • 40 потоков — потолок одновременных синхронных вызовов в процессе. Его поднимают через anyio.to_thread.current_default_thread_limiter().total_tokens в lifespan.
  • Воркеры не делят память: кеши, пулы соединений и модели множатся на N.

Компромиссы

  • Воркеры дают параллелизм по CPU, но в Kubernetes обычно запускают один процесс на контейнер и масштабируют репликами.
  • Зависимость с yield по умолчанию закрывается после отправки ответа. С FastAPI 0.121 Depends(get_db, scope="function") закрывает её сразу после эндпоинта — например, чтобы раньше вернуть соединение в пул.

Как проверить в проде

  • Задержка растёт, а CPU низкий — блокирующий вызов в async def или исчерпанный пул потоков для def.
  • py-spy dump --pid PID показывает стеки всех потоков: видно, что держит поток цикла.

Итог: FastAPI — Starlette с графом зависимостей по сигнатурам; производительность решают выбор async def или def, лимит пула потоков и число процессов.

Пример
from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import Depends, FastAPI, Request


class Model:                            # дорогой объект: один на процесс
    def predict(self, text: str) -> str:
        return text.upper()


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.model = Model()           # до первого запроса
    yield
    app.state.model = None              # при остановке сервера


app = FastAPI(lifespan=lifespan)


async def get_model(request: Request) -> Model:
    return request.app.state.model


async def get_session():
    session = {"open": True}            # например, сессия БД
    try:
        yield session                   # отдаём зависимость эндпоинту
    finally:
        session["open"] = False         # закрываем после ответа


@app.get("/predict")
async def predict(
    text: str,
    model: Annotated[Model, Depends(get_model)],
    session: Annotated[dict, Depends(get_session)],
) -> dict:
    return {"result": model.predict(text), "session": session["open"]}
# GET /predict?text=hi → {"result": "HI", "session": true}

В тесте модель подменяют без правки кода: app.dependency_overrides[get_model] = lambda: FakeModel().

Источники
Senior+

Потоки, процессы или asyncio: что выбрать для разных задач?

Заголовок раздела «Потоки, процессы или asyncio: что выбрать для разных задач?»
Ответ

Что это

  • I/O-bound — задача ждёт сеть, диск или базу. Процессор простаивает.
  • CPU-bound — задача считает: парсинг, сжатие, математика.
  • GIL — блокировка в CPython: байткод Python в процессе выполняет один поток за раз.

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

Задача Инструмент
много HTTP-запросов, есть async-библиотека asyncio
блокирующая библиотека без async потоки: ThreadPoolExecutor, asyncio.to_thread
вычисления на чистом Python процессы: ProcessPoolExecutor

Частая ошибка

Ускорять вычисления потоками. В обычной сборке CPython из-за GIL четыре потока считают не быстрее одного.

Как устроено

  • Поток, который ждёт ввод-вывод, отпускает GIL, поэтому потоки годятся для блокирующего I/O. Пул asyncio.to_thread по умолчанию — min(32, число CPU + 4) потоков.
  • Процесс — отдельный интерпретатор со своей памятью и своим GIL. Аргументы и результат передаются через pickle, поэтому функция должна быть объявлена на уровне модуля.
  • Из async-кода: await asyncio.to_thread(f, x) — поток, await loop.run_in_executor(pool, f, x) — любой пул, включая процессы.

Ловушки

  • Способ запуска процесса. В Windows и macOS это spawn: дочерний процесс заново импортирует модуль, поэтому код запуска прячут под if __name__ == "__main__":. В Linux с Python 3.14 по умолчанию forkserver вместо fork.
  • Большие данные. Массив на гигабайт в аргументе — это сериализация и копия. Помогают multiprocessing.shared_memory или передача пути к файлу.
  • Мелкие задачи. Если вызов короче накладных расходов на pickle, процессы только замедляют. Задачи укрупняют, например chunksize в pool.map.
  • Пул на каждый запрос. Пул процессов создают один раз при старте, а не в эндпоинте.

Правило

I/O с async-клиентом — asyncio, I/O с синхронным клиентом — to_thread, CPU на чистом Python — пул процессов, созданный один раз.

Под капотом

  • Многие C-расширения отпускают GIL на время тяжёлой работы, например NumPy и PyTorch во многих операциях. Такой код масштабируется и в потоках.
  • GPU — общий ресурс со своей очередью команд. Параллельные вызовы модели из разных потоков не ускоряют инференс, а конкурируют за видеопамять.

Инференс на GPU за HTTP

  1. Один процесс на GPU. Модель загружается в lifespan один раз; несколько --workers — несколько копий весов в видеопамяти.
  2. Очередь. Эндпоинт кладёт запрос в ограниченную asyncio.Queue и ждёт Future. Очередь полна — сразу 503, а не бесконечный рост задержки.
  3. Батчинг. Одна задача-воркер забирает из очереди до N запросов и запускает модель на пачке через to_thread, чтобы цикл событий продолжал принимать запросы.

Компромиссы

  • Размер пачки — пропускная способность против задержки: GPU загружен лучше, но первый запрос в пачке ждёт остальных.
  • Серверы инференса вроде NVIDIA Triton и vLLM умеют батчинг сами. Свой код оправдан, пока модель одна и логика простая.
  • Free-threaded CPython (экспериментальный в 3.13, официально поддерживается с 3.14 отдельной сборкой) позволяет потокам считать параллельно, но требует поддержки от C-расширений. Подробнее — в теме Язык Python.

Как проверить в проде

  • Одно ядро загружено на 100 %, остальные простаивают — код упирается в GIL.
  • Утилизация GPU в nvidia-smi низкая при растущей очереди — пачки слишком мелкие или подготовка данных блокирует цикл.
  • py-spy dump --pid PID показывает стеки всех потоков: видно, кто занят и кто ждёт.

Итог: модель конкурентности выбирают по тому, где тратится время, а GPU обслуживает один процесс с очередью и батчингом.

Пример
import asyncio
import time
from concurrent.futures import ProcessPoolExecutor


def cpu_heavy(n: int) -> int:           # чистый Python: держит GIL всё время
    return sum(i * i for i in range(n))


def blocking_io() -> str:               # библиотека без async API
    time.sleep(0.5)                     # на ожидании GIL отпускается
    return "done"


async def main() -> None:
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor(max_workers=2) as pool:
        results = await asyncio.gather(
            loop.run_in_executor(pool, cpu_heavy, 5_000_000),  # отдельный процесс
            loop.run_in_executor(pool, cpu_heavy, 5_000_000),  # и свой GIL
            asyncio.to_thread(blocking_io),                    # поток из пула цикла
        )
    print(results[2])                   # done


if __name__ == "__main__":              # обязательно при spawn (Windows, macOS)
    asyncio.run(main())
Очередь и батчинг для GPU · Senior
import asyncio


queue: asyncio.Queue = asyncio.Queue(maxsize=100)  # ограничена: обратное давление


def run_model(batch: list[str]) -> list[str]:  # инференс на GPU, блокирующий вызов
    return [text.upper() for text in batch]


async def gpu_worker(max_batch: int = 8) -> None:
    while True:
        batch = [await queue.get()]             # ждём первый запрос
        while len(batch) < max_batch and not queue.empty():
            batch.append(queue.get_nowait())    # добираем накопившиеся
        results = await asyncio.to_thread(run_model, [t for t, _ in batch])
        for (_, future), result in zip(batch, results):
            if not future.done():               # клиент мог отменить запрос
                future.set_result(result)


async def infer(text: str) -> str:              # вызывается из эндпоинта
    future = asyncio.get_running_loop().create_future()
    await queue.put((text, future))
    return await future


async def main() -> None:
    worker = asyncio.create_task(gpu_worker())  # единственный, кто трогает GPU
    print(await asyncio.gather(infer("a"), infer("b"), infer("c")))  # ['A', 'B', 'C']
    worker.cancel()


asyncio.run(main())

В сервисе воркер запускают в lifespan, а ошибку модели передают ожидающим через future.set_exception.

Источники