SELECT вызвал INSERT и упал: разбираем autoflush SQLAlchemy

Обработчик создаёт объект заказа по частям. Сначала добавляет его в Session, затем читает справочник, чтобы заполнить обязательное поле. Но ошибка NOT NULL возникает на строке SELECT, где записи вроде бы нет. ORM перед чтением автоматически отправляет накопленные изменения, и незавершённый объект попадает под ограничения таблицы раньше ожидаемого.

Почему чтение связано с записью

Autoflush поддерживает согласованность запросов с локальными изменениями Session. Иначе поиск мог бы не видеть только что добавленные или изменённые объекты. Но это означает, что ORM SELECT способен сначала вызвать flush. Ошибка относится к накопленному состоянию, а строка чтения всего лишь запускает его отправку в базу.

Первый выбор — собрать корректный объект до add. Если для сборки действительно нужно чтение в той же Session, небольшой участок no_autoflush позволяет отложить автоматическую отправку. После заполнения обязательных полей вызывается явный flush. Это помогает расположить проверку ограничений в понятном месте, а не отключить её вообще.

Сбой до заполнения и исправленный порядок

Нужен SQLAlchemy. Поле name имеет NOT NULL по аннотации Mapped[str]. Первый сценарий демонстрирует реальный IntegrityError. После rollback запускается второй сценарий: поиск выполняется без преждевременного flush, поле заполняется и запись сохраняется. В конце проверяется результат новой Session.

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

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://")
Base.metadata.create_all(engine)
from sqlalchemy import func
from sqlalchemy.exc import IntegrityError

with Session(engine) as db:
    db.add(Item())
    try:
        db.scalar(select(func.count()).select_from(Item))
    except IntegrityError:
        db.rollback()
        print({"select_triggered_flush": "IntegrityError"})
    else:
        raise AssertionError("unfinished object unexpectedly accepted")

with Session(engine) as db:
    pending = Item()
    db.add(pending)
    with db.no_autoflush:
        existing = db.scalar(select(func.count()).select_from(Item))
        assert existing == 0
        pending.name = "assembled"
    db.flush()
    db.commit()

with Session(engine) as db:
    names = db.scalars(select(Item.name)).all()
    assert names == ["assembled"]
    print({"stored": names})
engine.dispose()

Почему no_autoflush нельзя расширять бесконечно

Пока flush отложен, SQL-запросы не видят неотправленные изменения как строки базы. Поэтому проверка «такое имя уже существует» может пропустить объект, добавленный в текущую Session. Финальную уникальность всё равно обеспечивает ограничение таблицы. Ограничивайте контекст тем местом, где собирается объект, и явно обозначайте последующую точку проверки.

Commit выполняет flush независимо от настройки автоматической отправки перед чтением. begin_nested также имеет собственную обязательную отправку состояния. Поэтому общий autoflush=False не означает «ничего не записывать, пока я не вызову flush». Он меняет конкретные автоматические точки, а завершение транзакции всё равно должно проверить накопленные изменения.

После неудачного flush Session требует rollback перед продолжением работы. Простое except с повторным SELECT оставит её в ошибочном состоянии. В первом сценарии откат сделан явно, после чего Session закрывается. При управлении транзакцией через with db.begin исключение должно выйти из контекста, который выполнит нужную очистку.

Для диагностики включите журнал SQL в локальном окружении и посмотрите команды прямо перед исключением. Часто там обнаружится INSERT, который не виден на проблемной строке Python. Затем найдите место раннего add и решите, можно ли перенести его после сборки. no_autoflush стоит оставлять там, где зависимость сборки от чтения действительно нужна.

Если создание объекта разбросано по нескольким функциям, полезнее вернуть из подготовительной функции готовые значения и только потом добавить сущность. Это делает промежуточное недопустимое состояние локальным. Границы транзакции и Session разбираются в статье об одной Session на запрос; autoflush объясняет неожиданный SQL внутри этих границ.

Продолжить на практике

Разделите сборку команды и сохранение, а ошибку ограничения оставьте в регрессионном тесте. В продвинутом Python есть тестируемый дизайн и Repository: эти темы помогут сократить число функций, которые меняют Session в неожиданный момент. Откройте программу продвинутого Python.