asyncio і FastAPI

Як написати HTTP-сервіс на FastAPI, як asyncio перемикає корутини в одному потоці й чим це відрізняється від async/await у C#, де виконуються ендпоінти і як обрати між потоками, процесами та asyncio. На співбесіді з Python-бекенду це ядро практичних запитань.

Перед цією темою: Мова Python, Асинхронність у .NET.

Junior+

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

Відповідь

Що це

FastAPI — фреймворк для HTTP API на Python. Ендпоінт — функція з декоратором, який задає метод і шлях: @app.get("/items/{item_id}").

Як користуватися

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

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

Типова помилка

Перевіряти типи вручну. Якщо оголошено item_id: int, а прийшло abc, FastAPI сам відповість 422 з описом помилки.

Як влаштовано

Під час старту FastAPI читає сигнатури функцій і будує для кожного маршруту модель параметрів.
На запит він дістає значення зі шляху, рядка запиту, заголовків і тіла та перевіряє їх через Pydantic.
Помилки валідації збираються у відповідь 422: де поле (loc) і що не так (msg).
Результат серіалізується в 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"}

Що спитають далі

Як зробити query-параметр обов’язковим? — Не задавати значення за замовчуванням.
Що поверне FastAPI, якщо в тілі немає обов’язкового поля? — 422 зі списком помилок: loc, msg, type.
Навіщо вказувати тип результату ендпоінта? — Він перевіряє й фільтрує відповідь і потрапляє в OpenAPI-схему.
Як прийняти список у query-параметрі? — Annotated[list[str], Query()], інакше список вважається тілом.
Де подивитися й перевірити API без Postman? — На /docs: Swagger UI будується з тієї самої схеми.

Джерела

Middle+

Як працює asyncio і чим він відрізняється від async/await у C#?

Відповідь

Що це

Корутина — функція async def. Її виклик не виконує код, а створює об’єкт, на який треба дочекатися через await.
Цикл подій — диспетчер в одному потоці: поки одна корутина чекає відповіді мережі, він виконує іншу.
asyncio.run(main()) створює цикл, виконує main і закриває цикл.

Як користуватися

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

Типова помилка

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

Як влаштовано

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

Порівняння з 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())

Що спитають далі

Що буде, якщо викликати async def-функцію без await? — Створиться об’єкт корутини, код не виконається, Python попередить coroutine ... was never awaited.
Чим gather відрізняється від TaskGroup у разі помилки? — gather не скасовує решту задач, TaskGroup скасовує й піднімає ExceptionGroup.
Як викликати синхронну бібліотеку з async-коду? — await asyncio.to_thread(func, *args).
Чому не можна ковтати CancelledError? — TaskGroup, timeout і зупинка сервісу чекають, що скасована задача завершиться.
Чи бувають гонки в однопотоковому asyncio? — Так, якщо стан читають до await, а змінюють після; потрібен asyncio.Lock.

Джерела

Middle+

Як влаштований 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. Поки йде запит, сервер не відповідає нікому.

Як влаштовано

Uvicorn приймає запит і викликає ASGI-застосунок у своєму циклі подій.
FastAPI знаходить маршрут і розв’язує залежності. Кожна викликається один раз на запит, навіть якщо потрібна кільком.
Параметри проходять валідацію Pydantic v2; помилка — 422.
Ендпоінт і залежності async def викликаються в циклі, def — у пулі потоків AnyIO.
Після надсилання відповіді виконується код після 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().

Що спитають далі

Чому def-ендпоінт не блокує сервер, а async def з time.sleep блокує? — def іде в пул потоків, async def виконується в потоці циклу.
Скільки разів викличеться залежність, яка потрібна двом іншим? — Один раз за запит; Depends(dep, use_cache=False) вимикає кеш.
Як підмінити залежність у тесті? — app.dependency_overrides[get_db] = fake_get_db.
Чому lifespan не спрацьовує в тесті? — TestClient запускає його лише в блоці with TestClient(app) as client:.
Що замінило @app.on_event("startup")? — Параметр lifespan; події startup і shutdown застаріли.

Джерела

Senior+

Потоки, процеси чи 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

Один процес на GPU. Модель завантажується в lifespan один раз; кілька --workers — кілька копій ваг у відеопам’яті.
Черга. Ендпоінт кладе запит в обмежену asyncio.Queue і чекає Future. Черга повна — одразу 503, а не нескінченне зростання затримки.
Батчинг. Одна задача-воркер забирає з черги до 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.

Що спитають далі

Чому потоки прискорюють завантаження, але не обчислення? — Під час очікування введення-виведення GIL відпускається, а байткод Python виконує один потік за раз.
Що не можна передати в ProcessPoolExecutor? — Те, що не серіалізується pickle: лямбди, локальні функції, відкриті з’єднання.
Навіщо if __name__ == "__main__": під час роботи з процесами? — При spawn дочірній процес імпортує модуль заново й без перевірки знову запустить пул.
Чому для моделі на GPU не ставлять --workers 4? — Кожен воркер завантажить свою копію ваг у відеопам’ять.
Чим asyncio.to_thread відрізняється від run_in_executor? — to_thread завжди використовує пул потоків циклу й передає контекстні змінні; run_in_executor приймає будь-який виконавець.

Джерела