Handling IntegrityError on concurrent inserts

An IntegrityError poisons the transaction: every subsequent statement raises PendingRollbackError until you roll back, so the handler must roll back first and only then decide what to do — and in most cases the right answer is to remove the race with an upsert rather than to retry. This is the failure mode behind upserts and conflict resolution strategies.

Quick Answer

Before — a handler that raises a different error than it catches:

try:
    session.add(User(email=email))
    await session.commit()
except IntegrityError:
    # PendingRollbackError: the session is unusable until it is rolled back
    existing = await session.scalar(select(User).where(User.email == email))
    return existing

After — roll back first, then decide:

from sqlalchemy.exc import IntegrityError


async def get_or_create_user(session: AsyncSession, email: str) -> User:
    try:
        user = User(email=email)
        session.add(user)
        await session.flush()          # flush, not commit: the caller owns the boundary
        return user
    except IntegrityError as exc:
        await session.rollback()       # FIRST — nothing else works until this runs
        if constraint_of(exc) != "uq_users_email":
            raise                      # a different constraint is a different bug
        return await session.scalar(select(User).where(User.email == email))

Better — no error path at all:

stmt = insert(User).values(email=email)
stmt = stmt.on_conflict_do_update(
    index_elements=[User.email],
    set_={"email": stmt.excluded.email},    # a no-op update, so RETURNING yields the row
).returning(User)
user = (await session.execute(stmt)).scalar_one()

The third version has no race, no rollback, no re-query and no constraint-name check. When the answer to "what should happen on conflict" is "give me the row that exists", the exception path was never needed.

Execution Context & Async Workflow Integration

The rollback requirement is a property of the database, not of SQLAlchemy: PostgreSQL aborts the entire transaction on a constraint violation, and no further statement is accepted on that connection until the transaction ends. SQLAlchemy surfaces this as PendingRollbackError, which is a clearer failure than the driver's own, but the constraint is the database's.

That has a consequence for anything doing several inserts in one transaction. A single failure discards all of them, which is right when they are one unit of work and wrong when they are independent. For the independent case, a savepoint per item scopes the failure:

# Async — one item's conflict must not discard the whole batch
async def import_users(session: AsyncSession, emails: list[str]) -> tuple[int, int]:
    created = skipped = 0
    for email in emails:
        try:
            async with session.begin_nested():        # SAVEPOINT per item
                session.add(User(email=email))
        except IntegrityError:
            skipped += 1                              # only this savepoint rolled back
        else:
            created += 1
    await session.commit()
    return created, skipped

Exiting the begin_nested() block on an exception rolls back to the savepoint rather than to the start of the transaction, so the successfully created users survive. That is the correct shape for an import, and it is worth noting the cost: a savepoint per row is a round trip per row, so for a large import the upsert form with DO NOTHING is both simpler and dramatically faster.

Under async there is one more trap. session.rollback() is a coroutine, so a handler that forgets to await it appears to work — the coroutine is created and discarded, the session stays poisoned, and the next statement raises PendingRollbackError from a line that looks unrelated. A RuntimeWarning: coroutine ... was never awaited is the clue, and it is easy to miss in a busy log.

Resolving Warnings, Errors & Common Mistakes

Advanced: Extracting the Constraint Name Portably

Deciding what to do about a violation requires knowing which rule was broken, and the string representation of the exception is the wrong place to find it — it varies by driver, by server version and by locale.

from __future__ import annotations

from sqlalchemy.exc import IntegrityError


def constraint_of(exc: IntegrityError) -> str | None:
    """The violated constraint's name, from the driver exception rather than its text."""
    orig = getattr(exc, "orig", None)
    if orig is None:
        return None
    # asyncpg exposes it directly on the exception object.
    name = getattr(orig, "constraint_name", None)
    if name:
        return name
    # psycopg exposes it through the diagnostics object.
    diag = getattr(orig, "diag", None)
    return getattr(diag, "constraint_name", None) if diag is not None else None


class SlotTaken(Exception):
    pass


class EmailAlreadyRegistered(Exception):
    pass


# One place where database rules become domain outcomes.
CONSTRAINT_ERRORS: dict[str, type[Exception]] = {
    "uq_users_email": EmailAlreadyRegistered,
    "uq_reservations_slot_active": SlotTaken,
}


async def translate_integrity_error(session: AsyncSession, exc: IntegrityError) -> None:
    """Roll back, then raise the domain error — or re-raise if we do not know this rule."""
    await session.rollback()
    domain = CONSTRAINT_ERRORS.get(constraint_of(exc) or "")
    if domain is None:
        raise exc
    raise domain() from exc

Two things make this hold up. The rollback happens before any decision, so the session is usable no matter which branch is taken. And an unmapped constraint re-raises the original error rather than being converted into something generic, so a new constraint added by a migration produces a clear failure rather than a misleading domain exception.

Naming constraints explicitly in migrations is what makes the mapping readable:

# In the migration, rather than accepting an auto-generated name
op.create_unique_constraint("uq_users_email", "users", ["email"])

An auto-generated name is stable enough to map, but a chosen one documents at the mapping site what rule it represents — and it survives a table rename, which auto-generated names often do not.

When a Retry Is Correct, and When It Is Not

Retrying is the reflex response to a concurrency error, and for IntegrityError it is usually the wrong one. The distinction is whether a subsequent attempt could plausibly succeed.

A unique-constraint violation is generally not retryable. The conflicting row now exists and will keep existing, so attempt two fails identically to attempt one. Retrying converts a fast, clear failure into a slow, identical one. The exception is a violation caused by a row that is itself transient — a lock or claim record another worker will delete — where a bounded retry with backoff is reasonable.

A serialisation failure is retryable, and is a different exception. Under SERIALIZABLE or REPEATABLE READ, PostgreSQL raises SQLSTATE 40001, which surfaces as OperationalError rather than IntegrityError, and the whole point of that error is that a retry may succeed. That retry must re-read everything, as described in setting transaction isolation level per session.

A deadlock is retryable. SQLSTATE 40P01 means the database chose your transaction as the victim, and the same work attempted again will usually succeed once the other transaction has finished.

# Retry only what is genuinely retryable, and always with a bound
RETRYABLE = {"40001", "40P01"}


def is_retryable(exc: Exception) -> bool:
    sqlstate = getattr(getattr(exc, "orig", None), "sqlstate", None)
    return sqlstate in RETRYABLE

Reading the sqlstate rather than the exception class is what keeps this honest: OperationalError covers connection failures too, and retrying a unit of work because the connection dropped is a different decision from retrying because the database asked you to.

The broader point is that an IntegrityError under concurrency is usually a signal that the write should have been expressed differently. If losing the race is a business outcome, name it and return it. If either row would do, upsert. A retry loop around a unique violation is the shape that has neither decided which of those it is nor removed the race.

Frequently Asked Questions

Related

• Upserts and Conflict Resolution Strategies — The parent guide: choosing the strategy that removes this error path entirely.
• Writing PostgreSQL ON CONFLICT DO UPDATE upserts in SQLAlchemy — The construct that replaces get-or-create and its exception handler.
• Implementing optimistic locking with version counters — The read-modify-write case, where a retry genuinely is the answer.
• Setting transaction isolation level per session — Serialisation failures, which are a different error and are retryable.