Using savepoints with begin_nested() in AsyncSession

Wrap each statement that is allowed to fail in async with session.begin_nested(): and catch the exception outside that block — PostgreSQL aborts the whole transaction on any error, and only a ROLLBACK TO SAVEPOINT lets it continue. This guide belongs to transaction isolation and commit strategies.

Quick Answer

Without a savepoint, the first IntegrityError in an import poisons the transaction: PostgreSQL refuses every further statement until it is rolled back, and SQLAlchemy reports that as PendingRollbackError.

Before — one failure aborts everything that follows:

from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from shop.models import Customer


async def import_customers(session: AsyncSession, rows: list[dict]) -> int:
    imported = 0
    for row in rows:
        try:
            session.add(Customer(**row))
            await session.flush()
            imported += 1
        except IntegrityError:
            continue      # the transaction is already aborted
    await session.commit()
    return imported
# sqlalchemy.exc.PendingRollbackError: This Session's transaction has been rolled back due to
# a previous exception during flush. To begin a new transaction with this Session, first issue
# Session.rollback().

After — a savepoint per row, caught outside the block:

from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from shop.models import Customer


async def import_customers(session: AsyncSession, rows: list[dict]) -> tuple[int, list[str]]:
    imported, rejected = 0, []
    for row in rows:
        try:
            async with session.begin_nested():
                session.add(Customer(**row))
            imported += 1
        except IntegrityError:
            rejected.append(row["email"])
    await session.commit()
    return imported, rejected

Leaving the begin_nested() block flushes the new Customer and releases the savepoint. When the flush fails, the block rolls back to the savepoint and re-raises, and the except clause outside it records the rejection. The outer transaction is intact and commits every row that succeeded.

Execution Context & Async Workflow Integration

PostgreSQL treats a transaction as all-or-nothing in a stricter way than most databases: after any error, the transaction enters an aborted state and every subsequent statement fails with current transaction is aborted, commands ignored until end of transaction block, which asyncpg raises as InFailedSQLTransactionError. The only statements it accepts are ROLLBACK and ROLLBACK TO SAVEPOINT.

SQLAlchemy mirrors that on the session. When a flush fails, the session rolls back its transaction internally and marks itself as needing an explicit rollback(); any further use raises PendingRollbackError until you call it. That is correct and protective — it stops code from assuming earlier work in the transaction survived — and it is also why the naive try/except around a flush cannot continue.

AsyncSession.begin_nested() returns an AsyncSessionTransaction that manages a SQL SAVEPOINT. Entering it flushes any pending changes, so the savepoint marks a clean point, and emits SAVEPOINT sa_savepoint_1. Leaving it normally flushes again and emits RELEASE SAVEPOINT. Leaving it with an exception emits ROLLBACK TO SAVEPOINT, restores the session to its state at the savepoint, and re-raises. The outer transaction was never aborted from PostgreSQL's point of view, because the rollback happened within it.

Session state after a rolled-back savepoint follows the database. Objects added inside the block are expunged, so they are no longer pending. Objects modified inside the block are expired, so their next access reloads current values — which under async means an explicit await session.refresh(obj) rather than attribute access, for the reasons in fixing GreenletSpawnError. Objects untouched inside the block keep their state.

Nothing inside a released savepoint is durable. RELEASE SAVEPOINT merges the work into the outer transaction, which still commits or rolls back as a whole; a crash before the outer commit() loses released work too. A savepoint is a way to undo part of a transaction, not a way to commit part of one. If partial results must survive regardless, commit in chunks instead — the trade-offs are in the parent guide to transaction isolation and commit strategies.

Resolving Warnings, Errors & Common Mistakes

The inside-versus-outside mistake is the one to watch in review. This version looks almost identical to the correct code and does not work:

async with session.begin_nested():
    try:
        session.add(Customer(**row))
        await session.flush()
    except IntegrityError:
        pass   # the savepoint block never sees the error
# RELEASE SAVEPOINT is attempted on an aborted transaction and fails

The context manager can only roll back to the savepoint if the exception propagates through it. Swallowing the error inside means the block exits "successfully", tries to release a savepoint in a transaction PostgreSQL has already aborted, and fails with a second, more confusing error.

A related subtlety concerns IntegrityError raised by other statements. A unique violation from an UPDATE, a foreign-key violation from a DELETE, or a check-constraint violation all abort the transaction the same way, and all are recoverable with the same pattern. Errors that are not about the data — a lost connection, a serialization failure — are not: the right response to those is a rollback and a retry of the whole unit, as in handling serialization failures with retry logic.

Advanced: Batch Savepoints With Row-Level Fallback

A savepoint per row costs two extra round trips per row — SAVEPOINT and RELEASE — on top of the insert. For a few hundred rows that is invisible. For a hundred thousand it dominates the import. The usual optimisation is to optimise for the common case, where a batch has no bad rows, and fall back to row-by-row only for batches that fail.

from collections.abc import Sequence

from sqlalchemy import insert
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from shop.models import Customer


async def import_in_batches(
    session: AsyncSession, rows: Sequence[dict], batch_size: int = 500
) -> tuple[int, list[dict]]:
    imported, rejected = 0, []
    for start in range(0, len(rows), batch_size):
        batch = rows[start:start + batch_size]
        try:
            async with session.begin_nested():
                await session.execute(insert(Customer), batch)
            imported += len(batch)
            continue
        except IntegrityError:
            pass
        # The batch had at least one bad row: find it, keep the rest.
        for row in batch:
            try:
                async with session.begin_nested():
                    await session.execute(insert(Customer), [row])
                imported += 1
            except IntegrityError:
                rejected.append(row)
    await session.commit()
    return imported, rejected

The batch insert uses Core-style insert(Customer) with a list of dictionaries through the session, which SQLAlchemy executes as an efficient multi-row insert rather than one ORM flush per object. When the data is clean, the cost is one savepoint per five hundred rows.

Before reaching for either pattern, check whether the database can resolve the conflict without an error at all. If "bad row" means "duplicate key", INSERT ... ON CONFLICT DO NOTHING skips duplicates inside a single statement, with no aborted transaction and no savepoints, and RETURNING tells you which rows were inserted. Bulk upserting rows with INSERT ON CONFLICT covers it. Savepoints are for failures the database cannot resolve on its own: check constraints, foreign keys to rows that do not exist, or triggers that raise.

Savepoints in Long Transactions and in Tests

Two places use savepoints heavily enough that their cost and behaviour deserve a closer look: long-running transactions that accumulate many of them, and test suites that use them to isolate tests.

In PostgreSQL, every savepoint that writes anything becomes a subtransaction with its own transaction ID. Each backend caches up to 64 subtransaction IDs for the transaction it is running. Past that, the cache overflows, and every other session checking row visibility against that transaction has to consult pg_subtrans instead of the in-memory cache. On a busy primary this adds contention; on hot-standby replicas it is a well-known cause of sudden, system-wide query slowdowns while the overflowed transaction is open.

The practical rule is to avoid holding hundreds of writing savepoints in one open transaction. The batch pattern above already reduces the count by the batch size. For very large imports, commit in chunks as well, so no single transaction lives long enough to accumulate many:

from sqlalchemy.ext.asyncio import async_sessionmaker


async def import_large_file(Session: async_sessionmaker, rows: list[dict]) -> None:
    chunk = 10_000
    for start in range(0, len(rows), chunk):
        async with Session() as session:
            await import_in_batches(session, rows[start:start + chunk], batch_size=500)
            # import_in_batches commits: at most 20 savepoints per transaction here,
            # plus any row-level fallbacks for failed batches.

Tests use savepoints for the opposite reason: to make every test's writes disappear. The pattern binds the session to a connection with an open outer transaction and join_transaction_mode="create_savepoint", so that when code under test calls commit(), SQLAlchemy releases a savepoint instead of committing, and the fixture rolls back the outer transaction at teardown. It is fast and fully isolating, and it means code that itself uses begin_nested() runs as a savepoint inside a savepoint inside a transaction — which PostgreSQL handles without complaint. The fixture is described in rolling back database state between async tests.

Frequently Asked Questions

Related

• Transaction Isolation and Commit Strategies — The parent guide: isolation levels, commit boundaries and retries.
• Handling serialization failures with retry logic — Errors a savepoint cannot fix, and retrying the whole unit.
• Setting transaction isolation level per session — How isolation interacts with nested work.
• Handling IntegrityError on concurrent inserts — Recovering from races between concurrent writers.