Add bulk_upsert() and make bulk_create insert-only

.claude/rules/plain-postgres.md

@@ -111,11 +111,11 @@ Use `Model.query` to build querysets (e.g., `User.query.filter(is_active=True)`)
 - Use `.annotate(Count(...))` instead of calling `.count()` per row
 - Fetch all data in the view — templates should never trigger queries
 - Use `.exists()` not `.count() > 0`, `.count()` not `len(qs)`
-- Use `bulk_create`/`bulk_update` for batch ops, `.update()`/`.delete()` for mass ops
+- Use `bulk_create`/`bulk_update` for batch ops, `bulk_upsert` for atomic insert-or-update, `.update()`/`.delete()` for mass ops
 - Use `.values_list()` when you only need specific columns
 - Wrap multi-step writes in `transaction.atomic()`
 - Instance writes are `obj.create()` (always INSERT) and `obj.update()` (always UPDATE; `update(fields=[...])` limits the columns) — there is no `save()`, `force_insert`, or `force_update`. Constructing an instance then `create()`-ing it inserts; a hand-set `id` that collides raises `IntegrityError`.
-- `create()`/`update()` raise `ValidationError` (not raw `psycopg.IntegrityError`) on a declared unique/check constraint violation or a foreign key pointing at a missing row, even a raced one — the DB enforces it, so inside an open `transaction.atomic()` the violation aborts the transaction (wrap the write in its own `atomic()` to catch and keep using the transaction). Set-based writes (`QuerySet.update()`/`bulk_create()`) and `delete()` blocked by `RESTRICT` raise raw `psycopg.IntegrityError`. Retrying on conflict? `except (psycopg.IntegrityError, ValidationError)`, or `bulk_create(..., update_conflicts=True)`
+- `create()`/`update()` raise `ValidationError` (not raw `psycopg.IntegrityError`) on a declared unique/check constraint violation or a foreign key pointing at a missing row, even a raced one — the DB enforces it, so inside an open `transaction.atomic()` the violation aborts the transaction (wrap the write in its own `atomic()` to catch and keep using the transaction). Set-based writes (`QuerySet.update()`/`bulk_create()`) and `delete()` blocked by `RESTRICT` raise raw `psycopg.IntegrityError`. Retrying on conflict? `except (psycopg.IntegrityError, ValidationError)`, or `bulk_upsert(objs, update_fields=[...], unique_fields=[...])` for an atomic insert-or-update
 - Always paginate list queries — unbounded querysets get slower as data grows
 
 Run `uv run plain docs postgres` for full patterns with code examples.

plain-cache/plain/cache/core.py

@@ -106,12 +106,12 @@ def set_many(
         if not mapping:
             return
 
-        # bulk_create fires pre_save, so updated_at's update_now stamps a fresh
-        # now() at write time on its own. created_at (no update_now) would
-        # otherwise fall to its DB default, evaluated a hair later -- leaving a
-        # brand-new row with updated_at < created_at. Stamp created_at from an
-        # up-front `now` so created_at <= updated_at; it's omitted from
-        # update_fields, so it's preserved on conflict.
+        # bulk_upsert fires pre_save and refreshes update_now columns on the
+        # conflict path, so updated_at looks after itself. created_at (no
+        # update_now) would otherwise fall to its DB default, evaluated a hair
+        # later -- leaving a brand-new row with updated_at < created_at. Stamp
+        # created_at from an up-front `now` so created_at <= updated_at; being
+        # DB-owned it can't be named in update_fields, so it survives conflicts.
         now = timezone.now()
         expires_at = _coerce_expiration(expiration, now=now)
         items = []
@@ -122,11 +122,11 @@ def set_many(
             # construction so created_at <= updated_at (see comment above).
             item.created_at = now
             items.append(item)
-        self._model.query.bulk_create(
+        model = self._model
+        model.query.bulk_upsert(
             items,
-            update_conflicts=True,
-            update_fields=["value", "expires_at", "updated_at"],
-            unique_fields=["key"],
+            update_fields=[model.value, model.expires_at],
+            unique_fields=[model.key],
         )
 
     def get_or_set(
@@ -241,7 +241,7 @@ def touch(self, key: str, *, expiration: Expiration = None) -> bool:
         """
         # QuerySet.update() issues a direct SQL UPDATE and does NOT fire pre_save,
         # so updated_at's update_now won't bump on its own -- stamp it by hand.
-        # (set_many() relies on pre_save instead, since bulk_create does fire it.)
+        # (set_many() relies on pre_save instead, since bulk_upsert does fire it.)
         now = timezone.now()
         updated = (
             self._model.query.live()

plain-cache/tests/internal/test_set_timestamps.py

@@ -1,6 +1,6 @@
 """Timestamp invariants for the set-based write paths.
 
-bulk_create fires pre_save (so updated_at's update_now bumps on its own) while
+bulk_upsert fires pre_save (so updated_at's update_now bumps on its own) while
 QuerySet.update() does not -- see core.py for why set_many stamps created_at and
 touch stamps updated_at. These pin the observable invariants those choices buy.
 """

plain-postgres/plain/postgres/README.md

@@ -464,6 +464,64 @@ for name in names:
 Tag.query.bulk_create([Tag(name=name) for name in names])
 ```
 
+`bulk_create` is insert-only. To insert new rows and update the ones that
+already exist in a single statement, use `bulk_upsert` (below).
+
+#### Use `bulk_upsert` to insert-or-update in one statement
+
+`bulk_upsert(objs, *, update_fields, unique_fields, batch_size=None)` issues one
+`INSERT ... ON CONFLICT (unique_fields) DO UPDATE SET ... RETURNING` per batch.
+Rows that don't exist yet are inserted; rows that collide on `unique_fields` have
+their `update_fields` overwritten. You get back the objects you passed in, in the
+order you passed them (a new list — `objs` itself is never reordered), each with
+its DB-generated fields (primary key, DB defaults) populated.
+
+```python
+# Insert new items, refresh `value`/`expires_at` on any existing key.
+CachedItem.query.bulk_upsert(
+    [CachedItem(key=k, value=v, expires_at=exp) for k, v in items],
+    update_fields=[CachedItem.value, CachedItem.expires_at],
+    unique_fields=[CachedItem.key],
+)
+```
+
+- `update_fields` and `unique_fields` take field references (`Model.field`), not
+  strings. A foreign key is named by the relation itself — `Model.tenant`, which
+  resolves to the `tenant_id` column. (This is the one write API that takes
+  `Model.fk`. `returning()` refuses it, because there it would be ambiguous with
+  asking for the whole related object; here a column list can only mean the
+  column.)
+- `unique_fields` must name the **primary key** or a `UniqueConstraint` declared
+  on the model (no condition, no expressions) — this is the conflict target. A
+  unique `Index` is not enough; declare a `UniqueConstraint`.
+- `update_fields` must be concrete, non-primary-key, must not name the same
+  column twice (Postgres assigns each column once per statement), and must not
+  overlap `unique_fields`. A column the database fills in (`create_now`,
+  `generate=True`, `RandomStringField`) can't be named either — the update would
+  overwrite the stored value with a freshly evaluated default.
+- Every object must have a non-null value for every unique field. `NULL` never
+  conflicts in Postgres, so it can't be upserted. A database-generated column
+  (`create_now`, `generate=True`, `RandomStringField`) can't be a unique field
+  either — your objects never hold its value, so it could never conflict. Nor
+  can an `update_now=True` column, which is stamped again on every write.
+- **Two objects with the same unique key in one batch raise `ValueError`.**
+  Postgres won't touch a row twice in one statement. Split across batches it's
+  allowed — the first inserts, the second updates, and the later write wins.
+- **`update_now=True` columns are refreshed on a conflict automatically.** You
+  don't name them in `update_fields`; a row that gets updated gets a fresh
+  stamp, and the object handed back carries the same one.
+- **An `id` you set is kept on the insert path; on a conflict the stored row
+  wins.** A new row is written with the `id` you gave it. A conflicting one
+  already has an `id`, and that is the one hydrated back onto your object — the
+  row in the table is the truth. An `id` that collides with a _different_ row
+  raises `psycopg.errors.UniqueViolation`, like any set-based write.
+- Every object is sorted by its conflict key before anything is sent, so
+  concurrent `bulk_upsert` calls over overlapping keys lock rows in the same
+  order and can't deadlock each other. Returned rows are mapped onto the objects
+  by position, exactly as `bulk_create` does.
+- Like `bulk_create`, the write is against the table: a filter on the queryset
+  you call it from doesn't narrow or exclude anything.
+
 #### Use queryset `.update()` / `.delete()` for mass operations
 
 ```python
@@ -529,7 +587,7 @@ for row in deleted:
 - **`returning(Model.field, ...)`** returns a list of dicts with only those columns. Pass field references (`Model.field`), not strings; a many-to-many field or one from another model raises an error at the `returning()` call.
 - **A foreign key can't be named here.** At class level `Model.fk` is the relation — that is what lets `where()` traverse it, as in `Child.parent.name.equals(...)` — not its column, so `returning(Child.parent)` raises `FieldError`. Foreign key columns come back through no-argument `returning()`, which hands you whole instances.
 - Without `returning()`, `update()`/`delete()` return an `int` as before.
-- `returning()` only applies to `update()` and `delete()`. Any other write on the same queryset — `create()`, `bulk_create()`, `bulk_update()`, `get_or_create()`, `update_or_create()` — raises `TypeError` rather than quietly dropping it.
+- `returning()` only applies to `update()` and `delete()`. Any other write on the same queryset — `create()`, `bulk_create()`, `bulk_upsert()`, `bulk_update()`, `get_or_create()`, `update_or_create()` — raises `TypeError` rather than quietly dropping it.
 - `returning()` keeps the queryset's own class, so a custom `QuerySet` and its methods survive it. Chain your own methods before `returning()` — a type checker sees the returning shape after it, not your subclass.
 
 A row lock belongs on the read side of the write, and it composes in either order. The write then needs an open `transaction.atomic()`, and is emitted as a locking sub-select so the lock has somewhere to live — see [Locking a set-based write](#locking-a-set-based-write).
@@ -1388,7 +1446,7 @@ except (psycopg.IntegrityError, ValidationError):
     ...  # lost a race — reload and retry, or report it
 ```
 
-For a plain insert-or-update with no per-row logic, `bulk_create(..., update_conflicts=True, unique_fields=[...])` is an atomic upsert with no race to catch.
+For a plain insert-or-update with no per-row logic, `bulk_upsert(objs, update_fields=[...], unique_fields=[...])` is an atomic upsert with no race to catch.
 
 ### Indexes and constraints
 

plain-postgres/plain/postgres/agents/.claude/rules/plain-postgres.md

@@ -111,11 +111,11 @@ Use `Model.query` to build querysets (e.g., `User.query.filter(is_active=True)`)
 - Use `.annotate(Count(...))` instead of calling `.count()` per row
 - Fetch all data in the view — templates should never trigger queries
 - Use `.exists()` not `.count() > 0`, `.count()` not `len(qs)`
-- Use `bulk_create`/`bulk_update` for batch ops, `.update()`/`.delete()` for mass ops
+- Use `bulk_create`/`bulk_update` for batch ops, `bulk_upsert` for atomic insert-or-update, `.update()`/`.delete()` for mass ops
 - Use `.values_list()` when you only need specific columns
 - Wrap multi-step writes in `transaction.atomic()`
 - Instance writes are `obj.create()` (always INSERT) and `obj.update()` (always UPDATE; `update(fields=[...])` limits the columns) — there is no `save()`, `force_insert`, or `force_update`. Constructing an instance then `create()`-ing it inserts; a hand-set `id` that collides raises `IntegrityError`.
-- `create()`/`update()` raise `ValidationError` (not raw `psycopg.IntegrityError`) on a declared unique/check constraint violation or a foreign key pointing at a missing row, even a raced one — the DB enforces it, so inside an open `transaction.atomic()` the violation aborts the transaction (wrap the write in its own `atomic()` to catch and keep using the transaction). Set-based writes (`QuerySet.update()`/`bulk_create()`) and `delete()` blocked by `RESTRICT` raise raw `psycopg.IntegrityError`. Retrying on conflict? `except (psycopg.IntegrityError, ValidationError)`, or `bulk_create(..., update_conflicts=True)`
+- `create()`/`update()` raise `ValidationError` (not raw `psycopg.IntegrityError`) on a declared unique/check constraint violation or a foreign key pointing at a missing row, even a raced one — the DB enforces it, so inside an open `transaction.atomic()` the violation aborts the transaction (wrap the write in its own `atomic()` to catch and keep using the transaction). Set-based writes (`QuerySet.update()`/`bulk_create()`) and `delete()` blocked by `RESTRICT` raise raw `psycopg.IntegrityError`. Retrying on conflict? `except (psycopg.IntegrityError, ValidationError)`, or `bulk_upsert(objs, update_fields=[...], unique_fields=[...])` for an atomic insert-or-update
 - Always paginate list queries — unbounded querysets get slower as data grows
 
 Run `uv run plain docs postgres` for full patterns with code examples.

plain-postgres/plain/postgres/constants.py

@@ -9,5 +9,4 @@
 
 
 class OnConflict(Enum):
-    IGNORE = "ignore"
     UPDATE = "update"

plain-postgres/plain/postgres/dialect.py

@@ -592,8 +592,6 @@ def on_conflict_suffix_sql(
     update_fields: Iterable[str],
     unique_fields: Iterable[str],
 ) -> str:
-    if on_conflict == OnConflict.IGNORE:
-        return "ON CONFLICT DO NOTHING"
     if on_conflict == OnConflict.UPDATE:
         return "ON CONFLICT({}) DO UPDATE SET {}".format(
             ", ".join(map(quote_name, unique_fields)),

plain-postgres/plain/postgres/options.py

@@ -207,6 +207,17 @@ def total_unique_constraints(self) -> list[Any]:
             )
         ]
 
+    def unique_fields_match_constraint(self, field_names: set[str | None]) -> bool:
+        """True if field_names names the primary key, or a UniqueConstraint on
+        the model that has no condition and no expressions."""
+        pk_field = self.model._model_meta.get_forward_field("id")
+        if field_names == {pk_field.name}:
+            return True
+        for constraint in self.total_unique_constraints:
+            if set(constraint.fields) == field_names:
+                return True
+        return False
+
     def __repr__(self) -> str:
         return f"<Options for {self.model.__name__}>"