FastAPI возвращает 500 при дубликате: unique-ограничение и ответ 409

Два запроса одновременно создают один и тот же короткий код. Оба предварительно проверили отсутствие записи. Первый сохранился, второй получил IntegrityError и превратился в 500. Проверка SELECT полезна для интерфейса, но окончательное решение об уникальности должна принимать база.

Почему предварительного SELECT недостаточно

Между проверкой и вставкой может выполниться другой запрос. Уникальный индекс остаётся правильной защитой при этой гонке. Приложение должно распознать ожидаемое нарушение известного ограничения и вернуть клиенту конфликт 409. Клиент сможет выбрать другой код, а не повторять неисправный запрос как временный сбой сервера.

После неудачного flush или commit Session требует rollback перед продолжением работы. В примере commit находится в try, а rollback — в ветке известного IntegrityError. Мы не передаём клиенту текст SQL или исключения драйвера: в нём могут оказаться структура базы и значения параметров.

Проверяем дубликат и следующий успешный запрос

Для короткого примера есть одна уникальная колонка name, поэтому причина конфликта однозначна. Нужны FastAPI, HTTPX, SQLAlchemy 2. Три последовательных POST покажут, что конфликт не повредил приложение и не создал лишнюю запись.

Сохраните весь блок в файл и запустите Python 3.11+ в отдельном окружении с пакетами fastapi, httpx, sqlalchemy. Для моделей используется Pydantic v2; отдельный сервер для TestClient не нужен.

from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, String, select, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from sqlalchemy.pool import StaticPool

class Base(DeclarativeBase):
    pass
class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String, unique=True)

engine = create_engine("sqlite://", poolclass=StaticPool,
                       connect_args={"check_same_thread": False})
Base.metadata.create_all(engine)
app = FastAPI()
def get_db():
    with Session(engine) as session:
        yield session
Db = Annotated[Session, Depends(get_db)]

from sqlalchemy.exc import IntegrityError
from pydantic import BaseModel, Field
class ItemIn(BaseModel):
    name: str = Field(min_length=1, max_length=50)

@app.post("/items", status_code=201)
def create(data: ItemIn, db: Db):
    item = Item(name=data.name)
    db.add(item)
    try:
        db.commit()
    except IntegrityError:
        db.rollback()
        raise HTTPException(409, {"code": "name_taken"})
    db.refresh(item)
    return {"id": item.id, "name": item.name}

@app.get("/items")
def read(db: Db):
    return db.scalars(select(Item.name).order_by(Item.name)).all()

with TestClient(app) as client:
    assert client.post("/items", json={"name": "room-a"}).status_code == 201
    duplicate = client.post("/items", json={"name": "room-a"})
    assert duplicate.status_code == 409
    assert duplicate.json()["detail"] == {"code": "name_taken"}
    assert client.post("/items", json={"name": "room-b"}).status_code == 201
    assert client.get("/items").json() == ["room-a", "room-b"]
print("duplicate rejected; next insert succeeds")
engine.dispose()

Согласуйте смысл одинакового имени

room-a и Room-A могут считаться одним именем в продукте, но разными строками в базе. Определите нормализацию до вставки и поддержите её ограничением, соответствующим выбранной базе. Одного lower в Python недостаточно, если другие способы записи обходят приложение. Для международных имён отдельно проверьте Unicode и правила сравнения.

Ответ 409 должен иметь стабильный машинный код, если интерфейс показывает ошибку возле конкретного поля. Не заставляйте клиент разбирать фразу «UNIQUE constraint failed». Для человека можно добавить понятное сообщение, а код name_taken оставить неизменным между переводами.

Что доказывает локальный тест

Два последовательных одинаковых запроса проверяют действующее ограничение и обработку исключения. Они не воспроизводят реальное конкурентное расписание. От гонки защищает unique в базе; конкурентный интеграционный тест дополнительно полезен для вашего драйвера и уровня изоляции, когда это становится существенным сценарием продукта.

Если в одной операции есть несколько ограничений, определяйте, какое именно нарушено, через поддерживаемые сведения драйвера и выбранные имена ограничений. Не скрывайте поломку внешней связи под сообщением «имя занято». Преобразование в 409 — часть договора конкретной операции, а не универсальная маска для любой ошибки хранения.

Следующая практика

Закрепите вставку и чтение записи в открытом уроке SQLAlchemy 2. Следующим собственным экспериментом сделайте повторное создание одинакового имени и проверьте, что известный конфликт не ломает последующий корректный запрос. Откройте урок создания записей через SQLAlchemy.

Чтобы собрать маршруты, хранение и тесты в одном приложении, откройте бесплатную программу FastAPI. Когда понадобится отдельная PostgreSQL и воспроизводимый запуск окружения, продолжите в Docker и Docker Compose: вы соедините сервис с базой, настроите сеть, том и проверку готовности.