[3d] Add upsert() and remove update_or_create()

.claude/rules/plain-postgres.md

@@ -63,11 +63,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, `bulk_upsert` for atomic insert-or-update, `.update()`/`.delete()` for mass ops
+- Use `bulk_create`/`bulk_update` for batch ops, `upsert`/`bulk_upsert` for atomic insert-or-update (single row / many), `.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, 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()`) 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
+- `create()`/`update()` raise `ValidationError` (not raw `psycopg.IntegrityError`) on a declared unique/check constraint violation, 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()`) raise raw `psycopg.IntegrityError`. Retrying on conflict? `except (psycopg.IntegrityError, ValidationError)`, or `upsert(**values, unique_fields=[...])` (single row) / `bulk_upsert(objs, update_fields=[...], unique_fields=[...])` (many) 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-flags/plain/flags/flags.py

@@ -80,9 +80,10 @@ def retrieve_or_compute_value(self) -> Any:
 
             # Create an associated DB Flag that we can use to enable/disable
             # and tie the results to
-            flag_obj, _ = Flag.query.update_or_create(
+            flag_obj, _ = Flag.query.upsert(
                 name=flag_name,
                 defaults={"used_at": timezone.now()},
+                unique_fields=[Flag.name],
             )
 
             if not flag_obj.enabled:

plain-postgres/plain/postgres/README.md

@@ -420,6 +420,51 @@ CacheItem.query.bulk_upsert(
   objects by that key, so it's safe to run concurrently without deadlocking on
   overlapping keys.
 
+#### Use `upsert` for a single insert-or-update
+
+`upsert(*, unique_fields, defaults=None, create_defaults=None, conflict_defaults=None, **kwargs)`
+is the single-row counterpart to `bulk_upsert`. It runs one
+`INSERT ... ON CONFLICT (unique_fields) DO UPDATE SET ... RETURNING` statement and
+returns `(obj, created)` — `created` is `True` when a new row was inserted,
+`False` when the conflicting row was updated. The object is hydrated from the
+post-write row, so there's no second query.
+
+```python
+# Insert the flag, or refresh used_at on the existing one.
+flag, created = Flag.query.upsert(
+    name="beta-dashboard",
+    defaults={"used_at": timezone.now()},
+    unique_fields=[Flag.name],
+)
+```
+
+`unique_fields` takes field references (`Model.field`), like `bulk_upsert`. The
+value sources below stay string-keyed — they follow the `kwargs` idiom.
+
+Value sources:
+
+- `**kwargs` and `defaults` are applied on **both** insert and conflict-update.
+  `kwargs` carries the identifying values (including the unique fields).
+- `create_defaults` is applied on **insert only** — extras that must not change
+  when the row already exists.
+- `conflict_defaults` overrides the `DO UPDATE SET` for specific columns. A value
+  can be a plain value or an expression, so `{"count": F("count") + 1}` is an
+  atomic counter that reads the existing row:
+
+```python
+view, created = PageView.query.upsert(
+    path="/home",
+    conflict_defaults={"count": F("count") + 1},
+    unique_fields=[PageView.path],
+)
+```
+
+On conflict, every non-unique, non-PK column from `kwargs`/`defaults` is set to the
+value the INSERT proposed, minus any column a `conflict_defaults` override
+replaces. As with `bulk_upsert`, `unique_fields` must name the primary key or a
+`UniqueConstraint` (no condition, no expressions), every unique field must be
+non-null, and the merged result is not validated.
+
 #### Use queryset `.update()` / `.delete()` for mass operations
 
 ```python

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

@@ -63,11 +63,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, `bulk_upsert` for atomic insert-or-update, `.update()`/`.delete()` for mass ops
+- Use `bulk_create`/`bulk_update` for batch ops, `upsert`/`bulk_upsert` for atomic insert-or-update (single row / many), `.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, 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()`) 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
+- `create()`/`update()` raise `ValidationError` (not raw `psycopg.IntegrityError`) on a declared unique/check constraint violation, 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()`) raise raw `psycopg.IntegrityError`. Retrying on conflict? `except (psycopg.IntegrityError, ValidationError)`, or `upsert(**values, unique_fields=[...])` (single row) / `bulk_upsert(objs, update_fields=[...], unique_fields=[...])` (many) 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/dialect.py

@@ -583,15 +583,40 @@ def on_conflict_suffix_sql(
     on_conflict: OnConflict | None,
     update_fields: Iterable[str],
     unique_fields: Iterable[str],
+    conflict_overrides: dict[str, str] | None = None,
 ) -> str:
-    if on_conflict == OnConflict.UPDATE:
-        return "ON CONFLICT({}) DO UPDATE SET {}".format(
-            ", ".join(map(quote_name, unique_fields)),
-            ", ".join(
-                [
-                    f"{field} = EXCLUDED.{field}"
-                    for field in map(quote_name, update_fields)
-                ]
-            ),
-        )
-    return ""
+    if on_conflict != OnConflict.UPDATE:
+        return ""
+
+    conflict_overrides = conflict_overrides or {}
+    unique_columns = list(unique_fields)
+    update_columns = list(update_fields)
+
+    # Each base update column is set to its EXCLUDED value (the value the
+    # INSERT proposed), except where conflict_overrides supplies a per-column
+    # SQL fragment (e.g. an atomic "count = count + 1" counter).
+    assignments = []
+    for column in update_columns:
+        quoted = quote_name(column)
+        if column in conflict_overrides:
+            assignments.append(f"{quoted} = {conflict_overrides[column]}")
+        else:
+            assignments.append(f"{quoted} = EXCLUDED.{quoted}")
+
+    # Override-only columns aren't among the EXCLUDED update columns, so append
+    # them after (a counter column, say, that isn't otherwise being set).
+    for column, fragment in conflict_overrides.items():
+        if column not in update_columns:
+            assignments.append(f"{quote_name(column)} = {fragment}")
+
+    if not assignments:
+        # Nothing to update -- Postgres still requires a DO UPDATE SET body,
+        # and only DO UPDATE (not DO NOTHING) returns the conflicting row via
+        # RETURNING. Set a unique column to itself as a no-op.
+        first = quote_name(unique_columns[0])
+        assignments.append(f"{first} = EXCLUDED.{first}")
+
+    return "ON CONFLICT({}) DO UPDATE SET {}".format(
+        ", ".join(map(quote_name, unique_columns)),
+        ", ".join(assignments),
+    )

plain-postgres/plain/postgres/fields/related_managers.py

@@ -13,6 +13,7 @@
     from collections.abc import Callable, Iterable
 
     from plain.postgres.base import Model
+    from plain.postgres.fields.base import Field
     from plain.postgres.fields.related import ForeignKeyField, ManyToManyField
 
 import builtins
@@ -205,15 +206,31 @@ def create(self, **kwargs: Any) -> T:
         kwargs[self.field.name] = self.instance
         return self.model.query.create(**kwargs)
 
-    def get_or_create(self, **kwargs: Any) -> tuple[T, bool]:
+    def get_or_create(
+        self, *, defaults: dict[str, Any] | None = None, **kwargs: Any
+    ) -> tuple[T, bool]:
         self._check_fk_val()
         kwargs[self.field.name] = self.instance
-        return self.model.query.get_or_create(**kwargs)
+        return self.model.query.get_or_create(defaults=defaults, **kwargs)
 
-    def update_or_create(self, **kwargs: Any) -> tuple[T, bool]:
+    def upsert(
+        self,
+        *,
+        defaults: dict[str, Any] | None = None,
+        create_defaults: dict[str, Any] | None = None,
+        conflict_defaults: dict[str, Any] | None = None,
+        unique_fields: list[Field],
+        **kwargs: Any,
+    ) -> tuple[T, bool]:
         self._check_fk_val()
         kwargs[self.field.name] = self.instance
-        return self.model.query.update_or_create(**kwargs)
+        return self.model.query.upsert(
+            defaults=defaults,
+            create_defaults=create_defaults,
+            conflict_defaults=conflict_defaults,
+            unique_fields=unique_fields,
+            **kwargs,
+        )
 
     def remove(self, *objs: T, bulk: bool = True) -> None:
         # remove() is only provided if the ForeignKeyField can have a value of null
@@ -485,21 +502,38 @@ def create(
         return new_obj
 
     def get_or_create(
-        self, *, through_defaults: dict[str, Any] | None = None, **kwargs: Any
+        self,
+        *,
+        defaults: dict[str, Any] | None = None,
+        through_defaults: dict[str, Any] | None = None,
+        **kwargs: Any,
     ) -> tuple[T, bool]:
-        obj, created = self.model.query.get_or_create(**kwargs)
+        obj, created = self.model.query.get_or_create(defaults=defaults, **kwargs)
         # We only need to add() if created because if we got an object back
         # from get() then the relationship already exists.
         if created:
             self.add(obj, through_defaults=through_defaults)
         return obj, created
 
-    def update_or_create(
-        self, *, through_defaults: dict[str, Any] | None = None, **kwargs: Any
+    def upsert(
+        self,
+        *,
+        through_defaults: dict[str, Any] | None = None,
+        defaults: dict[str, Any] | None = None,
+        create_defaults: dict[str, Any] | None = None,
+        conflict_defaults: dict[str, Any] | None = None,
+        unique_fields: list[Field],
+        **kwargs: Any,
     ) -> tuple[T, bool]:
-        obj, created = self.model.query.update_or_create(**kwargs)
-        # We only need to add() if created because if we got an object back
-        # from get() then the relationship already exists.
+        obj, created = self.model.query.upsert(
+            defaults=defaults,
+            create_defaults=create_defaults,
+            conflict_defaults=conflict_defaults,
+            unique_fields=unique_fields,
+            **kwargs,
+        )
+        # We only need to add() if created because if the row already existed
+        # the relationship does too.
         if created:
             self.add(obj, through_defaults=through_defaults)
         return obj, created