Upserts and Conflict Resolution Strategies

Use insert(...).on_conflict_do_update(index_elements=[...], set_={...}) rather than checking for a row and then inserting it — the check-then-act version has a window between the two statements in which another transaction can insert the same row, and under any real concurrency it will. This page sits under advanced query patterns and bulk data operations, which covers the wider write-path picture.

Concept & Execution Model

The window between the check and the insert Two implementations of the same upsert. The select-then-insert version runs a SELECT that finds no row, then an INSERT — and between those two statements another transaction can insert the same key, so the second INSERT raises IntegrityError. Widening the transaction does not help, because under READ COMMITTED the SELECT cannot see the other transaction uncommitted row, and under SERIALIZABLE it becomes a serialisation failure instead. A single INSERT ... ON CONFLICT statement has no window at all: the database resolves the conflict atomically as part of the statement. SELECT, then INSERT the SELECT finds no row another transaction inserts the same key our INSERT raises IntegrityError and no isolation level closes the window INSERT ... ON CONFLICT DO UPDATE one statement, no window the database resolves it atomically returns the row either way safe at any isolation level This is not a rare race. At a hundred requests per second against a popular key it happens continuously, and the retry loop people add instead is strictly more code.

An upsert is a single statement that inserts a row or, if a uniqueness conflict occurs, updates the existing one instead. The database performs the conflict detection while holding the appropriate lock, which is why the operation is atomic and why no application-level check can match it.

SQLAlchemy exposes this through the dialect-specific insert() construct — sqlalchemy.dialects.postgresql.insert rather than the generic sqlalchemy.insert — because the syntax and semantics differ between backends.

from __future__ import annotations

import decimal

from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.ext.asyncio import AsyncSession

from myapp.models import Product


async def upsert_product(session: AsyncSession, sku: str, name: str,
                         price: decimal.Decimal) -> Product:
    stmt = insert(Product).values(sku=sku, name=name, price=price)
    stmt = stmt.on_conflict_do_update(
        index_elements=[Product.sku],          # must match a real unique index
        set_={
            "name": stmt.excluded.name,        # excluded = the row we proposed
            "price": stmt.excluded.price,
            "updated_at": func.now(),
        },
    ).returning(Product)

    result = await session.execute(stmt)
    return result.scalar_one()

Three details carry the whole construct. index_elements names the columns of an existing unique index or constraint; if no such index exists, PostgreSQL raises there is no unique or exclusion constraint matching the ON CONFLICT specification rather than silently doing something else, which is the good outcome. stmt.excluded is the proposed row — the values you passed to values() — and referring to Product.price instead would read the stored value, which compiles cleanly and quietly does nothing. And .returning(Product) gives you the row in both branches, which is what makes the function total rather than sometimes returning None.

Query Construction & Async Execution Patterns

The statement is built the same way under sync and async; only execution differs. What does change under async is that the retry loops people write around naive upserts become considerably more awkward, which is another argument for not needing them.

# Async — upsert and read the result in one round trip
from sqlalchemy import func
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

engine = create_async_engine("postgresql+asyncpg://app:secret@db.internal/orders")
Session = async_sessionmaker(engine, expire_on_commit=False)


async def record_reading(sensor_id: str, taken_at, value: float) -> None:
    """Only the newest reading wins, whatever order messages arrive in."""
    async with Session() as session, session.begin():
        stmt = insert(Reading).values(
            sensor_id=sensor_id, taken_at=taken_at, value=value
        )
        stmt = stmt.on_conflict_do_update(
            index_elements=[Reading.sensor_id],
            set_={"taken_at": stmt.excluded.taken_at, "value": stmt.excluded.value},
            # The condition that makes out-of-order delivery safe.
            where=Reading.taken_at < stmt.excluded.taken_at,
        )
        await session.execute(stmt)

# Sync — identical construction
def record_reading_sync(session, sensor_id: str, taken_at, value: float) -> None:
    stmt = insert(Reading).values(sensor_id=sensor_id, taken_at=taken_at, value=value)
    stmt = stmt.on_conflict_do_update(
        index_elements=[Reading.sensor_id],
        set_={"taken_at": stmt.excluded.taken_at, "value": stmt.excluded.value},
        where=Reading.taken_at < stmt.excluded.taken_at,
    )
    session.execute(stmt)

The where clause on on_conflict_do_update is the piece most people do not know exists, and it removes an entire class of bug. Without it, a delayed message carrying an older reading overwrites the newer one; with it, the update simply does not happen and the statement affects zero rows. Note that the predicate references the stored row via Reading.taken_at and the proposed row via stmt.excluded.taken_at — that asymmetry is the whole point.

Because this is a single statement, it composes with the bulk techniques in high-performance bulk inserts and updates: passing a list of dicts turns it into an executemany, and the conflict handling applies per row.

Four strategies, four different intents Four approaches. DO NOTHING is right when a duplicate simply means the work is already done — an idempotent event handler, a dedupe table — and it returns no row, which callers must handle. DO UPDATE is right when the incoming row is authoritative and should overwrite. A conditional DO UPDATE with a WHERE clause is right when only newer data should win, which is how out-of-order message delivery is handled without losing the newest value. And optimistic locking with a version counter is right when the new value depends on the old one, because ON CONFLICT alone cannot express read-modify-write safely. DO NOTHING a duplicate means "already done" returns no row — handle that idempotent handlers DO UPDATE the incoming row is authoritative last writer wins the common case DO UPDATE ... WHERE only newer data wins compare a timestamp or sequence out-of-order delivery version counter the new value depends on the old ON CONFLICT cannot do this read-modify-write The last column is the one people reach for ON CONFLICT to solve and cannot: incrementing a balance by an amount you read earlier needs the version check, not an upsert.

State Management & Session Boundaries

An upsert executed as a Core statement bypasses the ORM unit of work entirely: it does not populate the identity map, it does not fire ORM events, and instances already loaded in the session are not updated to reflect it. That is usually what you want for a write-heavy path, and it is occasionally a surprise.

# The session's copy is now stale — the upsert went straight to the database
product = await session.get(Product, product_id)      # price = 10.00
await upsert_product(session, product.sku, product.name, decimal.Decimal("12.00"))
assert product.price == decimal.Decimal("10.00")      # still the old value

await session.refresh(product)                        # now 12.00

Three rules keep this predictable. Prefer .returning() and use the returned row rather than a previously loaded instance, which sidesteps staleness entirely. If an ORM instance must stay authoritative, refresh() it after the upsert — or set synchronize_session appropriately if you are using an ORM-enabled update. And never mix the two in one unit of work without deciding which is the source of truth, because the flush order between a pending ORM change and an explicit Core statement is not something to reason about under time pressure.

Transaction boundaries matter for a different reason: ON CONFLICT DO UPDATE takes a row lock on the conflicting row for the remainder of the transaction. A long transaction that upserts a popular key early and then does slow work holds that lock throughout, and every other transaction touching that key waits. The remedy is the same as everywhere else — keep the transaction as short as the work allows, as the hold-time reasoning in tuning connection pools for cloud databases sets out.

Advanced Conflict Patterns: Counters, Merges and Partial Indexes

Three patterns come up repeatedly and each has a non-obvious form.

Incrementing rather than replacing. When the new value should be derived from the stored one, the set_ mapping can reference the existing column:

stmt = insert(DailyTotal).values(day=day, sku=sku, units=units)
stmt = stmt.on_conflict_do_update(
    index_elements=[DailyTotal.day, DailyTotal.sku],
    # The stored value plus the proposed one — computed by the database, so it is
    # correct under any concurrency without a read-modify-write in the application.
    set_={"units": DailyTotal.units + stmt.excluded.units},
)

This is genuinely atomic: a hundred concurrent workers incrementing the same counter produce the correct total with no locking in the application, because each UPDATE reads and writes under its own row lock.

Merging JSON rather than replacing it. For a JSONB column, replacing loses whatever keys the incoming row did not carry:

stmt = stmt.on_conflict_do_update(
    index_elements=[Profile.user_id],
    set_={"attributes": Profile.attributes.op("||")(stmt.excluded.attributes)},
)

Conflicting on a partial index. When uniqueness only applies to a subset of rows — one active subscription per customer, with historical rows retained — the conflict target must name the same predicate as the index:

# The index: CREATE UNIQUE INDEX ... ON subscriptions (customer_id) WHERE active
stmt = stmt.on_conflict_do_update(
    index_elements=[Subscription.customer_id],
    index_where=Subscription.active.is_(True),   # must match the index predicate
    set_={"plan": stmt.excluded.plan},
)

Omitting index_where produces no unique or exclusion constraint matching the ON CONFLICT specification, because PostgreSQL cannot tell which index you meant. The error is clear, which is more than can be said for the excluded mistake.

Four things to know before using ON CONFLICT Four requirements. There must be a unique index or constraint on the conflict target, because that is what the database detects the conflict against — no index means no conflict and a duplicate row. The set_ mapping decides which columns are overwritten, and forgetting one silently keeps the stale value. The excluded pseudo-table holds the row that was proposed, which is how the update reaches the incoming values rather than the stored ones. And an INSERT that conflicts still consumes a sequence value on PostgreSQL, so identifier gaps are expected and are not a sign of anything wrong. a unique index or constraint on the target index_elements=[...] must match one that exists without it there is no conflict, only a duplicate row the set_ mapping decides what is overwritten a column left out silently keeps its stale value this is the most common upsert bug in review excluded holds the row you proposed set_={"total": stmt.excluded.total} — not the stored value referencing the table name instead reads the OLD row a conflicting INSERT still burns a sequence value identifier gaps are normal and not a defect do not build anything that assumes contiguous ids The excluded reference is the one that produces silently wrong data rather than an error: set_={"total": Order.total} compiles fine and writes the value that was already there.

Hybrid Architectures & Migration Strategies

Not every backend has ON CONFLICT, and not every upsert should be one.

SQLite supports the same syntax through sqlalchemy.dialects.sqlite.insert, which matters because it means a test suite on SQLite can exercise upsert code paths — with the caveat that SQLite serialises writers, so it cannot demonstrate that the upsert is concurrency-safe. That verification needs a real PostgreSQL, as running tests against a PostgreSQL testcontainer describes.

MySQL offers ON DUPLICATE KEY UPDATE through its own dialect construct, with different semantics: it conflicts on any unique key rather than a named one, which makes it less precise and occasionally surprising on a table with several unique constraints.

Writing portable code across these means either a small dialect shim or accepting a database-specific query module. The shim is usually the wrong trade: it converges on the least capable dialect and loses where, partial-index targets and RETURNING. A module of PostgreSQL-specific queries behind a repository interface is clearer, and is the same seam described in mocking vs using a real database in SQLAlchemy tests.

Migrating an existing select-then-insert to an upsert is mechanical, with one prerequisite that is easy to miss: the unique index has to exist. On a table that has been accumulating duplicates because nothing enforced uniqueness, creating it requires deduplicating first, and doing that without locking the table needs the expand-and-contract approach from zero-downtime schema migration strategies.

Choosing Between an Upsert and a Guarded Insert

Not every write that could be an upsert should be one, and the distinction is about intent rather than about performance.

An upsert says "this key should end up holding these values, and I do not care whether it existed before". That is exactly right for a projection built from an event stream, a cache table, a nightly import, or any write where the incoming data is authoritative.

A guarded insert — DO NOTHING, or a plain insert with the IntegrityError handled — says "create this if it does not exist, and tell me if it did". That is right when the two cases mean different things to the caller: registering a user, claiming a job, reserving a slot. Turning those into a DO UPDATE would silently overwrite someone else's row and report success.

And neither is right when the new value depends on the old one in a way SQL cannot express in a single expression. Recomputing an aggregate from several tables, applying a state machine transition, or validating against the previous value are all read-modify-write, and their correct form is a version counter with a retry.

# A guarded insert: the caller needs to know which branch happened
stmt = (
    insert(Reservation)
    .values(slot_id=slot_id, user_id=user_id)
    .on_conflict_do_nothing(index_elements=[Reservation.slot_id])
    .returning(Reservation.id)
)
created_id = (await session.execute(stmt)).scalar_one_or_none()
if created_id is None:
    raise SlotAlreadyReserved(slot_id)      # a real outcome, not an error to swallow

The scalar_one_or_none() is doing real work here: DO NOTHING returns no row when the conflict fires, so the None is the answer rather than a failure to handle. Writing scalar_one() instead turns a normal business outcome into a NoResultFound traceback, which is the single most common bug in code using DO NOTHING.

Production Pitfalls & Anti-Patterns

  • sqlalchemy.exc.ProgrammingError: there is no unique or exclusion constraint matching the ON CONFLICT specification — the columns in index_elements do not correspond to a unique index, or the index is partial and index_where was omitted. Create the index, or name the predicate.
  • set_ referencing the model instead of excluded. set_={"price": Product.price} compiles and writes the value that was already stored, so the upsert silently does nothing on conflict. This is the most damaging mistake on this page because there is no error.
  • A column missing from set_. On conflict, that column keeps its old value. For updated_at in particular, the row then looks untouched.
  • Assuming contiguous identifiers. A conflicting INSERT still advances the sequence on PostgreSQL, so gaps are normal. Anything treating identifiers as a dense range is wrong for other reasons too.
  • DO NOTHING with .returning() and scalar_one(). When nothing is inserted, nothing is returned, and scalar_one() raises NoResultFound. Use scalar_one_or_none() and handle the None.
  • Upserting inside a long transaction. The row lock is held until commit, so a popular key becomes a serialisation point for every other writer.
  • Using an upsert for read-modify-write. Incrementing by an amount computed in Python from an earlier read is still a race — the increment has to be expressed in SQL, or guarded by a version counter.

Verifying an Upsert Is Actually Concurrency-Safe

An upsert that has never been tested under concurrency is an upsert nobody has verified, and the test is short enough that there is no excuse for skipping it. It needs a real PostgreSQL — SQLite serialises writers, so the race can never occur there — and it needs genuinely separate sessions, because two coroutines sharing one session serialise on the session itself rather than in the database.

"""Assert that N concurrent upserts of the same key produce exactly one row."""
import asyncio

import pytest
from sqlalchemy import func, select


@pytest.mark.asyncio
@pytest.mark.postgres
async def test_concurrent_upserts_produce_one_row(engine, clean_database):
    Session = async_sessionmaker(engine, expire_on_commit=False)

    async def attempt(price: str) -> None:
        # Its own session, therefore its own connection and its own transaction.
        async with Session() as session, session.begin():
            await upsert_product(session, "SKU-1", "Widget", decimal.Decimal(price))

    await asyncio.gather(*(attempt(f"{n}.00") for n in range(1, 21)))

    async with Session() as session:
        count = await session.scalar(
            select(func.count()).select_from(Product).where(Product.sku == "SKU-1")
        )
    assert count == 1

Run the same test against the select-then-insert implementation and it fails, usually with an IntegrityError from one of the twenty tasks — which is the demonstration that makes the case better than any amount of explanation. Keeping that failing variant as a commented reference in the test file is worth more than a comment saying the race exists.

Two refinements make the test sharper. Increase the concurrency until it fails reliably against the naive implementation; twenty tasks is usually enough, but on a fast local machine the window can be narrow enough that five is not. And assert on the final value as well as the row count when the upsert is a counter: twenty increments of one must produce twenty, and an implementation that reads the value in Python and writes it back produces something smaller and non-deterministic — which is precisely the read-modify-write failure that a version counter, not an upsert, is the fix for.

Frequently Asked Questions

Why does on_conflict_do_update need the dialect-specific insert()?

Because ON CONFLICT is PostgreSQL syntax with PostgreSQL semantics, and the generic sqlalchemy.insert() has no such method. Importing from sqlalchemy.dialects.postgresql makes the dependency explicit at the import line, which is preferable to a construct that compiles differently on different backends.

Does DO NOTHING return the existing row?

No — it returns nothing at all, because no row was inserted or updated. If you need the row either way, either use DO UPDATE with a no-op assignment such as setting a column to itself, or follow the upsert with a SELECT. The no-op update is usually preferable because it stays a single statement.

Is an upsert slower than a plain insert?

Marginally, because the unique index must be probed — which a plain insert into a table with that index does anyway. The comparison that matters is against select-then-insert, where the upsert is faster by one full round trip and is additionally correct under concurrency.

Can I upsert many rows at once?

Yes. Pass a list of dicts to execute() with the same on_conflict_do_update statement and it becomes an executemany, with the conflict clause applied per row. The one thing to watch is duplicates within a single batch, which PostgreSQL rejects with "cannot affect row a second time" — deduplicate the batch in Python first.

Does an upsert avoid deadlocks?

It reduces them, because there is one statement rather than two and the lock is held for less time. It does not eliminate them: two transactions upserting the same two keys in opposite orders can still deadlock. Sorting the keys before upserting a batch removes that, and is worth doing by default.