Массовые операции, долгие задачи и локализация в FastAPI

This article is currently available in Russian only — an English translation is in progress.

Три ситуации, которые выходят за рамки обычного CRUD: нужно обработать сразу несколько объектов, запустить длительную задачу и дать клиенту её отслеживать, или отдавать сообщения об ошибках на языке пользователя. Разберём каждую по порядку.

Массовые операции

Проблема

Допустим, нужно создать сразу 50 заказов. Слать 50 отдельных запросов — дорого. Можно отправить один запрос со списком. Но что делать, если 3 позиции из 50 не прошли проверку? Прерывать всё или продолжить остальные?

В большинстве случаев правильный ответ — продолжить. Такое поведение называется partial success: каждый элемент обрабатывается независимо, ошибка одного не останавливает остальные. Сервер возвращает 200 OK с результатом по каждому элементу.

Endpoint и модели

Массовая операция оформляется как отдельный маршрут. Запрос содержит список элементов, ответ — список результатов и итоговую статистику.

router = APIRouter(prefix="/api/v1")

@router.post(
    "/orders/batch",
    status_code=200,
    operation_id="batchCreateOrders",
    tags=["Orders"],
    summary="Массовое создание заказов (partial success)",
)
async def batch_create_orders(
    request: BatchCreateOrdersRequest,
    service: OrderService = Depends(get_order_service),
) -> BatchCreateOrdersResponse:
    return await service.batch_create(request.items)

Pydantic-модели для запроса и ответа:

from __future__ import annotations

from enum import StrEnum
from pydantic import BaseModel, ConfigDict, Field
from pydantic.alias_generators import to_camel


class ItemStatus(StrEnum):
    SUCCESS = "SUCCESS"
    ERROR = "ERROR"


class BatchItemError(BaseModel):
    model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)

    code: str
    detail: str


class BatchCreateOrderItem(BaseModel):
    model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)

    product_id: str
    quantity: int = Field(ge=1)


class BatchCreateOrderResult(BaseModel):
    model_config = ConfigDict(
        alias_generator=to_camel,
        populate_by_name=True,
        exclude_none=True,
    )

    index: int
    status: ItemStatus
    order_id: str | None = None
    error: BatchItemError | None = None


class BatchSummary(BaseModel):
    model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)

    total: int
    succeeded: int
    failed: int


class BatchCreateOrdersRequest(BaseModel):
    model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)

    items: list[BatchCreateOrderItem]


class BatchCreateOrdersResponse(BaseModel):
    model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)

    results: list[BatchCreateOrderResult]
    summary: BatchSummary

exclude_none=True на BatchCreateOrderResult убирает из ответа поле orderId при ошибке и error при успехе — JSON остаётся чистым.

Как выглядит запрос и ответ

POST /api/v1/orders/batch
Content-Type: application/json

{
  "items": [
    { "productId": "SKU-001", "quantity": 2 },
    { "productId": "SKU-002", "quantity": 1 },
    { "productId": "SKU-003", "quantity": 5 }
  ]
}

HTTP/1.1 200 OK
Content-Type: application/json

{
  "results": [
    { "index": 0, "status": "SUCCESS", "orderId": "ORD-881" },
    { "index": 1, "status": "ERROR", "error": { "code": "INSUFFICIENT_STOCK", "detail": "Товар SKU-002 отсутствует на складе" } },
    { "index": 2, "status": "SUCCESS", "orderId": "ORD-882" }
  ],
  "summary": { "total": 3, "succeeded": 2, "failed": 1 }
}

200 OK возвращается даже при частичной неудаче — это не ошибка запроса, а ожидаемый результат. Поле index соответствует позиции в исходном списке items (отсчёт от 0).

Ограничение размера

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

MAX_BATCH_SIZE = 100

@router.post("/orders/batch", ...)
async def batch_create_orders(
    request: BatchCreateOrdersRequest, ...
) -> BatchCreateOrdersResponse:
    if len(request.items) > MAX_BATCH_SIZE:
        raise BatchSizeExceededException(
            actual=len(request.items), max_allowed=MAX_BATCH_SIZE
        )
    ...

HTTP/1.1 400 Bad Request
Content-Type: application/problem+json

{
  "type": "urn:problem:order-service:batch-size-exceeded",
  "status": 400,
  "title": "Bad Request",
  "detail": "Размер превышает максимум (100 элементов), передано: 150",
  "code": "BATCH_SIZE_EXCEEDED"
}

Атомарность

По умолчанию — partial success. Если конкретная операция должна быть атомарной (ошибка одного элемента отменяет всё), это нужно явно оговорить в описании API. Тогда при неудаче возвращается 400 BATCH_TRANSACTION_FAILED с перечнем проблемных индексов.

Долгие задачи: запуск и отслеживание

Проблема

Некоторые операции занимают секунды или минуты: генерация отчёта, экспорт данных, перекодирование файла. Держать HTTP-соединение открытым на всё это время — плохая идея. Клиент получит таймаут.

Решение — асинхронная обработка с отслеживанием: клиент запускает задачу, получает идентификатор и периодически спрашивает о состоянии.

Запуск задачи — 202 Accepted

from fastapi import Response
from fastapi.responses import JSONResponse


@router.post(
    "/reports/generate",
    status_code=202,
    operation_id="generateReport",
    tags=["Reports"],
    summary="Запуск генерации отчёта",
)
async def generate_report(
    request: GenerateReportRequest,
    response: Response,
    service: ReportService = Depends(get_report_service),
) -> TaskAcceptedResponse:
    task = await service.submit_generate(request)
    status_url = f"/api/v1/tasks/{task.task_id}"
    response.headers["Location"] = status_url
    return TaskAcceptedResponse(
        task_id=task.task_id,
        status=TaskStatus.PENDING,
        created_at=task.created_at,
        status_url=status_url,
    )

class TaskStatus(StrEnum):
    PENDING = "PENDING"
    PROCESSING = "PROCESSING"
    COMPLETED = "COMPLETED"
    FAILED = "FAILED"


class TaskAcceptedResponse(BaseModel):
    model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)

    task_id: str
    status: TaskStatus
    created_at: datetime
    status_url: str

Ответ сервера:

HTTP/1.1 202 Accepted
Location: /api/v1/tasks/550e8400-e29b-41d4-a716-446655440000

{
  "taskId": "550e8400-e29b-41d4-a716-446655440000",
  "status": "PENDING",
  "createdAt": "2026-06-19T10:30:00Z",
  "statusUrl": "/api/v1/tasks/550e8400-e29b-41d4-a716-446655440000"
}

Заголовок Location указывает адрес для отслеживания. Поле statusUrl дублирует его в теле — для клиентов, которые не читают заголовки ответа.

Сервер может также добавить заголовок Retry-After с рекомендуемым интервалом опроса в секундах:

response.headers["Retry-After"] = "5"

Отслеживание состояния — GET /tasks/{id}

class TaskErrorDetail(BaseModel):
    model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)

    code: str
    detail: str


class TaskStatusResponse(BaseModel):
    model_config = ConfigDict(
        alias_generator=to_camel,
        populate_by_name=True,
        exclude_none=True,
    )

    task_id: str
    status: TaskStatus
    progress: int | None = None
    created_at: datetime
    completed_at: datetime | None = None
    result_url: str | None = None
    error: TaskErrorDetail | None = None


@router.get(
    "/tasks/{task_id}",
    operation_id="getTask",
    tags=["Tasks"],
    summary="Статус задачи",
)
async def get_task(
    task_id: str,
    service: TaskService = Depends(get_task_service),
) -> TaskStatusResponse:
    return await service.get_status(task_id)

Ответы на разных этапах выглядят так:

// Задача выполняется
{ "taskId": "550e8400-...", "status": "PROCESSING", "progress": 45, "createdAt": "2026-06-19T10:30:00Z" }

// Задача завершена успешно
{
  "taskId": "550e8400-...",
  "status": "COMPLETED",
  "progress": 100,
  "createdAt": "2026-06-19T10:30:00Z",
  "completedAt": "2026-06-19T10:35:00Z",
  "resultUrl": "/api/v1/reports/550e8400-..."
}

// Задача завершена с ошибкой
{
  "taskId": "550e8400-...",
  "status": "FAILED",
  "createdAt": "2026-06-19T10:30:00Z",
  "completedAt": "2026-06-19T10:32:00Z",
  "error": { "code": "REPORT_GENERATION_FAILED", "detail": "Данные за период отсутствуют" }
}

При COMPLETED обязательно присутствует resultUrl — ссылка на результат. При FAILED обязательно присутствует error с описанием проблемы. exclude_none=True убирает поля, которые ещё не заполнены.

Локализация ответов

Проблема

Пользователь из Германии видит сообщение об ошибке «Order not found». Хочется отдавать текст на языке пользователя. Но как передать язык, и что именно переводить?

Заголовок Accept-Language

Клиент указывает язык в стандартном HTTP-заголовке Accept-Language. FastAPI читает его напрямую:

from fastapi import Header


@router.get("/orders/{order_id}", operation_id="getOrder", tags=["Orders"])
async def get_order(
    order_id: str,
    accept_language: str = Header(default="ru", alias="Accept-Language"),
    service: OrderService = Depends(get_order_service),
) -> OrderResponse:
    return await service.get(order_id, locale=accept_language)

Если заголовок не передан — используется язык по умолчанию ru.

Что переводить, а что нет

Переводятся только сообщения для пользователя: поле detail в ответе об ошибке и message в списке нарушений валидации.

def localized_detail(key: str, locale: str) -> str:
    translations = {
        "ORDER_NOT_FOUND": {
            "ru": "Заказ не найден",
            "en": "Order not found",
        },
        "CUSTOMER_BLOCKED": {
            "ru": "Клиент заблокирован",
            "en": "Customer is blocked",
        },
    }
    return translations.get(key, {}).get(locale, translations.get(key, {}).get("ru", key))

// Accept-Language: ru
{ "code": "ORDER_NOT_FOUND", "detail": "Заказ не найден" }

// Accept-Language: en
{ "code": "ORDER_NOT_FOUND", "detail": "Order not found" }

Поля code, title, type и имена JSON-полей не переводятся. code — машиночитаемый идентификатор, по которому клиентский код делает switch. Если он окажется на кириллице или изменится при смене языка, клиент сломается.

// Правильно
{ "code": "PRODUCT_OUT_OF_STOCK", "title": "Bad Request", "detail": "Товар снят с продажи" }

// Частая ошибка
{ "code": "ТОВАР_НЕТ", "title": "Неверный запрос" }

Локализация в обработчике исключений

from fastapi import Request
from fastapi.responses import Response
import json


async def order_not_found_handler(request: Request, exc: OrderNotFoundException) -> Response:
    locale = request.headers.get("Accept-Language", "ru")
    detail = localized_detail("ORDER_NOT_FOUND", locale)
    body = {
        "type": "urn:problem:order-service:order-not-found",
        "status": 404,
        "title": "Not Found",
        "detail": detail,
        "code": "ORDER_NOT_FOUND",
    }
    return Response(
        content=json.dumps(body, ensure_ascii=False),
        status_code=404,
        media_type="application/problem+json",
    )

Коротко

Массовые операции возвращают 200 OK даже при частичной неудаче — каждый элемент обрабатывается независимо.
Ответ содержит results (результат по каждому элементу) и summary (итого: всего / успешно / с ошибкой).
Размер списка ограничивают явным лимитом; превышение — 400 BATCH_SIZE_EXCEEDED.
Атомарность (все или никто) — исключение, требует явного указания в описании API.
Долгая операция запускается через POST, сервер отвечает 202 Accepted с заголовком Location.
Клиент периодически делает GET по этому адресу и смотрит на status: PENDING → PROCESSING → COMPLETED / FAILED.
При COMPLETED в ответе есть resultUrl; при FAILED — error с кодом и описанием.
Язык ответа передаётся через Accept-Language; переводится только detail, машиночитаемые поля (code, title, type) остаются на английском.

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

Ошибки RFC 9457 в FastAPI — code, detail, violations, application/problem+json.
Заголовки и трассировка в FastAPI — Idempotency-Key, Location, traceparent.
JSON и формат ответов в FastAPI — content + метаданные пагинации, exclude_none.