Savepoints: partial rollback within a transaction

PostgreSQL transactions are atomic: all or nothing. If one operation fails, you ROLLBACK and everything is undone. But sometimes you want a partial rollback: "this item in the batch failed, but the other 99 should persist".

The answer is savepoints. You mark a point in the transaction. If something fails afterward, you ROLLBACK TO that point — everything after it is undone, everything before it stays. It's the primitive that lets you do batch processing that tolerates partial failures.

In this lesson you'll learn savepoints in pure SQL (SAVEPOINT/ROLLBACK TO/RELEASE), the SQLAlchemy wrapper (session.begin_nested()), and the critical difference between savepoints and a simple try/except — which does NOT solve the problem in PostgreSQL.

The problem savepoints solve

Consider: you import 100 items. Item 50 has a duplicate key.

Without savepoints (the intuition that fails)

async def import_batch(session, items):
    for item in items:
        try:
            session.add(MyModel(**item))
            await session.flush()
        except IntegrityError as e:
            print(f"Failed: {item}")
            # Continue with the next one
            continue

    await session.commit()

What you expect: items 1-49 and 51-100 persist, item 50 gets reported as an error.

What actually happens: item 50 fails, the transaction is left in an aborted state. Any subsequent query fails with InFailedSqlTransaction. Items 51-100 never run, items 1-49 roll back at the end. You lose the whole batch.

PostgreSQL is strict: once an operation fails in a transaction, the entire transaction is blocked until ROLLBACK. try/except doesn't fix it.

With savepoints (the solution)

async def import_batch(session, items):
    for item in items:
        savepoint_name = f"sp_{item['id']}"
        await session.execute(text(f"SAVEPOINT {savepoint_name}"))

        try:
            session.add(MyModel(**item))
            await session.flush()
            await session.execute(text(f"RELEASE SAVEPOINT {savepoint_name}"))
        except IntegrityError:
            await session.execute(text(f"ROLLBACK TO SAVEPOINT {savepoint_name}"))
            print(f"Failed: {item}")

    await session.commit()

Now item 50 fails, you ROLLBACK TO sp_50 (undo item 50), and continue with item 51. The items that succeeded stay, the ones that failed are reported. At the end, the commit persists everything.

Pure SQL

BEGIN;

INSERT INTO items (name) VALUES ('item 1');  -- success

SAVEPOINT sp1;
INSERT INTO items (name) VALUES ('item 2');  -- success
RELEASE SAVEPOINT sp1;  -- no longer needed

SAVEPOINT sp2;
INSERT INTO items (name) VALUES ('item 3');  -- say this fails on unique
ROLLBACK TO SAVEPOINT sp2;  -- undoes item 3 but NOT 1 and 2

INSERT INTO items (name) VALUES ('item 4');  -- success (the tx isn't aborted)

COMMIT;
-- Result: items 1, 2, 4 persisted. Item 3 never made it.

Three commands:

• SAVEPOINT name: creates a savepoint. Operations after it can be rolled back to here.
• RELEASE SAVEPOINT name: discards the savepoint (the changes between the savepoint and the release remain permanent in the tx). Not strictly necessary but it cleans up the state.
• ROLLBACK TO SAVEPOINT name: undoes all changes since the savepoint. The tx continues.

SQLAlchemy 2.0: session.begin_nested()

SQLAlchemy exposes savepoints via session.begin_nested(), which uses a context manager:

from sqlalchemy.exc import IntegrityError


async def import_batch(session, items):
    success_count = 0
    errors = []

    for item in items:
        try:
            async with session.begin_nested():
                session.add(MyModel(**item))
                # implicit flush on exiting the nested block
            success_count += 1
        except IntegrityError as e:
            errors.append({"item": item, "error": str(e)})

    await session.commit()
    return {"success": success_count, "errors": errors}

async with session.begin_nested():

• Creates a savepoint on entry.
• If the block exits without an exception → RELEASE SAVEPOINT (commit of the savepoint).
• If the block exits with an exception → ROLLBACK TO SAVEPOINT (discard).
• The exception propagates outward (catchable with an outer try/except).

Much cleaner than manual SQL.

Complete batch-processing pattern

from typing import Any
from pydantic import BaseModel, ValidationError
from sqlalchemy import insert
from sqlalchemy.exc import IntegrityError, DataError
from sqlalchemy.ext.asyncio import AsyncSession


class BatchResult(BaseModel):
    success: int
    failed: int
    errors: list[dict[str, Any]]


async def import_batch_with_savepoints(
    session: AsyncSession,
    raw_items: list[dict],
) -> BatchResult:
    """Import items one by one with partial rollback."""

    success = 0
    errors = []

    for i, raw in enumerate(raw_items):
        try:
            # Pydantic validation first (doesn't need a savepoint)
            try:
                validated = ItemCreate(**raw)
            except ValidationError as e:
                errors.append({
                    "index": i,
                    "stage": "validation",
                    "error": e.errors()
                })
                continue

            # DB operations inside a savepoint
            async with session.begin_nested():
                session.add(MyModel(**validated.model_dump()))
                # implicit flush on exit
            success += 1

        except IntegrityError as e:
            errors.append({
                "index": i,
                "stage": "db_integrity",
                "error": str(e.orig),
            })
        except DataError as e:
            errors.append({
                "index": i,
                "stage": "db_data",
                "error": str(e.orig),
            })
        except Exception as e:
            errors.append({
                "index": i,
                "stage": "unknown",
                "error": str(e),
            })

    await session.commit()

    return BatchResult(
        success=success,
        failed=len(errors),
        errors=errors,
    )

It reports each error with context (index, stage, detail). The client can re-process errors.

Endpoint with a partial-success response

from fastapi import APIRouter, Depends, status
from app.deps import get_db


router = APIRouter()


@router.post("/items/batch", response_model=BatchResult)
async def batch_import(
    items: list[dict],
    db: AsyncSession = Depends(get_db),
):
    result = await import_batch_with_savepoints(db, items)

    # Status code: 207 Multi-Status if there were errors
    if result.failed > 0 and result.success > 0:
        # Some OK, some failed — partial success
        # FastAPI doesn't expose 207 directly, return 200 with an informative body
        pass

    return result

The client receives:

{
  "success": 47,
  "failed": 3,
  "errors": [
    {"index": 12, "stage": "validation", "error": [...]},
    {"index": 50, "stage": "db_integrity", "error": "duplicate key value..."},
    {"index": 89, "stage": "db_data", "error": "value too long..."}
  ]
}

Much more useful than "one failed, everything rolled back, error 500".

Critical differences from a simple try/except

Case 1: PostgreSQL marks the tx as aborted

Without a savepoint:

try:
    session.add(item_with_dup_key)
    await session.flush()
except IntegrityError:
    pass  # ❌ The tx is aborted — you can't continue

# This fails with InFailedSqlTransaction
session.add(another_item)
await session.flush()
# Error: current transaction is aborted, commands ignored until end of transaction block

With a savepoint:

try:
    async with session.begin_nested():
        session.add(item_with_dup_key)
except IntegrityError:
    pass  # ✅ The savepoint rolled back the item, the tx is still OK

session.add(another_item)
await session.flush()  # Works normally

With a savepoint, the outer tx stays in a valid state. Without a savepoint, it's left aborted.

Case 2: rolling back changes in the DB

Without a savepoint, if a partial operation succeeded before the error:

try:
    session.add(item_a)  # success
    await session.flush()
    session.add(item_b)  # fails
    await session.flush()
except IntegrityError:
    pass

# item_a is NOT rolled back — it's in the tx
# But the tx is aborted, so the final commit fails → everything rolls back

With a savepoint, you can control what to do:

try:
    async with session.begin_nested():
        session.add(item_a)
        session.add(item_b)
        # If item_b fails, BOTH roll back (because they're in the same savepoint)
except IntegrityError:
    pass

Fine-grained rollback control.

Savepoint overhead

Savepoints have a cost:

• Each savepoint creates internal structures in PostgreSQL.
• ROLLBACK TO requires reverting WAL entries.
• In huge batches (millions of savepoints), the overhead adds up.

Typical benchmark:

Overhead ~30% for large batches. For medium batches (<10k items), acceptable. For millions, consider other approaches.

When NOT to use savepoints

1. A huge batch where you can validate up front and bulk insert.

If you have 1M items and can validate them up front (Pydantic) + bulk insert with COPY, the savepoint overhead (1M of them) outweighs the benefit.

2. Operations where you do NOT want tolerance for partial failures.

A financial case where "all or nothing" is the requirement. An atomic transaction without savepoints, fail-all if something goes wrong.

3. Loops within loops (nested savepoints).

PostgreSQL supports nested savepoints (a savepoint inside a savepoint), but the overhead multiplies. For a batch inside a batch, consider restructuring.

When to use savepoints

• A batch of moderate size (10-10k items) where you tolerate partial failures.
• Imports from external sources with imperfect data quality.
• Workflows with multiple steps where one step can fail but the rest must continue.
• Tests where you want partial rollback between setup and assertions.

Advanced patterns

Pattern 1: savepoint with retry

async def import_with_retry(session, item, max_retries=3):
    """If it fails on deadlock, retry."""
    for attempt in range(max_retries):
        try:
            async with session.begin_nested():
                session.add(MyModel(**item))
            return True  # success
        except DeadlockDetected:
            if attempt < max_retries - 1:
                await asyncio.sleep(0.1 * (2 ** attempt))  # exponential backoff
                continue
            raise
        except IntegrityError:
            return False  # data error, no retry
    return False

Pattern 2: nested savepoint

async def complex_import(session, batch):
    """Top-level savepoint for "the whole batch or nothing", inner savepoints per item."""
    try:
        async with session.begin_nested():  # Outer savepoint
            for item in batch:
                try:
                    async with session.begin_nested():  # Inner savepoint per item
                        await process(session, item)
                except RecoverableError:
                    pass  # skip this item, continue batch
            # If we reach here without unrecoverable error, commit batch
    except UnrecoverableError:
        # Whole batch rolls back
        raise

Useful for hierarchical operations: "the whole batch fails only if something serious like X happens, otherwise continue with recoverable items".

Pattern 3: savepoint for tests

@pytest.fixture
async def session_with_rollback(db_engine):
    """Fixture that rolls back changes at the end of the test."""
    SessionLocal = async_sessionmaker(db_engine)

    async with SessionLocal() as session:
        # Outer transaction
        async with session.begin():
            # Inner savepoint
            async with session.begin_nested():
                yield session
            # Implicit rollback at end of nested block
        # Outer commit — but everything inside was in a savepoint, so nothing persisted

The test keeps the DB clean between tests without a truncate (faster).

Traps and common mistakes

1. Assuming that try/except is enough.

try/except doesn't roll back changes in the DB. PostgreSQL marks the tx as aborted. You need a savepoint to tolerate it.

2. Forgetting session.commit() at the end.

Savepoints are sub-points of a tx. The outer tx still needs an explicit commit (or auto-commit if configured).

3. begin_nested() outside a transaction.

# ❌ Without an outer tx, there's no savepoint
async with session.begin_nested():
    session.add(item)

begin_nested() requires an outer tx. Make sure await session.begin() or equivalent runs before.

4. Savepoints in engine.connect() (Core, not ORM).

# Core pattern
async with engine.connect() as conn:
    async with conn.begin():
        async with conn.begin_nested():  # savepoint
            # ...

A different API from the ORM session. The idea is the same.

5. Naming conflicts in pure-SQL savepoints.

-- If you name two savepoints the same:
SAVEPOINT sp;
INSERT ...;
SAVEPOINT sp;  -- It's OK, but the first one gets "shadowed"

PostgreSQL allows the same name — ROLLBACK TO goes to the most recent one. If you want clear references, use unique names.

6. Performance assumption: "savepoints are free."

They're not free — ~30% overhead in large batches. For 10M items, other patterns win.

7. Nesting savepoints excessively.

3-4 levels is fine. More becomes confusing and the cost grows.

8. A generic catch instead of a specific one.

# ❌ Catches everything, even programming bugs
try:
    async with session.begin_nested():
        await operation()
except Exception:
    pass

# ✅ Specific catches
try:
    async with session.begin_nested():
        await operation()
except IntegrityError:
    pass  # only data integrity issues

A generic catch hides bugs.

Exercise: implement batch processing

Setup:

class Item(Base):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)
    value: Mapped[int] = mapped_column(Integer)

Step 1: try a batch without savepoints.

async def batch_no_savepoints(session, items):
    for item in items:
        try:
            session.add(Item(name=item["name"], value=item["value"]))
            await session.flush()
        except IntegrityError:
            print(f"Failed: {item}")
            # continue to the next

    await session.commit()


# Test with a duplicate
test_items = [
    {"name": "a", "value": 1},
    {"name": "b", "value": 2},
    {"name": "a", "value": 3},  # duplicate name
    {"name": "c", "value": 4},
]

await batch_no_savepoints(session, test_items)
# Expected: "Failed" on item 3, items a/b/c persisted
# Reality: ?

What happens? Check the table afterward.

Step 2: the same batch with savepoints.

async def batch_with_savepoints(session, items):
    for item in items:
        try:
            async with session.begin_nested():
                session.add(Item(name=item["name"], value=item["value"]))
        except IntegrityError:
            print(f"Failed: {item}")

    await session.commit()


await batch_with_savepoints(session, test_items)

What happens now?

Step 3: measure the overhead.

import time

# With 1000 items, all valid
items = [{"name": f"unique-{i}", "value": i} for i in range(1000)]

# Without savepoints (1 tx)
session.execute(text("TRUNCATE items"))
await session.commit()

start = time.perf_counter()
async with session.begin():
    for item in items:
        session.add(Item(name=item["name"], value=item["value"]))
elapsed_no_sp = time.perf_counter() - start

# With savepoints
session.execute(text("TRUNCATE items"))
await session.commit()

start = time.perf_counter()
async with session.begin():
    for item in items:
        async with session.begin_nested():
            session.add(Item(name=item["name"], value=item["value"]))
elapsed_with_sp = time.perf_counter() - start

print(f"No savepoints: {elapsed_no_sp:.2f}s")
print(f"With savepoints: {elapsed_with_sp:.2f}s")
print(f"Overhead: {(elapsed_with_sp / elapsed_no_sp - 1) * 100:.1f}%")

How much overhead?

Step 4: a real case — a partial-success endpoint.

Implement POST /items/batch that:

1. Accepts a list of items.
2. Processes each one with a savepoint.
3. Returns {"success": N, "failed": M, "errors": [...]}.

Step 5: test the endpoint with mixed data.

Failed: {'name': 'a', 'value': 3}
sqlalchemy.exc.PendingRollbackError: This Session's transaction has been rolled back due to a previous exception during flush.

Failed: {'name': 'a', 'value': 3}

No savepoints: 0.45s
With savepoints: 0.62s
Overhead: 37.8%

Summary and next step

What you learned:

• Savepoints = points within a tx where you can do a partial rollback.
• Pure SQL: SAVEPOINT name, ROLLBACK TO SAVEPOINT name, RELEASE SAVEPOINT name.
• SQLAlchemy: async with session.begin_nested() — a clean context manager.
• Difference from try/except: PostgreSQL aborts the tx on error without a savepoint. A savepoint lets you continue.
• Overhead ~30-40%: acceptable for medium batches, problematic for huge ones.
• Patterns: simple per-item, with retry, nested, test fixtures.
• Don't use for: huge batches, operations that require strict atomicity.

Before moving on, you should be able to:

• Implement batch processing with session.begin_nested().
• Distinguish between recoverable failures (continue) and unrecoverable ones (rollback all).
• Argue against a simple try/except for batch processing in PostgreSQL.
• Decide when savepoints are appropriate vs upfront validation.

In the next lesson we combine the module's two topics: advisory locks + savepoints in a single job runner. You'll implement the complete pattern: a global advisory lock for single-instance, a savepoint per batch item for fault tolerance. It's exactly what you'll use in the final capstone project.

Resources

1. PostgreSQL Docs — SAVEPOINT — reference.
2. PostgreSQL Docs — ROLLBACK TO SAVEPOINT — reference.
3. SQLAlchemy 2.0 — Session.begin_nested — reference.
4. SQLAlchemy — Joining a Session into an External Transaction (for testing) — a pattern for fixtures.
5. Brandur Leach — Postgres patterns — savepoints in production.

Lesson 06 of 08 — Module 6 — Advanced PostgreSQL for Backend Guide