Опирается на правила: R-BATCH-1..5, R-ASYNC-1..4, R-LOC-1..3 и X-коды из REST API Style Guide → раздел Batch, async, локализация.

Важно знать

  • Batch — partial success: ошибка одного элемента не останавливает остальные; атомарность указывается явно.
  • Endpoint: POST /resources/batch или POST /resources/batch/<action>.
  • Запрос: Pydantic-модель с полем items: list[...]. Ответ: 200 OK с results и summary.
  • Размер batch ограничен; превышение — 400 BATCH_SIZE_EXCEEDED.
  • Async — polling: 202 Accepted + Location + тело с taskId, status, statusUrl.
  • Статусы задачи: PENDING / PROCESSING / COMPLETED / FAILED.
  • COMPLETED требует resultUrl; FAILED требует error.
  • Локализация: Accept-Language влияет только на detail и violations[].message; code, title, type, имена полей не переводятся.

Три паттерна для случаев, выходящих за рамки стандартного CRUD: массовые операции, длительные задачи и многоязычные ответы. FastAPI обрабатывает каждый из них через Pydantic-модели и декораторы маршрутов — код-first подход, при котором Pydantic-модель и есть источник контракта.

Batch-операции

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)

Для batch-команды сегмент содержит действие:

@router.post(
    "/notifications/batch/send",
    status_code=200,
    operation_id="batchSendNotifications",
    tags=["Notifications"],
    summary="Массовая отправка уведомлений (partial success)",
)

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 при успехе — правило R-RSP-X1.

Пример запроса и ответа

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 — даже при partial failure. Это не ошибка запроса, а ожидаемый результат batch-обработки. index отсчитывается от 0 и указывает на позицию в исходном items.

Превышение размера batch

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": "Размер batch превышает максимум (100 элементов), передано: 150",
  "code": "BATCH_SIZE_EXCEEDED"
}

Атомарность

Partial success — поведение по умолчанию. Если нужна атомарность, это указывается явно в OpenAPI description и в документации:

«Все элементы обрабатываются в одной транзакции. Ошибка любого элемента вызывает откат всех. При partial failure — 400 BATCH_TRANSACTION_FAILED с перечнем failed-индексов.»

Async-операции

Submit — 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)",
)
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 в заголовке — обязателен (R-ASYNC-1). statusUrl в теле — дублирует Location для клиентов, которые не читают заголовки.

Polling — 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-задачи",
)
async def get_task(
    task_id: str,
    service: TaskService = Depends(get_task_service),
) -> TaskStatusResponse:
    return await service.get_status(task_id)

exclude_none=True на TaskStatusResponse исключает resultUrl пока задача не завершена и error если задача успешна.

Ответы на разных этапах:

// PROCESSING
{ "taskId": "550e8400-...", "status": "PROCESSING", "progress": 45, "createdAt": "2026-06-19T10:30:00Z" }

// COMPLETED
{
  "taskId": "550e8400-...",
  "status": "COMPLETED",
  "progress": 100,
  "createdAt": "2026-06-19T10:30:00Z",
  "completedAt": "2026-06-19T10:35:00Z",
  "resultUrl": "/api/v1/reports/550e8400-..."
}

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

Рекомендуемый интервал polling

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

response.headers["Retry-After"] = "5"  # рекомендованный интервал в секундах

Локализация

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

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 (R-LOC-2).

Что переводится

Только сообщения, предназначенные пользователю: detail в ProblemDetails и message в violations.

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" }

Что не переводится

R-LOC-X1: code, title, type, имена JSON-полей.

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

// Неправильно
{ "code": "ТОВАР_НЕТ", "title": "Неверный запрос" }

code — машиночитаемый идентификатор; клиентский код делает switch по нему. Если он локализован — клиент сломается при смене языка запроса.

Локализация в exception handler

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",
    )

Что запрещено

АнтипаттернПравилоЧто взамен
Batch с атомарностью по умолчаниюR-BATCH-3partial success; атомарность — явная декларация
4xx на partial failure batchR-BATCH-3200 OK + per-item status
Batch-ответ без summaryR-BATCH-3total, succeeded, failed обязательны
Отсутствие ограничения размера batchR-BATCH-4явный лимит в OpenAPI и коде
202 без заголовка LocationR-ASYNC-1response.headers["Location"] = status_url
COMPLETED без resultUrlR-ASYNC-4обязателен при финальном статусе
FAILED без errorR-ASYNC-4обязателен при финальном статусе
code: "ЗАКАЗ_ПУСТОЙ" (кириллица в enum)R-LOC-X1только английский
Переведённый title ("Не найдено")R-LOC-X1стандартное название HTTP-статуса
Игнорирование Accept-LanguageR-LOC-1читать заголовок; дефолт ru
422 на ошибку валидации (дефолт FastAPI)R-ERR-X3переопределить handler на 400

Куда дальше

  • python/errors.md — code, detail, violations, application/problem+json; переопределение 422→400 в FastAPI.
  • python/headers.md — Idempotency-Key для batch, Location для async, traceparent.
  • python/json-and-responses.md — content + метаданные пагинации, exclude_none.
  • python/rate-limiting-files-deprecation.md — 429 Retry-After, StreamingResponse для файлов, Sunset.
  • python/url-and-resources.md — kebab-case, redirect_slashes=False, /api/v1-префикс.
  • Resilience → async polling — реализация task-queue на backend.