SQLAlchemy перезаписала чужую правку: включаем version_id_col

Две Session загрузили одну запись. Первая сохранила новое название. Вторая по-прежнему держит старый объект и спустя время делает commit своего изменения. По умолчанию ORM знает, какую строку обновить по первичному ключу, но ей не обязательно известно, что объект уже устарел. Для обнаружения такого конфликта можно добавить счётчик версии строки.

Проверка привязана к flush ORM

У модели появляется обязательная колонка version. Настройка __mapper_args__ с version_id_col сообщает ORM, что UPDATE должен сопоставлять не только id, но и прежнюю версию. Успешное изменение увеличивает счётчик. Когда подходящая строка не найдена, SQLAlchemy выбрасывает StaleDataError вместо тихого сохранения поверх нового состояния.

Механизм полезен для долгоживущего объекта, который был прочитан раньше другой транзакции, а также для некоторых сценариев повторного использования состояния. Он работает в процессе flush отдельных ORM-объектов. Массовый UPDATE через Session.execute или старые bulk-методы не получает такую проверку автоматически. Для них условие версии нужно проектировать отдельно.

Колонка версии должна быть NOT NULL. Пустое значение разрушает смысл сравнения прежнего состояния. Если в проекте уже есть записи, миграция сначала заполняет версию, затем устанавливает ограничение и подключает mapper. Проверяйте все пути изменения таблицы: фоновая задача с обычным SQL способна обойти правило ORM.

Загрузим объект в двух Session

Пример использует SQLAlchemy 2 и временную SQLite в памяти. Он открывает две Session, читает запись в обе, завершает их читающие транзакции с expire_on_commit=False и затем сохраняет два разных изменения. Такое построение делает конфликт детерминированным и оставляет в памяти обе исходные версии, не пытаясь моделировать блокировки SQLite.

После отказа выполняется rollback. Затем отдельная Session читает строку и подтверждает, что первая правка сохранилась, а вторая не прошла. Этот последний шаг важен: пойманное исключение само по себе ещё не описывает итоговое содержимое базы.

from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from sqlalchemy.orm.exc import StaleDataError

class Base(DeclarativeBase):
    pass
class Document(Base):
    __tablename__ = "document"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String)
    version: Mapped[int] = mapped_column(nullable=False)
    __mapper_args__ = {"version_id_col": version}

engine = create_engine("sqlite://")
Base.metadata.create_all(engine)
with Session(engine) as db:
    db.add(Document(id=1, name="Draft"))
    db.commit()

with Session(engine, expire_on_commit=False) as first, Session(engine, expire_on_commit=False) as second:
    a = first.get(Document, 1)
    b = second.get(Document, 1)
    first.commit()
    second.commit()
    assert a.version == b.version == 1
    a.name = "Approved"
    first.commit()
    b.name = "Old revision"
    try:
        second.commit()
    except StaleDataError:
        second.rollback()
    else:
        raise AssertionError("stale object must fail")

with Session(engine) as db:
    saved = db.get(Document, 1)
    assert (saved.name, saved.version) == ("Approved", 2)
    print({"name": saved.name, "version": saved.version})
engine.dispose()

Как связать конфликт с поведением API

В FastAPI поймайте ожидаемый StaleDataError на уровне операции, откатите транзакцию и верните договорённый ответ о конфликте, например 409. Клиенту нужно перечитать состояние. Не повторяйте сохранение старого объекта автоматически: это превращает обнаруженный конфликт в намеренную перезапись. Возможность объединить изменения зависит от конкретных полей и продуктового сценария.

Если клиент открыл форму давно, а сервер при PUT заново загрузил свежий объект, одна version_id_col не обнаружит устаревание формы. ORM сравнит версию с момента своего свежего SELECT, а не с момента чтения клиентом. Для защиты этого промежутка передайте ожидаемую версию через If-Match или поле запроса и включите её в атомарное условие записи. Это отдельная граница контроля.

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

expire_on_commit=False в тесте нужен для сохранения старого состояния. Не включайте его во всём приложении только ради этого примера. Выбор политики истечения зависит от того, как долго живут ваши Session и когда данные должны перечитываться. На запрос обычно выдаётся отдельная Session, а её объекты не передаются другим параллельным задачам.

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

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

Сначала закрепите явные границы Session и commit. В бесплатном курсе FastAPI вы соберёте API с SQLAlchemy 2 и HTTP-тестами; затем дополните его сценарием устаревшего ORM-объекта из этого разбора. Перейдите к бесплатному курсу FastAPI.