[2d] Add bulk_upsert() and make bulk_create insert-only

.claude/rules/plain-postgres.md

@@ -95,11 +95,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,7 +106,7 @@ def set_many(
         if not mapping:
             return
 
-        # bulk_create fires pre_save, so updated_at's update_now stamps a fresh
+        # bulk_upsert 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
@@ -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, model.updated_at],
+            unique_fields=[model.key],
         )
 
     def get_or_set(

plain-postgres/plain/postgres/README.md

@@ -405,6 +405,38 @@ 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. Every object comes back — inserted or updated —
+with its DB-generated fields (primary key, DB defaults) populated.
+
+```python
+# Insert new items, refresh `value`/`expires_at` on any existing key.
+CacheItem.query.bulk_upsert(
+    [CacheItem(key=k, value=v, expires_at=exp) for k, v in items],
+    update_fields=[CacheItem.value, CacheItem.expires_at],
+    unique_fields=[CacheItem.key],
+)
+```
+
+- `update_fields` and `unique_fields` take field references (`Model.field`), not
+  strings.
+- `unique_fields` must name the **primary key** or a `UniqueConstraint` declared
+  on the model (no condition, no expressions) — this is the conflict target.
+- `update_fields` must be concrete, non-primary-key, and must not overlap
+  `unique_fields`.
+- Every object must have a non-null value for every unique field. `NULL` never
+  conflicts in Postgres, so it can't be upserted.
+- Each batch is sorted by the conflict key and matches returned rows back to
+  objects by that key, so it's safe to run concurrently without deadlocking on
+  overlapping keys.
+
 #### Use queryset `.update()` / `.delete()` for mass operations
 
 ```python
@@ -1224,7 +1256,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

@@ -95,11 +95,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

@@ -580,8 +580,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__}>"