APPLYDJINN
← SQLAlchemy and Alembic: reliable database workProgress & review ↗

Typorithm · medium

Savepoints and recovery from a failed flush

pythonsqlalchemyalembicdatabasesindependent-practice
Saved on this device
Full lesson · 4 learning steps Open any exercise for focused practice
1

Step 1 of 4 · intro card

Savepoints and recovery from a failed flush

0%

Concept and contract

Reject a duplicate insert while keeping valid work in the surrounding transaction.

Catching IntegrityError is not the same as restoring a usable transaction. A failed flush affects Session state, and a root rollback discards the root transaction’s pending work. Sometimes the domain wants all-or-nothing behavior; sometimes it deliberately wants to reject one record while retaining other valid records. Define that policy before choosing exception handling.

The owned fixture seeds one unique code, then accepts a sequence that may contain duplicates. Each candidate insert gets its own savepoint inside one outer transaction. The report must preserve unique accepted codes and count the rejected duplicate attempts.

Mechanism and worked example

The program flushes the seed before processing candidates.

The program flushes the seed before processing candidates. For each candidate it enters begin_nested, adds the Role and flushes inside that context. IntegrityError is caught outside the nested block so the savepoint context can roll back its failed insert before processing continues.

SQLAlchemy flushes pending work when begin_nested starts, even if ordinary autoflush is disabled. This is why the seed’s boundary is explicit. For candidates [seed, a, a, b], the final codes are [a, b, seed] with two duplicate attempts, not an empty table or a partially inactive Session.

Complete worked example

python

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

class Base(DeclarativeBase):
    pass

class Role(Base):
    __tablename__ = "roles"
    id: Mapped[int] = mapped_column(primary_key=True)
    code: Mapped[str] = mapped_column(String(50), unique=True)

def inspect_example(codes):
    engine = create_engine("sqlite+pysqlite://", connect_args={"autocommit": False})
    try:
        Base.metadata.create_all(engine)
        duplicates = 0
        with Session(engine) as session, session.begin():
            session.add(Role(code="seed"))
            session.flush()
            for code in codes:
                try:
                    with session.begin_nested():
                        session.add(Role(code=code))
                        session.flush()
                except IntegrityError:
                    duplicates += 1
        with Session(engine) as check:
            stored = list(check.scalars(select(Role.code).order_by(Role.code)))
        return {"codes": stored, "duplicates": duplicates}
    finally:
        engine.dispose()
Reasoning
The program flushes the seed before processing candidates. For each candidate it enters begin_nested, adds the Role and flushes inside that context. IntegrityError is caught outside the nested block so the savepoint context can roll back its failed insert before processing continues. SQLAlchemy flushes pending work when begin_nested starts, even if ordinary autoflush is disabled. This is why the seed’s boundary is explicit. For candidates [seed, a, a, b], the final codes are [a, b, seed] with two duplicate attempts, not an empty table or a partially inactive Session.
Download the SQLAlchemy and Alembic computer lab ↗

Focused decision: context is in the complete example

python

try:
    with session.begin_nested():
        session.add(Role(code=code))
        session.flush()
except IntegrityError:
    duplicates += 1
Database decision
The exception leaves the nested context before it is handled. The outer transaction can retain valid rows; a root rollback would deliberately discard that larger unit of work.
Download the SQLAlchemy and Alembic computer lab ↗

Counterexamples and common mistakes

The controlled mistake calls the root Session rollback when one nested insert fails and returns an empty report.

The controlled mistake calls the root Session rollback when one nested insert fails and returns an empty report. It discards the surrounding work rather than honoring the smaller failure boundary. Returning normally does not make that loss correct merely because the original duplicate error was caught.

Do not catch the exception inside the nested block and then continue using the failed context as if nothing happened. Do not assume a savepoint protects work flushed before it started. These details explain why a broad try/except wrapped around an entire import often has different semantics from one reviewed savepoint per candidate.

Controlled bug: observe the rejected database contract

python

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

class Base(DeclarativeBase):
    pass

class Role(Base):
    __tablename__ = "roles"
    id: Mapped[int] = mapped_column(primary_key=True)
    code: Mapped[str] = mapped_column(String(50), unique=True)

def inspect_example(codes):
    engine = create_engine("sqlite+pysqlite://", connect_args={"autocommit": False})
    try:
        Base.metadata.create_all(engine)
        duplicates = 0
        with Session(engine) as session, session.begin():
            session.add(Role(code="seed"))
            session.flush()
            for code in codes:
                try:
                    with session.begin_nested():
                        session.add(Role(code=code))
                        session.flush()
                except IntegrityError:
                    session.rollback()
                    return {"codes": [], "duplicates": duplicates + 1}
        with Session(engine) as check:
            stored = list(check.scalars(select(Role.code).order_by(Role.code)))
        return {"codes": stored, "duplicates": duplicates}
    finally:
        engine.dispose()
Reasoning
The controlled mistake calls the root Session rollback when one nested insert fails and returns an empty report. It discards the surrounding work rather than honoring the smaller failure boundary. Returning normally does not make that loss correct merely because the original duplicate error was caught. Do not catch the exception inside the nested block and then continue using the failed context as if nothing happened. Do not assume a savepoint protects work flushed before it started. These details explain why a broad try/except wrapped around an entire import often has different semantics from one reviewed savepoint per candidate.
Download the SQLAlchemy and Alembic computer lab ↗

Independent practice

First reconstruct the nested context and the placement of its exception handler.

First reconstruct the nested context and the placement of its exception handler. The independent task adds a small admission rule: skip empty codes before attempting an insert. Keep valid codes, duplicate counting and the root transaction policy unchanged. Empty text is currently allowed by the table’s unique column, so this is a new application rule.

Put the new guard before creating the savepoint. Do not count an intentionally skipped empty code as a duplicate failure. Use a case containing an empty code, an existing seed and a repeated valid code so that each kind of outcome remains distinguishable.

Step 1: Skip empty codes before creating a savepoint; keep valid rows and count only duplicate inserts. Step 2: Keep the existing duplicate handler and root transaction.

Practice starter: solve before reading the solution

python

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

class Base(DeclarativeBase):
    pass

class Role(Base):
    __tablename__ = "roles"
    id: Mapped[int] = mapped_column(primary_key=True)
    code: Mapped[str] = mapped_column(String(50), unique=True)

def inspect_example(codes):
    engine = create_engine("sqlite+pysqlite://", connect_args={"autocommit": False})
    try:
        Base.metadata.create_all(engine)
        duplicates = 0
        with Session(engine) as session, session.begin():
            session.add(Role(code="seed"))
            session.flush()
            for code in codes:
                try:
                    with session.begin_nested():
                        session.add(Role(code=code))
                        session.flush()
                except IntegrityError:
                    duplicates += 1
        with Session(engine) as check:
            stored = list(check.scalars(select(Role.code).order_by(Role.code)))
        return {"codes": stored, "duplicates": duplicates}
    finally:
        engine.dispose()

# Task: Skip empty codes before creating a savepoint; keep valid rows and count only duplicate inserts.
Reasoning
Skip empty codes before creating a savepoint; keep valid rows and count only duplicate inserts.
Download the SQLAlchemy and Alembic computer lab ↗
  • Hint 1: Add the empty-code guard at the start of the candidate loop.
  • Hint 2: Do not increment duplicates when the guard skips a value.

Solution, tests and trade-offs

The solution continues past empty codes before entering begin_nested.

The solution continues past empty codes before entering begin_nested. In the mixed fixture it retains [seed, x] and counts the repeated seed and repeated x, while excluding the empty string without calling it a duplicate. The old starter passes cases without empty text but fails the new mixed boundary case.

The acceptance suite checks the actual stored rows through a new Session after the outer transaction completes. This prevents an in-memory list of intended inserts from pretending that a rolled-back database retained them. Inputs remain unchanged, and the owned Engine is disposed at the end of each complete example.

Practice solution and acceptance checks

python

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

class Base(DeclarativeBase):
    pass

class Role(Base):
    __tablename__ = "roles"
    id: Mapped[int] = mapped_column(primary_key=True)
    code: Mapped[str] = mapped_column(String(50), unique=True)

def inspect_example(codes):
    engine = create_engine("sqlite+pysqlite://", connect_args={"autocommit": False})
    try:
        Base.metadata.create_all(engine)
        duplicates = 0
        with Session(engine) as session, session.begin():
            session.add(Role(code="seed"))
            session.flush()
            for code in codes:
                if not code:
                    continue
                try:
                    with session.begin_nested():
                        session.add(Role(code=code))
                        session.flush()
                except IntegrityError:
                    duplicates += 1
        with Session(engine) as check:
            stored = list(check.scalars(select(Role.code).order_by(Role.code)))
        return {"codes": stored, "duplicates": duplicates}
    finally:
        engine.dispose()
Reasoning
The solution continues past empty codes before entering begin_nested. In the mixed fixture it retains [seed, x] and counts the repeated seed and repeated x, while excluding the empty string without calling it a duplicate. The old starter passes cases without empty text but fails the new mixed boundary case. The acceptance suite checks the actual stored rows through a new Session after the outer transaction completes. This prevents an in-memory list of intended inserts from pretending that a rolled-back database retained them. Inputs remain unchanged, and the owned Engine is disposed at the end of each complete example.
Download the SQLAlchemy and Alembic computer lab ↗

Interview reasoning and next step

Draw the outer transaction and the individual savepoints, then mark where the seed is flushed and which work is rolled back after a duplicate.

Draw the outer transaction and the individual savepoints, then mark where the seed is flushed and which work is rolled back after a duplicate. Explain why an exception handler’s position changes whether a context manager can finish its recovery correctly. Distinguish rejected input from a database uniqueness violation.

Savepoints are a transaction tool, not a universal import policy. Another workflow may require aborting the entire batch, and some database failures invalidate more than one record’s work. Next, detect a stale entity update with a version counter and decide explicitly what a meaningful retry would do.

Optional stretch — after the core task works: Change an owned fixture to an all-or-nothing batch and compare its explicitly different rollback contract.

Computer practice: https://applydjinn.com/typorithm/sqlalchemy-lab-20261107.zip. Install the pinned requirements in an isolated Python 3.13 environment and run verify.py. Edit the working exercise baseline, then run --practice sa-savepoints. The lab uses disposable SQLite databases and actual SQLAlchemy/Alembic calls; it does not connect to application accounts, production databases or paid services. The app includes prepared results and explained questions without an editor or Python runtime.

Reject a duplicate insert while keeping valid work in the surrounding transaction.