Транзакция в FastAPI: как сохранить две записи целиком или откатить обе

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

Транзакция выражает бизнес-границу

Не вызывайте commit после каждого add, если несколько изменений составляют одно действие пользователя. Контекст session.begin() объединяет их: нормальный выход фиксирует изменения, исключение откатывает. Ошибку можно преобразовать в ответ HTTP после выхода из блока. Если поймать её внутри и продолжить как ни в чём не бывало, состояние транзакции станет труднее понимать.

В примере используются две записи справочника с уникальными именами. Это небольшой способ увидеть атомарность без большого домена заказа. Одинаковые имена заставят базу отвергнуть вторую запись; тест проверит, что первая тоже исчезла. Затем успешный запрос докажет, что приложение продолжает работать.

Проверяем атомарность через endpoint

Используются FastAPI, HTTPX и SQLAlchemy 2. Ограничение unique действительно находится в таблице SQLite, а не имитируется условием Python. flush отправляет изменения до выхода из транзакции, чтобы нарушение ограничения стало видно в определённом месте.

Сохраните весь блок в файл и запустите 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
class Pair(BaseModel):
    first: str
    second: str

@app.post("/pairs", status_code=201)
def create_pair(data: Pair, db: Db):
    try:
        with db.begin():
            db.add_all([Item(name=data.first), Item(name=data.second)])
            db.flush()
    except IntegrityError:
        raise HTTPException(409, "names must be unique")
    return {"created": 2}

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

with TestClient(app) as client:
    assert client.post("/pairs", json={"first": "A", "second": "A"}).status_code == 409
    assert client.get("/items").json() == []
    assert client.post("/pairs", json={"first": "A", "second": "B"}).status_code == 201
    assert client.get("/items").json() == ["A", "B"]
print("failed pair rolled back; successful pair committed")
engine.dispose()

Что происходит после ошибки

Контекст begin выполняет rollback при исключении. Отдельный ручной rollback в этой конструкции не нужен. При ручном управлении через commit и try/except он, напротив, необходим перед повторным использованием той же Session после неудачного flush. Выберите одну ясную схему и придерживайтесь её в одном участке кода.

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

Как тестировать настоящий бизнес-сценарий

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

Транзакция базы не откатывает письмо или запрос во внешнюю платёжную систему. Если действие выходит за пределы одной базы, нужен отдельный порядок координации: например, запись задания на отправку в той же транзакции и его выполнение после commit. Начните с честной границы атомарности, чтобы не считать удалённый побочный эффект частью локального rollback.

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

В открытом уроке чтения и создания через SQLAlchemy 2 вы отработаете путь от HTTP-команды до сохранённой записи. Затем объедините два изменения и добавьте отказ второй вставки, чтобы самостоятельно проверить атомарность. Перейдите к бесплатной практике SQLAlchemy 2.

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