← Back to the section

Three situations that go beyond ordinary CRUD: you need to process several objects at once, launch a long-running task and let the client track it, or return error messages in the user's language. Let's take each in turn.

Batch Operations

The problem

Suppose you need to create 50 orders at once. Sending 50 separate requests is expensive. You can send a single request with a list. But what do you do if 3 of the 50 items fail validation? Abort everything, or continue with the rest?

In most cases the right answer is to continue. This behavior is called partial success: each item is processed independently, and one item's error doesn't stop the others. The server returns 200 OK with a result for every item.

Endpoint and models

A batch operation is built as a separate route. The request contains a list of items; the response contains a list of results and summary statistics.

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

@router.post(
    "/orders/batch",
    status_code=200,
    operation_id="batchCreateOrders",
    tags=["Orders"],
    summary="Batch order creation (partial success)",
)
async def batch_create_orders(
    request: BatchCreateOrdersRequest,
    service: OrderService = Depends(get_order_service),
) -> BatchCreateOrdersResponse:
    return await service.batch_create(request.items)

Pydantic models for the request and response:

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 on BatchCreateOrderResult removes the orderId field from the response on error and the error field on success — the JSON stays clean.

What the request and response look like

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 is returned even on a partial failure — this isn't a request error, it's the expected result. The index field corresponds to the position in the original items list (starting from 0).

Size limit

Accepting a list of any length is dangerous — it's unbounded load. You need to explicitly cap the maximum size and return 400 when it's exceeded:

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

Atomicity

By default it's partial success. If a specific operation must be atomic (one item's error cancels everything), that has to be explicitly stated in the API description. In that case a failure returns 400 BATCH_TRANSACTION_FAILED with a list of the problematic indices.

Long-Running Tasks: Launching and Tracking

The problem

Some operations take seconds or minutes: generating a report, exporting data, transcoding a file. Keeping an HTTP connection open for all that time is a bad idea. The client will get a timeout.

The solution is asynchronous processing with tracking: the client launches a task, receives an identifier, and periodically asks about its state.

Launching a task — 202 Accepted

from fastapi import Response
from fastapi.responses import JSONResponse


@router.post(
    "/reports/generate",
    status_code=202,
    operation_id="generateReport",
    tags=["Reports"],
    summary="Launch report generation",
)
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

The server's response:

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

The Location header points to the address for tracking. The statusUrl field duplicates it in the body — for clients that don't read response headers.

The server can also add a Retry-After header with a recommended polling interval in seconds:

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

Tracking state — 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="Task status",
)
async def get_task(
    task_id: str,
    service: TaskService = Depends(get_task_service),
) -> TaskStatusResponse:
    return await service.get_status(task_id)

Responses at the various stages look like this:

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

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

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

On COMPLETED, resultUrl — a link to the result — is always present. On FAILED, error with a problem description is always present. exclude_none=True removes fields that aren't filled in yet.

Response Localization

The problem

A user from Germany sees the error message "Order not found". You'd like to return the text in the user's language. But how do you pass the language, and what exactly do you translate?

The Accept-Language header

The client specifies the language in the standard HTTP header Accept-Language. FastAPI reads it directly:

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)

If the header isn't passed, the default language ru is used.

What to translate and what not to

Only user-facing messages are translated: the detail field in an error response and message in the list of validation 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" }

The code, title, type fields and the JSON field names are not translated. code is a machine-readable identifier that client code uses in a switch. If it ends up in Cyrillic or changes when the language changes, the client will break.

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

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

Localization in the 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",
    )

In short

  • Batch operations return 200 OK even on a partial failure — each item is processed independently.
  • The response contains results (a result for each item) and summary (totals: total / succeeded / failed).
  • The list size is capped by an explicit limit; exceeding it gives 400 BATCH_SIZE_EXCEEDED.
  • Atomicity (all or nothing) is the exception; it requires an explicit statement in the API description.
  • A long operation is launched via POST, and the server responds with 202 Accepted and a Location header.
  • The client periodically does a GET on that address and watches status: PENDING → PROCESSING → COMPLETED / FAILED.
  • On COMPLETED the response has resultUrl; on FAILED — error with a code and description.
  • The response language is passed via Accept-Language; only detail is translated, while the machine-readable fields (code, title, type) stay in English.
  • RFC 9457 errors in FastAPI — code, detail, violations, application/problem+json.
  • Headers and tracing in FastAPI — Idempotency-Key, Location, traceparent.
  • JSON and response format in FastAPI — content + pagination metadata, exclude_none.