Три ситуации, которые выходят за рамки обычного 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)",
response_model_exclude_none=True,
)
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)
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
Чтобы в ответе не было orderId при ошибке и error при успехе, на маршрут вешают response_model_exclude_none=True: настройки модели для этого в Pydantic нет, exclude_none — параметр сериализации.
Как выглядит запрос и ответ
живой пример
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 с перечнем проблемных индексов.
Долгие задачи: запуск и отслеживание
Проблема
Клиент получает 202 и адрес задачи, затем опрашивает статус: из обработки задача уходит либо в результат, либо в ошибку с кодом.
Некоторые операции занимают секунды или минуты: генерация отчёта, экспорт данных, перекодирование файла. Держать 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)
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="Статус задачи",
response_model_exclude_none=True,
)
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 с описанием проблемы. response_model_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. В бою заголовок приходит списком с весами, например de-DE,de;q=0.9,en;q=0.8: его разбирают и сопоставляют со списком поддерживаемых языков (babel.core.negotiate_locale делает это за одну строку), а не берут строку целиком как код языка.
Что переводить, а что нет
Переводятся только сообщения для пользователя: поле 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",
)
Глубже: вебхуки: события потребителю вместо опросарасширенное
Асинхронная операция выше заканчивается тем, что клиент опрашивает статус. Когда клиентов сорок тысяч, а событий у каждого несколько в день, опрос превращается в основную нагрузку на API, и правильный ответ это вебхук: потребитель регистрирует адрес, а вы сами присылаете ему событие POST-запросом.
Контракт вебхука состоит из четырёх частей. Тело: событие с id, type, occurredAt и данными; по id потребитель отбрасывает повтор. Подпись: заголовок Webhook-Signature с HMAC-SHA256 от тела на общем секрете плюс отметка времени, чтобы перехваченный запрос нельзя было прислать через час; потребитель сверяет через hmac.compare_digest, а не ==. Доставка: таймаут пять секунд, успехом считается любой 2xx, остальное повтор с нарастающей паузой из той же очереди задач, где живёт outbox. Отключение: после серии отказов за сутки адрес помечают неработающим и пишут владельцу, иначе очередь копит мусор.
На стороне отправителя это обычный исходящий HTTP из фоновой задачи, а не из обработчика запроса: событие сначала попадает в outbox той же транзакцией, что и изменение, иначе при падении сервиса потребитель так и не узнает о событии. Со стороны потребителя эндпоинт вебхука отвечает быстро и кладёт работу в очередь, а не выполняет её синхронно.
Опрос при этом не исчезает: он остаётся запасным путём для тех, кто вебхук не настроил, и способом сверки «не пропустили ли событие».
Коротко
- Массовые операции возвращают
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) остаются на английском. - При тысячах клиентов опрос заменяют вебхуком: подписанное HMAC событие с идентификатором для идемпотентности, повторы с паузой из очереди, отключение адреса после серии отказов.
Что почитать дальше
- Ошибки RFC 9457 в FastAPI —
code,detail,violations,application/problem+json. - Заголовки и трассировка в FastAPI —
Idempotency-Key,Location,traceparent. - JSON и формат ответов в FastAPI —
content+ метаданные пагинации,exclude_none.