Bulk upserting rows with insert().on_conflict_do_update()

Pass a list of dicts to session.execute() with an on_conflict_do_update() statement and SQLAlchemy issues one executemany per chunk with the conflict clause applied per row — but the batch must be deduplicated on the conflict key first, or PostgreSQL rejects it, and sorted, or concurrent batches deadlock. This is the bulk form of upserts and conflict resolution strategies.

Quick Answer

Before — a loop of single upserts:

for row in rows:                      # one round trip each: 1 000 000 round trips
    stmt = insert(Product).values(**row)
    stmt = stmt.on_conflict_do_update(index_elements=[Product.sku],
                                      set_={"price": stmt.excluded.price})
    await session.execute(stmt)

After — one statement per chunk:

from __future__ import annotations

from collections.abc import Iterable, Iterator
from itertools import islice

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

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


def _chunks(rows: Iterable[dict], size: int) -> Iterator[list[dict]]:
    it = iter(rows)
    while chunk := list(islice(it, size)):
        yield chunk


async def bulk_upsert_products(session: AsyncSession, rows: Iterable[dict],
                               chunk_size: int = 2_000) -> int:
    stmt = insert(Product)
    stmt = stmt.on_conflict_do_update(
        index_elements=[Product.sku],
        set_={name: getattr(stmt.excluded, name) for name in UPSERT_COLUMNS},
    )

    affected = 0
    for chunk in _chunks(rows, chunk_size):
        # Mandatory: one row per key, or PostgreSQL rejects the whole statement.
        deduped = {row["sku"]: row for row in chunk}
        # Sorted: concurrent batches then take row locks in the same order.
        batch = [deduped[key] for key in sorted(deduped)]
        result = await session.execute(stmt, batch)
        affected += result.rowcount or 0
    return affected

The dict comprehension keeps the last row for each key, which is the usual intent when rows arrive in arrival order. Whichever rule you want, make it explicit — silently keeping the first is a different decision.

Execution Context & Async Workflow Integration

Passing a list of parameter dictionaries as the second argument to execute() is what switches SQLAlchemy into executemany mode. The statement is compiled once and the parameter sets are sent together, which is where nearly all of the speed-up comes from — the same mechanism described in batch inserting millions of rows with SQLAlchemy Core execute, with the conflict clause applied per row.

Adding RETURNING changes the strategy. SQLAlchemy's insertmanyvalues machinery rewrites the operation into batched VALUES statements so results can come back, which is slower than a plain executemany but far faster than per-row execution.

# Async — bulk upsert that needs the resulting ids back
async def bulk_upsert_returning(session: AsyncSession, rows: list[dict]) -> list[int]:
    stmt = insert(Product)
    stmt = stmt.on_conflict_do_update(
        index_elements=[Product.sku],
        set_={"price": stmt.excluded.price},
    ).returning(Product.id)

    deduped = {row["sku"]: row for row in rows}
    batch = [deduped[key] for key in sorted(deduped)]
    result = await session.execute(stmt, batch)
    return list(result.scalars())

Two constraints come with the transaction shape. Running every chunk inside one session.begin() makes the whole load atomic, which is usually what an import wants — but it also means every row lock is held until the end, so a very long load blocks other writers on those keys for its duration. Committing per chunk releases the locks progressively at the cost of a partially applied import if the job fails halfway.

Which to choose is a domain question rather than a technical one: a nightly full refresh should be atomic, while a continuous ingest stream is better off committing per chunk and being resumable. For the resumable case, recording the last successfully processed key alongside the commit turns a failed run into a restart rather than a redo.

Resolving Warnings, Errors & Common Mistakes

Advanced: Deduplication Rules and Staging Tables

Deduplicating a batch requires deciding which duplicate wins, and the answer is domain-specific often enough that it deserves to be explicit rather than incidental.

from __future__ import annotations

import datetime
from collections.abc import Iterable


def dedupe_latest(rows: Iterable[dict], key: str, timestamp: str) -> list[dict]:
    """Keep the row with the newest timestamp per key — correct for out-of-order feeds."""
    best: dict[str, dict] = {}
    for row in rows:
        current = best.get(row[key])
        if current is None or row[timestamp] > current[timestamp]:
            best[row[key]] = row
    return [best[k] for k in sorted(best)]


def dedupe_merged(rows: Iterable[dict], key: str) -> list[dict]:
    """Combine partial updates for the same key, later fields winning."""
    merged: dict[str, dict] = {}
    for row in rows:
        merged.setdefault(row[key], {}).update(
            {k: v for k, v in row.items() if v is not None}
        )
    return [merged[k] for k in sorted(merged)]

The second form matters for feeds where each message carries only the fields that changed: keeping the last message outright would null out fields an earlier message had set, whereas merging preserves them.

For very large loads, a staging table is frequently faster than a large upsert, because it separates the fast part from the expensive part:

# 1. COPY into an unlogged staging table — the fastest write PostgreSQL offers
await session.execute(text(
    "CREATE UNLOGGED TABLE IF NOT EXISTS products_stage "
    "(LIKE products INCLUDING DEFAULTS)"
))
await session.execute(text("TRUNCATE products_stage"))
# ... COPY the rows in ...

# 2. One set-based merge, with the deduplication done in SQL
await session.execute(text("""
    INSERT INTO products (sku, name, price, updated_at)
    SELECT DISTINCT ON (sku) sku, name, price, updated_at
    FROM   products_stage
    ORDER  BY sku, updated_at DESC
    ON CONFLICT (sku) DO UPDATE
       SET name = EXCLUDED.name,
           price = EXCLUDED.price,
           updated_at = EXCLUDED.updated_at
"""))

DISTINCT ON (sku) ... ORDER BY sku, updated_at DESC is PostgreSQL's deduplication idiom and does in one pass what the Python function above does row by row. The staging table being UNLOGGED means it is not written to the write-ahead log, which roughly halves the cost of loading it — acceptable precisely because losing it on a crash costs nothing.

This shape is worth the extra table above roughly a million rows, or whenever the source is a file that can be COPY-ed directly rather than parsed in Python.

Sizing the Chunk and Measuring the Result

Chunk size is the one parameter people tune first and understand last. The constraint that actually binds is the parameter count rather than the row count: PostgreSQL's protocol limits a single bind message to 65 535 parameters, so a table with twelve upserted columns caps out at roughly 5 400 rows per statement regardless of how much memory you have.

MAX_BIND_PARAMS = 65_535


def safe_chunk_size(columns: int, requested: int = 5_000) -> int:
    """Never exceed the protocol limit, whatever was asked for."""
    return max(1, min(requested, MAX_BIND_PARAMS // max(columns, 1)))

Within that ceiling, the curve flattens quickly. Going from one row to five hundred is transformative; five hundred to five thousand is a further improvement of a few tens of percent; beyond that the gains are noise and the risk of hitting a proxy statement-size limit rises. Two to five thousand is the range worth starting in.

Measure with the discipline in benchmarking Core executemany bulk insert performance: a production-shaped table with its real indexes, the first run discarded, and the transaction boundary held constant between variants. An upsert benchmarked against an empty table tells you almost nothing, because the conflict path — which is the expensive one — is never taken.

The number worth reporting is rows per second with the expected conflict ratio. A load where ninety per cent of rows already exist behaves very differently from one where ten per cent do, because the update path writes a new row version and leaves the old one for vacuum. On a table that is upserted continuously, that dead-tuple churn is frequently the real cost, and it shows up as autovacuum activity rather than in the load's own timings — which is why watching table bloat after adopting a bulk upsert is worth as much as timing the load itself.

Frequently Asked Questions

Related

• Upserts and Conflict Resolution Strategies — The parent guide: choosing the conflict strategy this bulk form applies per row.
• Writing PostgreSQL ON CONFLICT DO UPDATE upserts in SQLAlchemy — The single-row construct, including excluded and conditional updates.
• Batch inserting millions of rows with SQLAlchemy Core execute — The chunking and executemany mechanics this builds on.
• Benchmarking Core executemany bulk insert performance — Measuring a bulk upsert honestly, with a realistic conflict ratio.