Module 7: Bulk Operations

Atomic upserts with `ON CONFLICT`

Next →

"Upsert" = INSERT or UPDATE. The typical case: you import data where some records already exist. Without upsert, your options are:

  1. SELECT then INSERT/UPDATE: race condition between the check and the write.
  2. Try INSERT, catch the error, UPDATE: ugly, two round-trips, possible deadlocks.
  3. Truncate + re-insert: loses data, not idempotent.

PostgreSQL has the canonical answer since 9.5: INSERT ... ON CONFLICT. It's atomic (no race), idempotent (re-running the same INSERT causes no errors), and bulk-friendly (combines with executemany and COPY-to-temp).

In this capsule you'll learn the exact syntax, the three variants (DO NOTHING, DO UPDATE, partial), how to use EXCLUDED correctly, and the most common anti-patterns.

The basic syntax

INSERT INTO tasks (id, title, status)
VALUES (1, 'Task 1', 'pending')
ON CONFLICT (id) DO UPDATE SET
    title = EXCLUDED.title,
    status = EXCLUDED.status,
    updated_at = NOW();

Reading it step by step:

  • INSERT INTO ... VALUES: try to insert.
  • ON CONFLICT (id): if there's a conflict on the id column (because it already exists).
  • DO UPDATE SET ...: instead of failing, do this UPDATE.
  • EXCLUDED.title: refers to the value that was going to be inserted. PostgreSQL creates a pseudo-table EXCLUDED holding the proposed row.

EXCLUDED is the key keyword. It lets you refer to the "new" value in the UPDATE, distinguishing it from the existing row.

Three variants of ON CONFLICT

Variant 1: DO NOTHING (skip duplicates)

INSERT INTO tasks (id, title) VALUES (1, 'Task 1')
ON CONFLICT (id) DO NOTHING;

If id=1 already exists, it does nothing. Useful for idempotent imports where you don't want to overwrite.

Typical use case: an event log where duplicates are possible but the first one wins.

Variant 2: DO UPDATE (overwrite)

INSERT INTO tasks (id, title, status) VALUES (1, 'Updated', 'completed')
ON CONFLICT (id) DO UPDATE SET
    title = EXCLUDED.title,
    status = EXCLUDED.status;

If id=1 exists, it updates title and status with the new values.

Typical use case: syncing from an external source. The latest value always wins.

Variant 3: partial DO UPDATE (with WHERE)

INSERT INTO tasks (id, title, status, version) VALUES (1, 'Updated', 'completed', 5)
ON CONFLICT (id) DO UPDATE SET
    title = EXCLUDED.title,
    status = EXCLUDED.status,
    version = EXCLUDED.version
WHERE tasks.version < EXCLUDED.version;  -- Only update if the new version is newer

Conditional update. The row is updated only if the condition holds. Useful for "last-write-wins" with timestamps or version numbers.

Conflict target: which column(s)

ON CONFLICT (cols) requires the columns to have a UNIQUE or PRIMARY KEY constraint:

-- ✅ id is PK
INSERT INTO tasks (id, ...) VALUES (...) ON CONFLICT (id) DO UPDATE ...;

-- ✅ email is UNIQUE
INSERT INTO users (email, ...) VALUES (...) ON CONFLICT (email) DO UPDATE ...;

-- ❌ status is not UNIQUE — error
INSERT INTO tasks (...) VALUES (...) ON CONFLICT (status) DO UPDATE ...;
-- ERROR: there is no unique or exclusion constraint matching the ON CONFLICT specification

Conflict on a composite key

-- Composite UNIQUE constraint
ALTER TABLE order_items ADD CONSTRAINT uq_order_book UNIQUE (order_id, book_id);

INSERT INTO order_items (order_id, book_id, quantity) VALUES (1, 42, 3)
ON CONFLICT (order_id, book_id) DO UPDATE SET
    quantity = order_items.quantity + EXCLUDED.quantity;

Useful for "add if it doesn't exist, sum the quantity if it does".

Conflict on a constraint name

-- If you have several UNIQUE constraints, you can specify which one
INSERT INTO ...
ON CONFLICT ON CONSTRAINT my_unique_constraint DO UPDATE ...;

Implementation in SQLAlchemy 2.0

Single row upsert

from sqlalchemy.dialects.postgresql import insert as pg_insert


async def upsert_task(session, task_data: dict):
    stmt = pg_insert(Task).values(**task_data)
    stmt = stmt.on_conflict_do_update(
        index_elements=["id"],  # ON CONFLICT (id)
        set_={
            "title": stmt.excluded.title,
            "status": stmt.excluded.status,
        }
    )

    await session.execute(stmt)
    await session.commit()

stmt.excluded is the translation of EXCLUDED.col in SQL.

Bulk upsert with ON CONFLICT

async def bulk_upsert_tasks(session, records: list[dict]):
    stmt = pg_insert(Task).values(records)
    stmt = stmt.on_conflict_do_update(
        index_elements=["id"],
        set_={
            "title": stmt.excluded.title,
            "status": stmt.excluded.status,
        }
    )

    await session.execute(stmt)
    await session.commit()

pg_insert(Task).values(list_of_dicts) runs a multi-row INSERT with ON CONFLICT on each row. Atomic.

DO NOTHING

stmt = pg_insert(Task).values(records)
stmt = stmt.on_conflict_do_nothing(index_elements=["id"])

await session.execute(stmt)

Conditional UPDATE

from sqlalchemy import text

stmt = pg_insert(Task).values(records)
stmt = stmt.on_conflict_do_update(
    index_elements=["id"],
    set_={
        "title": stmt.excluded.title,
        "version": stmt.excluded.version,
    },
    where=Task.version < stmt.excluded.version,  # Conditional
)

Complete pattern in an endpoint

from datetime import datetime, timezone
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.dialects.postgresql import insert as pg_insert


router = APIRouter()


@router.post("/tasks/sync")
async def sync_tasks(
    tasks: list[TaskSyncData],
    db: AsyncSession = Depends(get_db),
):
    """Sync tasks from an external source (idempotent).

    - If the task doesn't exist (by external_id), insert it.
    - If it exists, update it with the new values.
    """
    records = [
        {
            "external_id": t.external_id,
            "title": t.title,
            "status": t.status,
            "synced_at": datetime.now(timezone.utc),
        }
        for t in tasks
    ]

    stmt = pg_insert(Task).values(records)
    stmt = stmt.on_conflict_do_update(
        index_elements=["external_id"],
        set_={
            "title": stmt.excluded.title,
            "status": stmt.excluded.status,
            "synced_at": stmt.excluded.synced_at,
        }
    )

    await db.execute(stmt)
    await db.commit()

    return {"synced": len(records)}

The client can call this endpoint multiple times with the same data — it always reaches the same final state. Perfect idempotency.

Common use cases

1. Sync from an external source

# Cron job that syncs with an external system
async def sync_from_external():
    external_data = fetch_from_external_api()
    await bulk_upsert_tasks(external_data)
    # Re-running it is safe

2. Incremental counter

-- Tracking page views: increment the counter or create it if it doesn't exist
INSERT INTO page_views (page_id, view_count, last_viewed)
VALUES ($1, 1, NOW())
ON CONFLICT (page_id) DO UPDATE SET
    view_count = page_views.view_count + 1,
    last_viewed = NOW();

page_views.view_count + 1 (reference to the original table) is important. EXCLUDED.view_count + 1 would be wrong (always 1+1=2).

3. Idempotency for deduplicated messages

-- Dedupe messages by message_id
INSERT INTO messages (message_id, content, created_at) VALUES ($1, $2, $3)
ON CONFLICT (message_id) DO NOTHING
RETURNING id;

If the message already exists, it returns no row. The app knows it already processed it.

4. Last-write-wins with version

INSERT INTO documents (id, content, version) VALUES ($1, $2, $3)
ON CONFLICT (id) DO UPDATE SET
    content = EXCLUDED.content,
    version = EXCLUDED.version
WHERE documents.version < EXCLUDED.version;

Only updates if the new version is higher. Race-safe.

Referencing the current table vs EXCLUDED

Common confusion: when to use tasks.col vs EXCLUDED.col:

INSERT INTO tasks (id, view_count) VALUES (1, 1)
ON CONFLICT (id) DO UPDATE SET
    view_count = tasks.view_count + 1;     -- current value + 1
    -- vs
    view_count = EXCLUDED.view_count + 1;  -- the value that was going to be inserted (1) + 1 = always 2
  • tasks.col = the current value of the existing row.
  • EXCLUDED.col = the value you were going to insert.

For counters: tasks.view_count + 1. For overwrite: EXCLUDED.col. For incremental sum: tasks.col + EXCLUDED.col.

Limitations

ON CONFLICT doesn't work with direct COPY

-- ❌ There's no COPY ... ON CONFLICT
COPY tasks FROM STDIN ON CONFLICT (id) DO UPDATE SET ...;  -- Syntax error

For bulk upserts, the pattern is COPY into a temp table + INSERT...ON CONFLICT from the temp (capsule 06).

Only one conflict target column at a time

You can't have multiple ON CONFLICT clauses with different columns in a single query. Only one target.

RETURNING can be confusing

INSERT ... ON CONFLICT DO UPDATE ... RETURNING id, (xmax = 0) AS inserted;

xmax = 0 is a trick to tell whether it was an INSERT (true) or an UPDATE (false). Useful for reporting stats.

Pitfalls and common mistakes

1. ON CONFLICT (col) on a column without a UNIQUE constraint.

Error: "there is no unique or exclusion constraint matching the ON CONFLICT specification". Check that you have UNIQUE or PRIMARY KEY on that column.

2. Confusing EXCLUDED with NEW.

NEW only exists in PL/pgSQL triggers. In ON CONFLICT, it's EXCLUDED.

3. EXCLUDED for counters.

-- ❌ Wrong: always sums 1+1=2 (or the initial value)
ON CONFLICT (id) DO UPDATE SET count = EXCLUDED.count + 1;

-- ✅ Right: increments the current value
ON CONFLICT (id) DO UPDATE SET count = tasks.count + 1;

4. Forgetting columns in SET.

If your UPDATE only sets title, other columns keep their existing value. For a "full overwrite", list every column.

5. WHERE on ON CONFLICT with an incorrect condition.

-- If the condition isn't met, it does NOT fail — it simply doesn't update.
INSERT ... ON CONFLICT (id) DO UPDATE SET ... WHERE FALSE;

This is like DO NOTHING. Be careful with the WHERE logic.

6. ON CONFLICT DO UPDATE with PL/pgSQL triggers.

Triggers fire on the underlying INSERT and on the resulting UPDATE. If your trigger has side effects (logging, audit), it fires twice. Check.

7. Expected vs real performance.

ON CONFLICT adds per-row overhead (checking the unique constraint). On very large bulk loads, it can be ~10-30% slower than a plain INSERT. For 100k rows: 4s without ON CONFLICT, 5-6s with it.

8. bulk_insert (SQLAlchemy ORM bulk) does NOT support ON CONFLICT.

session.execute(insert(Model), [...]) doesn't support on_conflict_do_update easily. Use pg_insert from sqlalchemy.dialects.postgresql, which does.

Exercise: implement a sync endpoint

Setup:

class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    external_id: Mapped[str] = mapped_column(String(100), unique=True)
    title: Mapped[str] = mapped_column(String(200))
    status: Mapped[str] = mapped_column(String(50))
    synced_at: Mapped[datetime] = mapped_column(server_default=func.now())

external_id is UNIQUE — the conflict target.

Step 1: implement POST /tasks/sync with upsert.

from sqlalchemy.dialects.postgresql import insert as pg_insert


@router.post("/tasks/sync")
async def sync_tasks(tasks: list[TaskSync], db: AsyncSession = Depends(get_db)):
    records = [t.model_dump() for t in tasks]
    stmt = pg_insert(Task).values(records)
    stmt = stmt.on_conflict_do_update(
        index_elements=["external_id"],
        set_={
            "title": stmt.excluded.title,
            "status": stmt.excluded.status,
            "synced_at": stmt.excluded.synced_at,
        }
    )
    await db.execute(stmt)
    await db.commit()
    return {"synced": len(records)}

Step 2: test idempotency.

# Run twice — the second must not fail
data = [{"external_id": "ext-1", "title": "T1", "status": "pending"}]
await sync_tasks(data, db)
await sync_tasks(data, db)  # No error, same final state

Step 3: test that values get updated.

# First run: insert
await sync_tasks([{"external_id": "ext-1", "title": "Original", "status": "pending"}], db)

# Second: update
await sync_tasks([{"external_id": "ext-1", "title": "Updated", "status": "completed"}], db)

# Verify
task = await db.scalar(select(Task).where(Task.external_id == "ext-1"))
assert task.title == "Updated"
assert task.status == "completed"

Step 4: add reporting (how many inserts vs updates).

stmt = pg_insert(Task).values(records).returning(
    Task.id,
    text("(xmax = 0) AS inserted")  # true if insert, false if update
)
stmt = stmt.on_conflict_do_update(...)

result = await db.execute(stmt)
rows = result.all()

inserted = sum(1 for r in rows if r.inserted)
updated = len(rows) - inserted

return {"inserted": inserted, "updated": updated}

Step 5: add a conditional update (only if the version is higher).

stmt = stmt.on_conflict_do_update(
    index_elements=["external_id"],
    set_={...},
    where=Task.version < stmt.excluded.version,
)
See discussion

Step 1 — implementation: works out-of-the-box.

Step 2 — idempotency:

Run the same payload 100 times. The final state is the same. No errors.

Step 3 — update:

Existing tasks get updated. synced_at too, which gives you auditability.

Step 4 — reporting:

xmax = 0 is a PostgreSQL trick to identify whether it was an INSERT vs an UPDATE. Useful for returning stats to the client.

Step 5 — conditional:

If a task has version=5 locally and you receive data with version=3 (out-of-order), it doesn't get updated. Protection against delivery race conditions.

Key takeaways:

  1. ON CONFLICT solves idempotency at the SQL level, atomically.
  2. Combined with bulk insert, it gives you re-runnable imports.
  3. EXCLUDED vs table.col matters a lot — tell them apart clearly.
  4. A conditional WHERE protects against out-of-order updates.

Summary and next step

What you learned:

  • INSERT ... ON CONFLICT (col): PostgreSQL's native atomic upsert.
  • 3 variants: DO NOTHING (skip), DO UPDATE (overwrite), DO UPDATE WHERE (conditional).
  • EXCLUDED = the value that was going to be inserted. table.col = the current value.
  • Conflict target: requires a UNIQUE/PK constraint on the column(s).
  • SQLAlchemy 2.0: use pg_insert from sqlalchemy.dialects.postgresql, not the generic insert.
  • Bulk upsert works with a list of dicts.
  • Limitation: direct COPY doesn't support ON CONFLICT — it needs the temp-table pattern (capsule 06).

Before moving on, you should be able to:

  • Write SQL INSERT ... ON CONFLICT (col) DO UPDATE SET col = EXCLUDED.col.
  • Implement an upsert with pg_insert in SQLAlchemy 2.0.
  • Tell when to use EXCLUDED vs table.col.
  • Apply WHERE on ON CONFLICT for conditional updates.

In the next capsule we combine COPY with ON CONFLICT — the canonical pattern for large bulk upserts. Since COPY doesn't support ON CONFLICT directly, the pattern is: COPY into a temp table → INSERT...ON CONFLICT from the temp. You'll learn the complete setup and why this pattern is the one any serious PostgreSQL ETL uses.

Resources

  1. PostgreSQL Docs — INSERT ... ON CONFLICT — official reference.
  2. SQLAlchemy 2.0 — pg_insert.on_conflict_do_update — reference.
  3. PostgreSQL Wiki — UPSERT — history and cases.
  4. Brandur Leach — Postgres queries patterns — patterns with ON CONFLICT.
  5. Citus Data — UPSERT performance — cases at scale.
  6. Heap — Idempotent ETL — ON CONFLICT in a pipeline.

Capsule 05 of 08 — Module 7 — SQL Patterns for Production APIs Guide

← Previous`bulk_insert_mappings` and its limitationsNext →Bulk upserts with a temp table: the canonical pattern