Fix upserts with only primary key

Author: bdowning

sql_athame/dataclasses.py

@@ -875,19 +875,31 @@ def upsert_sql(
             force_update
         )  # Remove force_update from manual insert_only too
         all_insert_only = manual_insert_only | auto_insert_only
-        cached = cls._cached(
-            ("upsert_sql", tuple(sorted(all_insert_only))),
-            lambda: sql(
-                " ON CONFLICT ({pks}) DO UPDATE SET {assignments}",
+
+        def generate_upsert_fragment():
+            updatable_fields = cls.field_names(
+                exclude=(*cls.primary_key_names, *all_insert_only)
+            )
+            return sql(
+                " ON CONFLICT ({pks}) DO {action}",
                 insert_sql=insert_sql,
                 pks=sql.list(cls.primary_key_names_sql()),
-                assignments=sql.list(
-                    sql("{field}=EXCLUDED.{field}", field=x)
-                    for x in cls.field_names_sql(
-                        exclude=(*cls.primary_key_names, *all_insert_only)
+                action=(
+                    sql(
+                        "UPDATE SET {assignments}",
+                        assignments=sql.list(
+                            sql("{field}=EXCLUDED.{field}", field=sql.identifier(field))
+                            for field in updatable_fields
+                        ),
                     )
+                    if updatable_fields
+                    else sql.literal("NOTHING")
                 ),
-            ).flatten(),
+            ).flatten()
+
+        cached = cls._cached(
+            ("upsert_sql", tuple(sorted(all_insert_only))),
+            generate_upsert_fragment,
         )
         return Fragment([insert_sql, cached])
 

tests/test_dataclasses.py

@@ -440,3 +440,42 @@ class Test(ModelBase, table_name="table", primary_key="id"):
     assert '"name"=EXCLUDED."name"' not in partial_query  # Excluded manually
     assert '"updated_at"=EXCLUDED."updated_at"' in partial_query
     assert '"created_at"=EXCLUDED."created_at"' in partial_query  # Force updated
+
+
+def test_primary_key_only_table_upsert():
+    """Test that upsert works correctly for tables with only primary key columns."""
+
+    @dataclass
+    class PrimaryKeyOnly(ModelBase, table_name="pk_only", primary_key="id"):
+        id: uuid.UUID
+
+    test_instance = PrimaryKeyOnly(uuid.uuid4())
+    insert_sql = test_instance.insert_sql()
+
+    # Test that upsert generates valid SQL with DO NOTHING
+    upsert_sql = PrimaryKeyOnly.upsert_sql(insert_sql)
+    query, params = upsert_sql.query()
+
+    # Should contain ON CONFLICT DO NOTHING since there are no updatable fields
+    assert "ON CONFLICT" in query
+    assert "DO NOTHING" in query
+    assert "DO UPDATE SET" not in query
+    assert len(params) == 1  # Only the ID parameter
+
+    # Test with compound primary key (still no other fields)
+    @dataclass
+    class CompoundPrimaryKeyOnly(
+        ModelBase, table_name="compound_pk_only", primary_key=("id1", "id2")
+    ):
+        id1: int
+        id2: str
+
+    compound_instance = CompoundPrimaryKeyOnly(1, "test")
+    compound_insert = compound_instance.insert_sql()
+    compound_upsert = CompoundPrimaryKeyOnly.upsert_sql(compound_insert)
+    compound_query, compound_params = compound_upsert.query()
+
+    assert "ON CONFLICT" in compound_query
+    assert "DO NOTHING" in compound_query
+    assert "DO UPDATE SET" not in compound_query
+    assert len(compound_params) == 2  # Both ID parameters