Повтор POST в FastAPI: как вернуть тот же заказ по Idempotency-Key

Клиент отправил POST /orders и через пять секунд получил тайм-аут. Сервер за это время мог успеть записать заказ. С точки зрения клиента результат неизвестен, поэтому приложение пытается отправить запрос снова. Если каждый POST создаёт отдельную запись, покупатель видит два заказа и уже не может понять, какой из них отменять.

Клиенту нужен способ обозначить намерение, которое переживёт сетевой сбой. Перед первой отправкой он создаёт случайный ключ и прикладывает его в заголовке Idempotency-Key. Повтор той же операции отправляет прежний ключ. Другая покупка получает новый ключ, даже если клиент заказал ровно тот же товар.

Зафиксируйте заказ и ключ вместе

Серверу надо ответить на два вопроса. Был ли этот ключ уже использован данным покупателем? Совпадают ли параметры с первоначальным запросом? Первый вопрос позволяет вернуть прежний заказ. Второй выявляет ошибку интеграции, когда приложение случайно сохраняет ключ после смены товара или количества.

Ключ, параметры и идентификатор заказа удобно связать ограничением уникальности по паре «владелец и ключ». В примере все данные, нужные для повторного ответа, лежат в строке заказа. Выборка и вставка идут в одной транзакции. Отдельный кэш для ключей создал бы промежуток, в котором заказ записан, а ответ о прежней попытке ещё потерян при сбое.

Проверьте две одновременные отправки

В SQLite BEGIN IMMEDIATE начинает транзакцию записи до проверки ключа. Пока первая транзакция открыта, вторая ждёт освобождения записи в пределах тайм-аута соединения. Когда первая зафиксируется, вторая прочитает готовый заказ. Ограничение UNIQUE остаётся обязательным свойством таблицы и действует также для других участков кода, которые добавляют заказы.

Пример запускает два вызова в разных потоках через FastAPI TestClient. Локальная база создаётся в отдельном временном каталоге. buyer задан константой, чтобы проверка была самодостаточной. В рабочем приложении его возвращает зависимость, которая установила личность покупателя.

Сохраните пример в файл demo.py, установите зависимости командой python -m pip install fastapi httpx и запустите python demo.py.

from concurrent.futures import ThreadPoolExecutor
from contextlib import closing
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Annotated
import sqlite3

from fastapi import FastAPI, Header, HTTPException
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field


class OrderInput(BaseModel):
    sku: str
    quantity: int = Field(gt=0)


def build_app(db_path: Path) -> FastAPI:
    with closing(sqlite3.connect(db_path)) as db, db:
        db.execute('''
            CREATE TABLE orders (
                id INTEGER PRIMARY KEY,
                buyer TEXT NOT NULL,
                request_key TEXT NOT NULL,
                sku TEXT NOT NULL,
                quantity INTEGER NOT NULL,
                UNIQUE (buyer, request_key)
            )
        ''')

    app = FastAPI()

    @app.post("/orders", status_code=201)
    def create_order(
        order: OrderInput,
        key: Annotated[str, Header(alias="Idempotency-Key", min_length=1, max_length=128)],
    ):
        buyer = "buyer-42"
        with closing(sqlite3.connect(db_path, timeout=5)) as db, db:
            db.execute("BEGIN IMMEDIATE")
            existing = db.execute(
                "SELECT id, sku, quantity FROM orders WHERE buyer = ? AND request_key = ?",
                (buyer, key),
            ).fetchone()
            if existing:
                order_id, sku, quantity = existing
                if (sku, quantity) != (order.sku, order.quantity):
                    raise HTTPException(409, "Key already used with different input")
            else:
                inserted = db.execute(
                    "INSERT INTO orders (buyer, request_key, sku, quantity) VALUES (?, ?, ?, ?)",
                    (buyer, key, order.sku, order.quantity),
                )
                order_id = inserted.lastrowid
        return {"id": order_id, "sku": order.sku, "quantity": order.quantity}

    return app


with TemporaryDirectory() as directory:
    db_path = Path(directory) / "orders.db"
    app = build_app(db_path)
    headers = {"Idempotency-Key": "checkout-a9c"}

    def post(quantity: int):
        with TestClient(app) as client:
            return client.post("/orders", headers=headers, json={"sku": "PEN", "quantity": quantity})

    with ThreadPoolExecutor(max_workers=2) as workers:
        first, second = workers.map(post, [2, 2])

    assert (first.status_code, second.status_code) == (201, 201)
    assert first.json() == second.json() == {"id": 1, "sku": "PEN", "quantity": 2}
    assert post(3).status_code == 409
    with closing(sqlite3.connect(db_path)) as db:
        assert db.execute("SELECT COUNT(*) FROM orders").fetchone()[0] == 1
    print(first.json())

Оба клиента получают один и тот же ответ с id=1. Третий запрос оставляет тот же ключ, но меняет количество. Он получает 409, и строка в базе остаётся единственной. Запись заказа фиксируется при выходе из контекстного менеджера соединения. Если сервер завершится после коммита и до отправки ответа, следующая попытка уже увидит записанные данные.

Пример возвращает 201 при повторе по выбранному контракту. Опишите этот ответ в документации клиента. При изменяемых заказах отделите первоначальный ответ от текущего состояния. Например, храните код и тело первого ответа вместе с ключом. Иначе повтор создания через неделю может получить уже отредактированную версию заказа, и клиент сочтёт её исходной.

Где заканчивается транзакция

Приведённая схема защищает изменение внутри SQLite. Если из обработчика заодно вызвать платёжный шлюз, внешняя операция переживёт откат локальной транзакции. Для такой интеграции сохраните намерение отправки в базе и передайте стабильный ключ внешнему API, который поддерживает повторы. Для клиента отдельно опишите период хранения ключа. Пока у запроса остаётся право повториться, серверу нужна запись о прежней попытке.

Перед выпуском добавьте проверку одинакового ключа у двух покупателей и имитацию потери ответа после коммита. В тесте сбоя после коммита отбросьте ответ первого запроса, повторите его с прежним ключом и сверьте идентификатор с записью в базе. Так вы проверите именно тот сценарий, ради которого клиенту нужен заголовок.

Чтобы собрать API целиком, пройдите бесплатный курс Latorn по FastAPI. В одном проекте вы настроите проверку входных данных, подключите SQLAlchemy, добавите авторизацию и научитесь проверять маршруты через pytest и TestClient. Открыть программу курса FastAPI или начать с открытого урока про TestClient.