Writing PostgreSQL ON CONFLICT DO UPDATE upserts in SQLAlchemy

Import insert from sqlalchemy.dialects.postgresql, name the conflict target with index_elements, and build set_ from stmt.excluded — referencing the model column instead of excluded compiles without complaint and writes back the value that was already there. This is the construct behind upserts and conflict resolution strategies, in detail.

Quick Answer

excluded versus the mapped column Two set_ mappings that look almost identical. Using the mapped column — Product.price — refers to the row currently stored, so on conflict the update writes the stored value back over itself and the upsert appears to do nothing. Using stmt.excluded.price refers to the row proposed by this INSERT, which is what you actually want to write. Both compile, both execute, and only one changes anything, so the failure shows up as data that never updates rather than as an exception. The same distinction is what makes an increment work: Product.stock plus stmt.excluded.stock deliberately combines both. set_={"price": Product.price} refers to the row already stored writes the stored value over itself compiles, runs, changes nothing no error anywhere set_={"price": stmt.excluded.price} refers to the row you proposed writes the incoming value what you meant combine both for an increment This is the highest-value review check on any upsert: every entry in set_ should mention excluded, unless it deliberately combines the stored value with the proposed one.

Before — select, then insert or update:

existing = await session.scalar(select(Product).where(Product.sku == sku))
if existing is None:
    session.add(Product(sku=sku, name=name, price=price))   # races
else:
    existing.name, existing.price = name, price
await session.commit()

After — one statement, no window:

from __future__ import annotations

import decimal

from sqlalchemy import func
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],
        set_={
            "name": stmt.excluded.name,       # excluded = the proposed row
            "price": stmt.excluded.price,
            "updated_at": func.now(),
        },
    ).returning(Product)
    return (await session.execute(stmt)).scalar_one()

Note the import: sqlalchemy.dialects.postgresql.insert, not sqlalchemy.insert. The generic construct has no on_conflict_do_update method, and the AttributeError you get from using it is the clearest error on this page.

Execution Context & Async Workflow Integration

stmt.excluded is available only after values() has been called, because it is derived from the columns the statement is inserting. Building the statement in the wrong order produces a confusing failure:

# Wrong — excluded does not know about columns that have not been named yet
stmt = insert(Product)
stmt = stmt.on_conflict_do_update(
    index_elements=[Product.sku],
    set_={"price": stmt.excluded.price},      # AttributeError, or silently wrong
).values(sku=sku, name=name, price=price)

# Right — values() first, then the conflict clause
stmt = insert(Product).values(sku=sku, name=name, price=price)
stmt = stmt.on_conflict_do_update(index_elements=[Product.sku],
                                  set_={"price": stmt.excluded.price})

Under async the statement is executed exactly like any other, and the one thing worth knowing is that RETURNING gets you the resulting row in the same round trip. That matters more than usual here, because the alternative — upsert, then select — reintroduces a second statement and with it a window in which another transaction can change the row again.

# Async — a batch upsert, still one round trip per chunk
async def upsert_products(session: AsyncSession, rows: list[dict]) -> None:
    stmt = insert(Product)
    stmt = stmt.on_conflict_do_update(
        index_elements=[Product.sku],
        set_={"name": stmt.excluded.name, "price": stmt.excluded.price},
    )
    # Sorting by the conflict key makes deadlocks between concurrent batches
    # impossible: every batch takes its row locks in the same order.
    await session.execute(stmt, sorted(rows, key=lambda r: r["sku"]))

Sorting the batch is cheap insurance. Two concurrent batches touching the same two keys in opposite orders deadlock, and PostgreSQL resolves that by killing one of them — a failure that appears only under load and is tedious to reproduce deliberately.

Four ways to name what you are conflicting on Four forms. index_elements listing the columns is the usual form and matches any unique index over exactly those columns. Adding index_where is required when the index is partial, because PostgreSQL cannot otherwise tell which index you mean and refuses. Passing constraint with a name targets a named unique or exclusion constraint directly, which is clearer when the constraint has a meaningful name and is the only way to target an exclusion constraint. And the primary key is just another unique index, so index_elements naming the primary-key column works with no special handling. index_elements=[cols] matches a unique index on exactly those columns the usual form index_elements + index_where required for a partial unique index must match the index predicate constraint="uq_name" targets a named constraint the only way to target an exclusion constraint the primary key just another unique index no special handling If PostgreSQL says there is no matching constraint, one of these four is wrong — most often a partial index whose predicate was not repeated in index_where.

Resolving Warnings, Errors & Common Mistakes

Error / WarningRoot CauseProduction Fix
AttributeError: 'Insert' object has no attribute 'on_conflict_do_update'Imported insert from sqlalchemy rather than from sqlalchemy.dialects.postgresql.Fix the import; the dialect construct is a different class.
ProgrammingError: there is no unique or exclusion constraint matching the ON CONFLICT specificationNo unique index over those columns, or the index is partial and index_where was omitted.Create the index, or repeat its predicate in index_where.
The upsert runs but nothing changesset_ referenced the mapped column instead of stmt.excluded.Every entry in set_ should mention excluded, unless it deliberately combines both.
CardinalityViolation: ON CONFLICT DO UPDATE command cannot affect row a second timeThe same conflict key appears twice in one executemany batch.Deduplicate the batch in Python, keeping whichever row should win.
NoResultFound from scalar_one()The where clause suppressed the update, so RETURNING produced nothing.Use scalar_one_or_none() and treat None as "skipped".
Deadlocks under concurrent batchesTwo batches took row locks in different orders.Sort each batch by the conflict key before executing.
A column silently keeps its old valueIt was omitted from set_.Build set_ from a comprehension over the columns you intend to overwrite, so omissions are deliberate.

Advanced: Conditional Updates and Partial Indexes

Two options turn ON CONFLICT from a blunt overwrite into something precise enough for real ingest paths.

The where clause decides whether the update happens at all, and is evaluated with access to both the stored row and the proposed one:

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},
    # Reading.taken_at is the STORED row; stmt.excluded.taken_at is the PROPOSED one.
    where=Reading.taken_at < stmt.excluded.taken_at,
)

That single line makes an ingest path robust against out-of-order delivery: a message that arrives late carrying an older reading is simply ignored, without the application having to read the current value first — which would reintroduce the race the upsert was chosen to avoid.

index_where names the predicate of a partial unique index, which is required whenever the uniqueness rule applies to a subset of rows:

# CREATE UNIQUE INDEX uq_active_subscription ON subscriptions (customer_id)
#   WHERE status = 'active';
stmt = insert(Subscription).values(customer_id=cid, plan=plan, status="active")
stmt = stmt.on_conflict_do_update(
    index_elements=[Subscription.customer_id],
    index_where=Subscription.status == "active",     # must match the index exactly
    set_={"plan": stmt.excluded.plan},
)

The predicate must match the index's own predicate, not merely be compatible with it — PostgreSQL performs a syntactic comparison and rejects anything it cannot prove equivalent. In practice that means keeping the model-level expression and the migration's DDL in step, which is one more reason to generate both from the same place.

Building set_ programmatically removes the omitted-column class of bug entirely:

UPSERT_COLUMNS = ("name", "price", "description", "updated_at")

stmt = stmt.on_conflict_do_update(
    index_elements=[Product.sku],
    set_={name: getattr(stmt.excluded, name) for name in UPSERT_COLUMNS},
)

Now adding a column to the upsert means adding it to one tuple, and it is impossible to reference the stored value by accident.

What the database does with the statement Five steps inside a single statement. PostgreSQL attempts the insert. If it violates the named unique index, the conflict path begins rather than an error being raised. The conflicting row is locked for the remainder of the transaction, which is what makes the operation atomic and also what makes a popular key a serialisation point in a long transaction. The optional where clause is evaluated against the stored row and the proposed one; if it is false, the statement affects zero rows and no update happens. Otherwise the set_ assignments are applied. RETURNING then yields whichever row resulted, or nothing at all if the where clause suppressed the update. attempt the INSERT the ordinary path no conflict → done uniqueness violation on the named index the conflict path, not an error this is why the index must exist lock the conflicting row held until COMMIT a popular key serialises here evaluate the where clause false → zero rows affected and RETURNING yields nothing apply set_ and RETURNING the resulting row comes back The where clause suppressing the update also suppresses RETURNING, so a caller using scalar_one() will raise NoResultFound on exactly the rows it was meant to skip.

Reading the Result: RETURNING and Affected Rows

An upsert has three possible outcomes — inserted, updated, or skipped by the where clause — and callers frequently need to know which. RETURNING gives you the row; distinguishing the branches takes one more column.

from sqlalchemy import literal_column

stmt = insert(Product).values(sku=sku, name=name, price=price)
stmt = stmt.on_conflict_do_update(
    index_elements=[Product.sku],
    set_={"name": stmt.excluded.name, "price": stmt.excluded.price},
).returning(
    Product.id,
    # xmax is 0 for a freshly inserted row and non-zero for an updated one.
    (literal_column("xmax") == 0).label("inserted"),
)

row = (await session.execute(stmt)).one_or_none()
if row is None:
    outcome = "skipped"                     # only possible with a where clause
elif row.inserted:
    outcome = "inserted"
else:
    outcome = "updated"

The xmax trick is PostgreSQL-specific and relies on an implementation detail of MVCC, so it is worth a comment wherever it appears. It is reliable in practice and is the only way to distinguish the branches without a second statement — but if the distinction is load-bearing for business logic rather than for metrics, a guarded insert with DO NOTHING expresses the intent more honestly, as the parent guide argues.

result.rowcount is the cheaper signal when you do not need the row itself: it is 1 when the statement inserted or updated and 0 when the where clause suppressed the update. That is enough for a metric counting skipped out-of-order messages, which is usually the number worth watching on an ingest path — a sudden rise means upstream ordering has changed, and that is worth knowing before the data does something surprising.

A Review Checklist for Any Upsert

Five questions catch nearly every upsert defect, and all of them can be answered by reading the statement rather than by running it.

Does every entry in set_ mention excluded? If one references the mapped column instead, that column is being written back over itself and the upsert silently does nothing for it. The exception is a deliberate combination such as Product.stock + stmt.excluded.stock, which mentions both.

Does a unique index actually exist over index_elements? PostgreSQL raises if not, which is the good case — but a partial index needs its predicate repeated in index_where, and forgetting that produces the same error for a reason that is less obvious.

Is every column that should change present in set_? A column omitted keeps its stored value, and for updated_at in particular the row then looks untouched. Building set_ from a named tuple of columns removes this class of mistake entirely.

Is the batch deduplicated and sorted? Duplicates within one statement are rejected outright; unsorted batches deadlock against each other under concurrency. Both are properties of the caller, not of the statement.

Does the caller handle the skipped case? A where clause that suppresses the update also suppresses RETURNING, so scalar_one() raises where scalar_one_or_none() would return None.

Frequently Asked Questions

Can I use ON CONFLICT with the ORM rather than Core?

The statement is a Core construct, but it can return ORM entities with .returning(Model) and be executed through an AsyncSession. What it does not do is participate in the unit of work: no ORM events fire and the identity map is not updated, so an instance loaded earlier in the same session keeps its old values until refreshed.

Why must values() come before on_conflict_do_update()?

stmt.excluded is built from the columns the insert names, so it does not exist until values() has been called. Calling the conflict clause first either raises AttributeError or, worse, produces a statement referencing columns the insert does not set.

What happens if two rows in one batch have the same key?

PostgreSQL raises CardinalityViolation: ON CONFLICT DO UPDATE command cannot affect row a second time. It will not silently pick one. Deduplicate in Python before executing, keeping whichever row should win — usually the last by timestamp.

Does ON CONFLICT work on SQLite for tests?

Yes, through sqlalchemy.dialects.sqlite.insert, with on_conflict_do_update and on_conflict_do_nothing both available. The syntax is close enough that the code paths are exercised, but SQLite serialises writers so it cannot demonstrate the concurrency property that motivated the upsert.