Дубликат сорвал импорт: сохраняем хорошие строки через SAVEPOINT

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

Сначала определите политику пакета

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

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

Дубликат должен находиться между хорошими строками

Понадобится SQLAlchemy. Для SQLite пример явно управляет BEGIN через события движка: это сохраняет вложенные SAVEPOINT внутри настоящей внешней транзакции и при старом режиме управления транзакциями sqlite3. В таблице есть настоящее ограничение unique. Импортируем A, повтор A и B. Сначала отдельная проверка вызывает сбой внешней операции после успешного SAVEPOINT и подтверждает полный откат. Затем новая Session проверяет принятые A и B и индекс отклонённого дубликата.

from sqlalchemy import create_engine, select, String, event
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://")

@event.listens_for(engine, "connect")
def disable_driver_begin(connection, record):
    connection.isolation_level = None

@event.listens_for(engine, "begin")
def explicit_begin(connection):
    connection.exec_driver_sql("BEGIN")

Base.metadata.create_all(engine)
from sqlalchemy.exc import IntegrityError

# A successful SAVEPOINT must still roll back with the outer transaction.
with Session(engine) as db:
    try:
        with db.begin():
            with db.begin_nested():
                db.add(Item(name="must-rollback"))
                db.flush()
            raise RuntimeError("outer operation failed")
    except RuntimeError:
        pass
with Session(engine) as db:
    assert db.scalars(select(Item.name)).all() == []
print({"outer_failure": "all rows rolled back"})

accepted = []
rejected = []
with Session(engine) as db:
    with db.begin():
        for index, name in enumerate(["A", "A", "B"]):
            try:
                with db.begin_nested():
                    db.add(Item(name=name))
                    db.flush()
            except IntegrityError:
                rejected.append(index)
            else:
                accepted.append(index)

with Session(engine) as db:
    names = db.scalars(select(Item.name).order_by(Item.name)).all()
    assert names == ["A", "B"]
    assert accepted == [0, 2] and rejected == [1]
    print({"accepted_rows": accepted, "rejected_rows": rejected, "stored": names})
engine.dispose()

Почему важно место add и flush

begin_nested перед созданием SAVEPOINT выполняет flush уже накопленного состояния. Поэтому объект очередной строки добавляется внутри вложенного блока, а не до него. Если подготовить сразу весь пакет через add_all, ожидаемая ошибка способна возникнуть раньше нужной точки и сорвать внешний участок. Явный flush делает момент проверки ограничения понятным.

Не вызывайте db.rollback в обработчике строки: в современном Session это откатит внешнюю транзакцию, а не только текущую точку. В примере локальным откатом управляет контекст begin_nested. Аналогично commit внешней Session фиксирует внешнюю транзакцию. Ясные вложенные контексты помогают не перепутать эти уровни при последующем рефакторинге.

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

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

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

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

Попрактикуйтесь разделять сервис обработки и хранение, а отчёт проверять через pytest. В продвинутом Python есть Repository, собственные исключения и тесты со stub: они помогут оформить импорт без смешения транспортных и предметных причин отказа. Откройте программу продвинутого Python.