Веб и API: FastAPI, Django, валидация, асинхронные приложения

Наш сквозной сервис — агрегатор событий — до сих пор жил как библиотека: функции, модели, хранилище. Пора выставить его наружу. Снаружи это выглядит так: клиент открывает TCP-соединение, шлёт POST /events с JSON, ждёт ответа не дольше секунды, и таких клиентов — тысяча одновременно.

Между сокетом и вашей функцией create_event() лежит слой, который обязан:

1. разобрать HTTP (см. «HTTP» в сетевом треке);
2. выбрать обработчик по методу и пути;
3. превратить сырые байты в типизированные объекты и отвергнуть мусор;
4. выполнить ваш код, не заблокировав остальные 999 соединений;
5. сериализовать результат, отдав ровно те поля, которые разрешено отдавать;
6. пережить перезапуск, не оборвав запросы на середине.

Пункты 1, 2, 5 решены давно и одинаково у всех. Настоящая разница между Python-фреймворками — в пунктах 3, 4 и 6. Эта статья — про них.

Азы HTTP и «что такое запрос» здесь не повторяются: если это новая для вас территория, начните с курса «Программирование с нуля». Мы разбираем язык и его экосистему на профессиональном уровне.

Два контракта: WSGI и ASGI

Python-фреймворки не разговаривают с сокетом напрямую. Между сервером и приложением стоит стандартизованный интерфейс — и их ровно два.

WSGI (PEP 3333, 2010 год) — приложение это вызываемый объект от двух аргументов, возвращающий итерируемое тело:

def app(environ, start_response):
    """Минимальное WSGI-приложение целиком. Никакого фреймворка не нужно."""
    body = b'{"ok": true}'
    start_response("200 OK", [
        ("Content-Type", "application/json"),
        ("Content-Length", str(len(body))),
    ])
    return [body]

# Запуск: gunicorn module:app   или   python -m wsgiref.simple_server

Контракт синхронный по определению: пока функция не вернула тело, поток занят. Один одновременный запрос = один поток или процесс.

ASGI (спецификация, 2018 год) — приложение это корутина от трёх аргументов: описания соединения и двух каналов сообщений.

async def app(scope, receive, send):
    """Минимальное ASGI-приложение. scope описывает соединение, send отправляет события."""
    assert scope["type"] == "http"          # ещё бывают "websocket" и "lifespan"

    body = b'{"ok": true}'
    await send({
        "type": "http.response.start",
        "status": 200,
        "headers": [(b"content-type", b"application/json"),
                    (b"content-length", str(len(body)).encode())],
    })
    await send({"type": "http.response.body", "body": body})

# Запуск: uvicorn module:app

Разница не косметическая. Ответ отправляется не одним return, а потоком сообщений — поэтому ASGI умеет то, чего WSGI не умеет принципиально: стриминг, WebSocket, Server-Sent Events, двусторонние протоколы и события жизненного цикла (lifespan) — старт и остановка приложения.

Смешивать нельзя напрямую, но есть адаптеры из asgiref: WsgiToAsgi заворачивает старое приложение в ASGI-сервер, AsgiToWsgi — наоборот. Оба платят потоком на запрос, то есть теряют главное преимущество ASGI. Адаптер — миграционный костыль, а не архитектура.

Модель конкурентности: где на самом деле исполняется ваш код

Главная практическая ловушка ASGI: цикл событий — один поток на воркер. Всё, что в нём выполняется без await, останавливает все соединения этого воркера. Не замедляет — именно останавливает.

FastAPI (точнее, лежащий под ним Starlette) прячет это за удобством: обработчик можно объявить и async def, и обычным def. Но это два разных мира исполнения.

Правило, которое стоит выучить наизусть:

async def — только если внутри честный await на асинхронных библиотеках. Есть синхронный вызов (requests, psycopg2, time.sleep, тяжёлый CPU) — либо объявляйте обработчик обычным def и отдайте его пулу потоков, либо оборачивайте вызов в await asyncio.to_thread(...).

Худший вариант — async def с синхронным вызовом внутри. Он выглядит современно и работает идеально на одном разработчике, а под нагрузкой даёт худшие цифры, чем обычный Flask. Подробный разбор GIL, потоков и цикла событий — в «Конкурентности».

Пул потоков для def-обработчиков в AnyIO по умолчанию ограничен 40 слотами на воркер. Сорок одновременных синхронных запросов — и сорок первый ждёт молча: в стандартных метриках это выглядит как «сервис тормозит», без единой ошибки в логах.

Как выбрать инструмент

Честный выбор без религии:

• Публичный или внутренний JSON-API, много I/O, нужен контракт — FastAPI. Он выиграл именно потому, что превратил аннотации типов в исполняемую спецификацию.
• Продукт с сущностями, ролями, отчётами, админкой и сроком жизни в годы — Django. За один вечер вы получаете то, что на FastAPI собирается неделями: миграции, права, формы, готовый бэкофис.
• Скрипт, вебхук, прототип на два эндпоинта — Flask или голый Starlette. Не тащите фреймворк на 50 зависимостей ради одного POST /webhook.
• Долгоживущие соединения: WebSocket, SSE, чат, стриминг LLM — только ASGI (см. «WebSocket и realtime»).
• gRPC и строгие межсервисные контракты — не веб-фреймворк вообще, а grpcio (см. «RPC и gRPC»). Про выбор стиля API в целом — «Стили API».

Скелет сервиса на FastAPI

Ниже — не игрушечный hello world, а каркас, который можно разворачивать. Всё, что важно, прокомментировано.

from __future__ import annotations

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from datetime import UTC, datetime
from typing import Annotated
from uuid import UUID, uuid4

import asyncpg
from fastapi import Depends, FastAPI, HTTPException, Query, Request, status
from pydantic import BaseModel, ConfigDict, Field


# ---------- контракты данных ----------

class EventIn(BaseModel):
    """Что клиенту РАЗРЕШЕНО прислать. Отдельный класс от того, что мы отдаём."""

    # extra="forbid": опечатка "kynd" даст 422, а не молча потерянное поле
    model_config = ConfigDict(extra="forbid")

    source: Annotated[str, Field(min_length=1, max_length=64)]
    kind: Annotated[str, Field(pattern=r"^[a-z][a-z0-9_]{0,31}$")]
    value: float
    at: datetime | None = None          # None значит "проставь сервером"


class EventOut(BaseModel):
    """Что мы отдаём наружу. Ни одного лишнего поля."""

    id: UUID
    source: str
    kind: str
    value: float
    at: datetime


# ---------- жизненный цикл ----------

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """Пул соединений создаётся ПОСЛЕ форка воркера и закрывается при остановке."""
    app.state.pool = await asyncpg.create_pool(
        dsn="postgresql://app@localhost/events",
        min_size=2, max_size=10, command_timeout=5.0,
    )
    try:
        yield                            # здесь приложение обслуживает запросы
    finally:
        await app.state.pool.close()     # graceful shutdown: дожидаемся возврата соединений


app = FastAPI(title="Агрегатор событий", version="1.0.0", lifespan=lifespan)


# ---------- зависимости ----------

async def get_pool(request: Request) -> asyncpg.Pool:
    return request.app.state.pool

# Псевдоним типа — так зависимость переиспользуется без копипасты Depends(...)
PoolDep = Annotated[asyncpg.Pool, Depends(get_pool)]


# ---------- обработчики ----------

@app.post("/events", response_model=EventOut, status_code=status.HTTP_201_CREATED)
async def create_event(payload: EventIn, pool: PoolDep) -> EventOut:
    # payload уже провалидирован: здесь не нужны никакие if isinstance
    at = payload.at or datetime.now(UTC)
    row = await pool.fetchrow(
        """INSERT INTO events (id, source, kind, value, at)
           VALUES ($1, $2, $3, $4, $5)
           RETURNING id, source, kind, value, at""",
        uuid4(), payload.source, payload.kind, payload.value, at,
    )
    return EventOut.model_validate(dict(row))


@app.get("/events", response_model=list[EventOut])
async def list_events(
    pool: PoolDep,
    source: Annotated[str | None, Query(max_length=64)] = None,
    limit: Annotated[int, Query(ge=1, le=1000)] = 100,
) -> list[EventOut]:
    # ge/le — это не украшение: без верхней границы limit=100000000 положит сервис
    rows = await pool.fetch(
        """SELECT id, source, kind, value, at FROM events
           WHERE ($1::text IS NULL OR source = $1)
           ORDER BY at DESC LIMIT $2""",
        source, limit,
    )
    return [EventOut.model_validate(dict(r)) for r in rows]


@app.get("/events/{event_id}", response_model=EventOut)
async def get_event(event_id: UUID, pool: PoolDep) -> EventOut:
    row = await pool.fetchrow("SELECT * FROM events WHERE id = $1", event_id)
    if row is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Событие не найдено")
    return EventOut.model_validate(dict(row))

Что здесь принципиально:

• Аннотация типа — это и валидация, и документация, и код разбора. event_id: UUID означает, что строка из пути будет разобрана в UUID, а /events/abc вернёт 422 ещё до входа в функцию.
• response_model — фильтр, а не украшение. Он приводит результат к объявленной схеме и выбрасывает всё лишнее. Именно этот механизм не даёт password_hash из ORM-объекта уехать в JSON.
• Пул создаётся в lifespan, а не на уровне модуля. Открытый на импорте сокет переживёт fork() и будет разделён несколькими процессами — классический источник загадочных ошибок протокола (см. «Модули и пакеты»).

Pydantic v2: валидация как граница системы

Pydantic — не «библиотека проверки полей». Это парсер: на входе неизвестные байты, на выходе — объект, про который типы известны точно. Дальше по коду проверок больше нет, потому что невалидное состояние непредставимо.

Вторая версия переписана: ядро pydantic-core — на Rust, и это дало кратный выигрыш по скорости относительно v1 (бенчмарки авторов). Практическое следствие: валидация перестала быть узким местом в типовом API — при условии, что вы её не используете неправильно.

Ограничения через Annotated

from typing import Annotated
from pydantic import BaseModel, Field, StringConstraints

Email = Annotated[str, StringConstraints(strip_whitespace=True, to_lower=True, max_length=254)]
Percent = Annotated[float, Field(ge=0.0, le=100.0)]

class Report(BaseModel):
    owner: Email
    coverage: Percent
    tags: Annotated[list[str], Field(max_length=10)]   # не больше 10 элементов

Annotated отделяет тип от ограничений: Percent — обычный float для mypy и одновременно проверяемое ограничение для pydantic. Про сам механизм — в «Аннотациях типов».

Валидаторы: поля и модель целиком

from pydantic import BaseModel, field_validator, model_validator


class Window(BaseModel):
    start: datetime
    end: datetime
    tz: str = "UTC"

    @field_validator("start", "end")
    @classmethod
    def must_be_aware(cls, v: datetime) -> datetime:
        """Наивные datetime — источник ошибок на границах часовых поясов."""
        if v.tzinfo is None:
            raise ValueError("нужен datetime с таймзоной")
        return v

    @model_validator(mode="after")
    def check_order(self) -> "Window":
        # mode="after" видит уже приведённые типы всех полей
        if self.start >= self.end:
            raise ValueError("start должен быть строго раньше end")
        return self

Внутри валидатора бросайте ValueError или AssertionError — pydantic сам превратит их в структурированную ошибку с путём до поля. Бросать HTTPException из модели нельзя: модель не должна знать про HTTP.

Строгость: во что превращается "5"

По умолчанию pydantic работает в lax-режиме и приводит типы: строка "5" для поля int станет 5, "yes" для bool станет True. Для API, который принимает JSON от чужого клиента, это чаще вред, чем польза.

from pydantic import BaseModel, ConfigDict

class Strict(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid")
    count: int

Strict(count="5")     # ValidationError: Input should be a valid integer

Разумный корпоративный дефолт: extra="forbid" всегда, strict=True — на входных моделях публичного API. Тихо проглоченная опечатка в имени поля — самый дорогой класс багов в JSON-API: клиент уверен, что передал retry_count, сервер работает со значением по умолчанию.

TypeAdapter и разбор без модели

Валидировать можно любой тип, не только BaseModel:

from pydantic import TypeAdapter

EventList = TypeAdapter(list[EventIn])

# Строится один раз (это дорого) и переиспользуется — например, на уровне модуля
events = EventList.validate_json(raw_bytes)   # разбор JSON и валидация в одном проходе

Два практических факта:

• model_validate_json(raw) быстрее, чем json.loads(raw) + model_validate(obj): pydantic разбирает JSON на Rust и сразу строит объект, минуя промежуточные Python-словари.
• Создание TypeAdapter внутри обработчика — распространённая ошибка: каждый запрос заново компилирует схему. Выносите в модуль.

Три модели вместо одной

Соблазн переиспользовать один класс для входа, домена, БД и ответа велик, а расплата приходит через полгода: любое изменение схемы БД ломает публичный контракт.

Да, это больше кода. Но это единственный способ выпустить /v2/events с новым полем, не трогая таблицу, и переименовать колонку в БД, не сломав клиентов. Подробнее о разделении слоёв — в следующей статье и в «Репозиториях».

Зависимости: DI, встроенный в сигнатуру

Depends — механизм, который заменяет FastAPI и контейнер зависимостей, и middleware, и декораторы авторизации. Функция объявляет, что ей нужно, — фреймворк это добывает.

from typing import Annotated
from fastapi import Depends, Header, HTTPException, status


async def current_user(
    authorization: Annotated[str | None, Header()] = None,
) -> User:
    if authorization is None or not authorization.startswith("Bearer "):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, headers={"WWW-Authenticate": "Bearer"})
    return decode_token(authorization.removeprefix("Bearer "))


CurrentUser = Annotated[User, Depends(current_user)]


async def get_uow(pool: PoolDep) -> AsyncIterator[UnitOfWork]:
    """Зависимость с yield: код после yield — это гарантированный teardown."""
    async with pool.acquire() as conn:
        async with conn.transaction():        # коммит при выходе, откат при исключении
            yield UnitOfWork(conn)


@app.post("/events/bulk")
async def bulk(items: list[EventIn], user: CurrentUser,
               uow: Annotated[UnitOfWork, Depends(get_uow)]) -> dict[str, int]:
    for item in items:
        await uow.add(item.to_domain(owner=user.id))
    return {"accepted": len(items)}

Четыре свойства, которые надо знать:

1. Граф, а не список. Зависимости зависят от зависимостей; FastAPI вычисляет порядок сам.
2. Кеш в пределах запроса. Если current_user затребован тремя зависимостями, он вычислится один раз. Отключается через Depends(f, use_cache=False).
3. yield = контекстный менеджер (см. «Идиоматику»). Ресурс освобождается даже при исключении в обработчике.
4. Подмена в тестах одной строкой — app.dependency_overrides[get_uow] = fake_uow.

Важная тонкость: код после yield выполняется до фоновых задач BackgroundTasks (изменение в FastAPI 0.106, документация). То есть сессия БД, полученная через Depends, в фоновой задаче уже закрыта. Фоновой задаче нужен собственный ресурс.

Ошибки: коды, формат, границы

Веб-слой — место, где доменные исключения превращаются в HTTP-коды. Логика этого перевода не должна расползаться по обработчикам.

from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError


class DomainError(Exception):
    """База доменных ошибок — см. статью про исключения."""
    status = 400
    title = "Ошибка запроса"


class QuotaExceeded(DomainError):
    status = 429
    title = "Превышена квота источника"


@app.exception_handler(DomainError)
async def domain_error_handler(request: Request, exc: DomainError) -> JSONResponse:
    # Формат RFC 9457 (Problem Details) — машиночитаемая ошибка вместо {"detail": "..."}
    return JSONResponse(
        status_code=exc.status,
        media_type="application/problem+json",
        content={
            "type": f"https://errors.example.com/{type(exc).__name__.lower()}",
            "title": exc.title,
            "status": exc.status,
            "detail": str(exc),
            "instance": request.url.path,
        },
    )


@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
    return JSONResponse(
        status_code=422,
        media_type="application/problem+json",
        content={"title": "Тело запроса не прошло валидацию",
                 "status": 422, "errors": exc.errors()},
    )

Правила, за которые платят кровью:

• Никогда не отдавайте трейсбек клиенту. debug=True в проде — это выданный наружу исходный код, пути на диске и иногда значения переменных окружения.
• 422 против 400. FastAPI отдаёт 422 на ошибку схемы; 400 оставьте для семантически неверных, но синтаксически корректных запросов. Главное — определиться и задокументировать.
• Ошибка аутентификации — 401, ошибка прав — 403. Не наоборот и не 404 «чтобы не палить существование ресурса», если это не осознанное требование безопасности.
• Исключение внутри middleware не попадает в exception_handler: обработчики исключений сами реализованы как middleware и стоят внутри вашего. Логируйте в middleware явно.

Про иерархию исключений, цепочки и логирование — «Исключения».

Асинхронность на практике

Внешние HTTP-вызовы

requests в асинхронном сервисе — запрещённый приём: он синхронный. Стандарт де-факто — httpx.

import asyncio
import httpx

# Клиент живёт всё время работы приложения: он держит пул TCP-соединений.
# Создавать httpx.AsyncClient на каждый запрос — терять keep-alive и TLS-хендшейки.
client = httpx.AsyncClient(
    limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
    # Таймаут ОБЯЗАТЕЛЕН: по умолчанию httpx ждёт 5 секунд, а requests — вечно
    timeout=httpx.Timeout(connect=2.0, read=5.0, write=5.0, pool=1.0),
)


async def fetch_all(urls: list[str]) -> list[dict | BaseException]:
    """Опрос сотни источников: конкурентно, но с ограничением параллелизма."""
    sem = asyncio.Semaphore(20)          # не больше 20 запросов одновременно

    async def one(url: str) -> dict:
        async with sem:
            resp = await client.get(url)
            resp.raise_for_status()
            return resp.json()

    # return_exceptions=True: одна упавшая ссылка не отменяет остальные 99
    return await asyncio.gather(*(one(u) for u in urls), return_exceptions=True)

Три ошибки, которые делают почти все:

1. Нет семафора. asyncio.gather на 10 000 URL честно откроет 10 000 соединений и упрётся в лимит файловых дескрипторов — либо ваш, либо чужого сервиса.
2. Нет таймаута или он один на всё. Разделяйте connect и read: медленный TLS и медленный ответ — разные аварии с разной реакцией.
3. return_exceptions забыт, и первое же исключение убивает всю пачку, оставляя остальные корутины в подвешенном состоянии.

Retry, circuit breaker и бюджеты таймаутов — в «Паттернах устойчивости».

Блокирующий код в асинхронном сервисе

import asyncio

@app.post("/render")
async def render(payload: RenderIn) -> RenderOut:
    # Pillow, cryptography, парсинг PDF, numpy — всё это синхронный CPU.
    # to_thread снимает блокировку с цикла событий (GIL при этом частично отпускается
    # внутри C-кода библиотеки — см. статью про конкурентность).
    png = await asyncio.to_thread(render_png, payload.template, payload.data)
    return RenderOut(size=len(png))

Для настоящего CPU-bound (чистый Python, десятки миллисекунд и больше) поток не спасёт: GIL не даст параллелизма. Такое выносят в отдельный пул процессов или в очередь задач.

База данных

Асинхронный сервис требует асинхронного драйвера по всей цепочке:

Размер пула — параметр, который считают, а не угадывают: max_size умножается на число воркеров и не должен превышать max_connections базы. Четыре воркера по пулу в 20 соединений против дефолтных 100 у PostgreSQL — это отказ в соединении на пятом инстансе. При десятках инстансов ставьте PgBouncer (см. «Индексы и планы» про то, что делать дальше с запросами).

Жизненный цикл и graceful shutdown

Между SIGTERM и смертью процесса должно пройти достаточно времени, чтобы балансировщик перестал слать трафик (readiness уже 503), а начатые запросы завершились. Типовая ошибка — terminationGracePeriodSeconds меньше, чем таймаут самого долгого запроса: клиенты получают оборванные соединения при каждом деплое (см. «Kubernetes»).

Как ловить блокировки цикла

import asyncio, logging

def install_lag_detector(loop: asyncio.AbstractEventLoop) -> None:
    loop.set_debug(True)
    loop.slow_callback_duration = 0.1   # предупреждать про коллбэки дольше 100 мс

С PYTHONASYNCIODEBUG=1 цикл сам напишет в лог Executing <Task ...> took 0.412 seconds. Это первое, что нужно включить, когда p99 растёт «без причины». Дальше — профилирование (см. «Производительность»).

Django: другая философия

Django решает другую задачу. FastAPI даёт вам транспорт и контракт, всё остальное — ваше. Django даёт продукт: ORM, миграции, аутентификацию, права, формы, шаблоны, админку, интернационализацию, защиту от CSRF и XSS — и требует делать по-своему.

# models.py — схема БД описывается на Python, миграции генерируются
from django.db import models

class Source(models.Model):
    name = models.CharField(max_length=64, unique=True)

class Event(models.Model):
    source = models.ForeignKey(Source, on_delete=models.PROTECT, related_name="events")
    kind = models.CharField(max_length=32, db_index=True)
    value = models.FloatField()
    at = models.DateTimeField(db_index=True)

    class Meta:
        indexes = [models.Index(fields=["source", "-at"])]

Грабли ORM: N+1

Самая частая и самая дорогая ошибка Django-проектов:

# ПЛОХО: 1 запрос за событиями + по одному на каждое обращение к source
for event in Event.objects.all()[:100]:
    print(event.source.name)          # 101 запрос, O(N) обращений к сети

# ХОРОШО: один SQL с JOIN
for event in Event.objects.select_related("source")[:100]:
    print(event.source.name)          # 1 запрос

# Для обратных связей и many-to-many — prefetch_related: 2 запроса вместо N
for source in Source.objects.prefetch_related("events")[:10]:
    print(source.name, len(source.events.all()))

По времени разница не в константе, а в структуре: 101 сетевой round-trip по 1 мс — это 101 мс даже при мгновенной базе. Диагностика — django-debug-toolbar локально и логирование django.db.backends (LOGGING с уровнем DEBUG) в staging.

Ещё три ловушки того же семейства:

qs = Event.objects.filter(kind="click")      # запроса ещё НЕ было: QuerySet ленив
print(qs.count())                            # SELECT COUNT(*)
print(len(qs))                               # второй запрос: SELECT * — и всё в память

# Большая выборка — только итератором, иначе результат целиком окажется в RAM
for e in Event.objects.filter(at__gte=start).iterator(chunk_size=2000):
    process(e)

# Массовая вставка одним запросом вместо N
Event.objects.bulk_create(objects, batch_size=1000)

Async в Django: что реально асинхронно

Django поддерживает ASGI и async def-вьюхи, но ORM внутри асинхронных методов (aget, acreate, acount, async for) по-прежнему исполняет синхронный код в пуле потоков — нативного асинхронного драйвера у Django ORM нет (документация).

from django.http import JsonResponse

async def stats(request):
    total = await Event.objects.acount()                  # под капотом — поток
    recent = [e.kind async for e in Event.objects.order_by("-at")[:10]]
    return JsonResponse({"total": total, "recent": recent})

Практический вывод: async в Django выигрывает там, где вы ждёте внешние сервисы (HTTP, брокеры, LLM-провайдеры) или держите долгие соединения. Ускорения работы с собственной базой он не даёт. Мост между мирами — asgiref:

from asgiref.sync import sync_to_async, async_to_sync

# thread_sensitive=True (по умолчанию) держит вызовы в одном потоке —
# это критично: соединения Django ORM привязаны к потоку
result = await sync_to_async(legacy_function, thread_sensitive=True)(arg)

DRF или django-ninja

DRF — зрелый, огромный, со своим слоем сериализаторов, ViewSet-ов и разрешений; ему больше десяти лет и он умеет всё. django-ninja — FastAPI-подобный слой поверх Django с pydantic-схемами и автогенерацией OpenAPI. Для нового проекта на Django, где нужен именно JSON-API, ninja обычно даёт меньше кода и лучшие типы; DRF выбирают ради экосистемы расширений и когда команда его знает.

Когда Django объективно выигрывает

• Есть внутренние пользователи, которым нужен бэкофис: django.contrib.admin — это недели сэкономленной работы.
• Схема данных живёт и меняется: миграции Django — лучший в экосистеме Python инструмент.
• Нужны сессии, права, группы, восстановление пароля, i18n «прямо сейчас».
• Проект переживёт несколько составов команды: соглашения Django документированы и предсказуемы.

Проигрывает он там, где нужен высокочастотный асинхронный I/O, WebSocket-и в большом количестве или сервис на три эндпоинта.

Безопасность веб-слоя: минимум, который нельзя пропустить

Полный разбор — в треке безопасности, здесь только контрольные точки для Python-сервиса.

• CORS. allow_origins=["*"] вместе с allow_credentials=True — конфигурация, которую браузер отвергнет, а разработчик «починит», разрешив всё подряд. Перечисляйте домены явно.
• Размер тела. Ни uvicorn, ни FastAPI не ограничивают его по умолчанию — лимит ставится на прокси (client_max_body_size в nginx) или собственным middleware. Без него один POST на 4 ГБ выносит воркер по памяти.
• Токены и сессии. JWT без проверки alg, aud и срока — типовая дыра; см. «JWT и токены» и «Аутентификация».
• SQL. Параметризованные запросы всегда — и в asyncpg ($1), и в SQLAlchemy. f-строка с пользовательским вводом в SQL — это инъекция, независимо от фреймворка.
• SSRF. Сервис, который ходит по URL из тела запроса, обязан проверять хост: иначе клиент прочитает ваш метадата-сервис облака.
• Rate limiting делается на входе (nginx, API gateway) — счётчик в памяти Python-воркера бесполезен, потому что воркеров четыре и у каждого свой счётчик.
• Сводный чек-лист — OWASP API Security Top 10 и «Безопасность API».

Фоновая работа

FastAPI умеет BackgroundTasks — задача выполняется после отправки ответа, в том же процессе:

from fastapi import BackgroundTasks

@app.post("/events")
async def create(payload: EventIn, bg: BackgroundTasks) -> dict[str, str]:
    event = await store(payload)
    bg.add_task(notify_webhook, event.id)   # ответ уйдёт раньше, чем выполнится задача
    return {"status": "accepted"}

Это удобно и опасно. BackgroundTasks живёт в памяти воркера: перезапуск, деплой или OOM — и задача исчезла бесследно, без ретраев и без следа в логах. Правило простое: если потеря задачи допустима (отправить метрику, прогреть кеш) — можно. Если нет (списать деньги, отправить письмо, обновить агрегат) — нужна очередь с персистентностью:

И отдельно: если задача обязана произойти вместе с записью в БД (например, «сохранили событие → отправили в шину»), одной очереди мало — нужен паттерн Outbox, иначе транзакция и отправка разъедутся. См. «Событийную архитектуру».

Запуск в проде

# Вариант 1: gunicorn как супервизор + uvicorn-воркеры
# Важно: с uvicorn 0.30 gunicorn-воркер вынесен в отдельный пакет uvicorn-worker
pip install "uvicorn[standard]" uvicorn-worker gunicorn

gunicorn app.main:app \
    --worker-class uvicorn_worker.UvicornWorker \
    --workers 4 \
    --bind 0.0.0.0:8000 \
    --timeout 60 --graceful-timeout 30 \
    --max-requests 2000 --max-requests-jitter 200   # мягкий обход утечек памяти

# Вариант 2: uvicorn сам по себе (проще, если оркестратор уже следит за процессом)
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4 \
    --proxy-headers --forwarded-allow-ips='10.0.0.0/8' \
    --limit-concurrency 512 --timeout-graceful-shutdown 30

Что здесь важно:

• Число воркеров. Для асинхронного сервиса — по числу доступных ядер (в контейнере это cpu.limit, а не os.cpu_count(), который врёт про хост). Больше воркеров не ускорит: ядер всё равно столько же, а память умножится.
• Память. Каждый воркер — отдельный процесс со своей копией интерпретатора и библиотек: типичный API занимает 80–250 МБ, а с pandas или ML-моделью — гигабайты. Считайте workers × RSS до того, как выставите лимит пода.
• Никакого общего состояния между воркерами. Словарь-кеш в модуле существует в четырёх экземплярах: hit rate падает вчетверо, инвалидация не работает, а рассылка по WebSocket дойдёт только до клиентов «своего» воркера. Общее состояние — только во внешнем хранилище (Redis и его pub/sub; см. «Кеширование и масштабирование»).
• --preload у gunicorn экономит память за счёт copy-on-write, но требует, чтобы на импорте не открывалось ни одного сокета: унаследованные через fork() соединения к БД ломаются непредсказуемо.
• --proxy-headers обязателен за балансировщиком, иначе IP клиента в логах — это IP прокси, а request.url соберётся по схеме http вместо https.
• Две разные пробы. /livez не трогает зависимости (иначе моргнувшая база перезапустит все поды разом), /readyz проверяет пул БД и снимается при SIGTERM.

Тестирование веб-слоя

import pytest
from httpx import ASGITransport, AsyncClient
from app.main import app, get_uow


@pytest.fixture
async def client():
    # Подменяем зависимости, а не патчим модули: контракт остаётся типизированным
    app.dependency_overrides[get_uow] = fake_uow_factory
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c
    app.dependency_overrides.clear()


async def test_reject_unknown_field(client):
    resp = await client.post("/events", json={"source": "web", "kynd": "click", "value": 1})
    assert resp.status_code == 422
    assert resp.json()["errors"][0]["loc"] == ["body", "kynd"]

Две ловушки:

• ASGITransport не выполняет lifespan: пулы не поднимутся. Либо подменяйте их через dependency_overrides, либо используйте asgi-lifespan / синхронный TestClient как контекстный менеджер (with TestClient(app) as c: — он lifespan запускает).
• Тесты на реальную БД должны откатывать транзакцию, а не чистить таблицы: иначе они становятся последовательными и медленными.

Подробно — «Тестирование» в этом треке и «Тестирование API».

Производительность и честные границы

Порядки величин (измеряйте у себя, эти числа — только для калибровки ожиданий):

Что реально помогает:

from fastapi.responses import ORJSONResponse

# orjson быстрее стандартного json в разы на сериализации;
# ставится как класс ответа по умолчанию для всего приложения
app = FastAPI(default_response_class=ORJSONResponse)

• Отдавайте большие выборки постранично и не сериализуйте то, что клиент не просил.
• uvloop и httptools (ставятся с uvicorn[standard]) дают заметный прирост бесплатно.
• Кешируйте на границе: HTTP-кеш и Redis экономят больше, чем любая микрооптимизация кода.
• Профилируйте до оптимизации: узкое место чаще в SQL, чем в Python (см. «Производительность» и «Нагрузочное тестирование»).

Где Python в вебе объективно проигрывает. По пропускной способности на ядро — Go, Java и Rust быстрее в разы (для JSON-API типично 3–10×), и по памяти на инстанс тоже. Нет разделяемой памяти между воркерами. Холодный старт с тяжёлыми зависимостями измеряется секундами, что неприятно в serverless. Один блокирующий вызов ломает весь воркер — модель требует дисциплины, которую компилятор не проверит.

Куда Python тащить не стоит: API-шлюзы и прокси на сотни тысяч rps; сервисы с бюджетом латентности меньше миллисекунды; тяжёлая потоковая CPU-обработка в реальном времени без нативных библиотек; сильно ограниченные по памяти edge-среды.

Где выигрывает — и выигрывает крупно: скорость разработки и изменения контрактов; соседство с данными и ML (модель, обученная в том же языке, вызывается без сериализации через границу процессов — см. «Работу с данными»); Django-админка как готовый бэкофис; экосистема интеграций; доступность людей на рынке. Для подавляющего большинства бизнес-API узкое место — база и сеть, а не интерпретатор, и тогда разница в 5× по CPU не стоит разницы в скорости доставки функций.

Грабли, на которые наступают все

1. async def с requests, psycopg2 или time.sleep внутри. Симптом: под нагрузкой растёт p99 у всех эндпоинтов сразу, включая /health.
2. Клиент httpx.AsyncClient создаётся внутри обработчика. Каждый запрос — новый TLS-хендшейк и утечка сокетов. Клиент создаётся в lifespan.
3. Depends() в теле функции, а не в сигнатуре. FastAPI разрешает зависимости только по сигнатуре; в теле это просто объект Depends.
4. Мутабельный дефолт в pydantic-модели. Здесь, в отличие от обычных функций (см. «Функции»), pydantic делает копию — но привычку писать Field(default_factory=list) терять не стоит.
5. ORM-модель используется как response_model. Наружу уезжают технические поля.
6. Порядок маршрутов. /events/{event_id} объявлен раньше /events/stats — и stats уйдёт в параметр пути.
7. Пул соединений создан на уровне модуля. После fork() его делят все воркеры, ошибки протокола появляются «случайно» под нагрузкой.
8. Нет верхней границы у limit, page_size, размера тела и числа элементов в массиве. Отказ в обслуживании одним запросом.
9. BaseHTTPMiddleware для всего. Он ломает стриминговые ответы и усложняет обработку исключений; для сквозной логики пишите чистый ASGI-middleware (документация Starlette).
10. Наивные datetime в API. Всегда UTC с таймзоной на входе и выходе; локальное время — забота клиента.
11. Логирование по строке в каждый обработчик. Нужен один структурный лог доступа с trace-id, а не десять print.
12. Миграции применяются самим приложением на старте. Четыре воркера стартуют одновременно и устраивают гонку на схеме БД. Миграции — отдельный шаг деплоя.

Мини-итог

• Веб-фреймворк — это тонкий слой поверх одного из двух контрактов: WSGI (поток на запрос) или ASGI (цикл событий). Всё остальное следует из этого выбора.
• В ASGI один синхронный вызов останавливает весь воркер. Это главное, что нужно контролировать в асинхронном Python-сервисе.
• FastAPI превращает аннотации типов в исполняемый контракт: валидацию, фильтрацию ответа и OpenAPI. Pydantic v2 достаточно быстр, чтобы это было бесплатно.
• Держите отдельные модели для входа, домена и ответа: это единственный способ менять схему БД и публичный API независимо.
• Django выигрывает продуктовой полнотой — ORM, миграции, админка, права; его async полезен для внешнего I/O, но не ускоряет собственную базу.
• В проде считайте воркеры и память, помните об отсутствии общей памяти между ними, ставьте таймауты на всех уровнях и разводите liveness с readiness.
• Python в вебе проигрывает по CPU и памяти в разы, а выигрывает скоростью изменений и экосистемой. Для типового бизнес-API это выгодный обмен.

Источники

• FastAPI documentation — в частности разделы Dependencies, Bigger Applications и Async.
• Starlette documentation — маршрутизация, middleware, тестирование.
• ASGI specification и PEP 3333 (WSGI) — первоисточники обоих контрактов.
• Pydantic v2 documentation, разделы Validators и Performance.
• Django documentation: Async support, Database optimization.
• uvicorn и gunicorn — настройки запуска, воркеры, сигналы.
• httpx — асинхронный HTTP-клиент, таймауты и лимиты.
• SQLAlchemy 2.0 asyncio и asyncpg.
• RFC 9457: Problem Details for HTTP APIs — стандартный формат ошибок вместо самодельного.
• OWASP API Security Top 10.
• Harry Percival, Bob Gregory, Architecture Patterns with Python — слои, репозитории и сервисы поверх Flask/FastAPI, книга доступна бесплатно онлайн.
• William Vincent, Django for Professionals — продакшн-практики Django.
• Daniel Roy Greenfeld, Audrey Roy Greenfeld, Two Scoops of Django — сборник проверенных соглашений.
• Caleb Hattingh, Using Asyncio in Python — модель исполнения асинхронного кода без магии.

Что дальше

Продакшн-архитектура: слои, зависимости, конфигурация, логирование — как разложить сервис по слоям так, чтобы веб-фреймворк оставался деталью реализации: порты и адаптеры без тяжёлого DI-контейнера, конфигурация через окружение и pydantic-settings, структурное логирование с trace-id и границы, за которыми домен не знает ни про HTTP, ни про SQL.