Fix for use_old_upsert branch

Refs https://github.com/simonw/sqlite-utils/pull/653#issuecomment-2864951112

sqlite_utils/db.py

@@ -3086,45 +3086,48 @@ class Table(Queryable):
         # At this point we need compatibility UPSERT for SQLite < 3.24.0
         # (INSERT OR IGNORE + second UPDATE stage)
         queries_and_params = []
-
+        if isinstance(pk, str):
-        insert_sql = (
+            pks = [pk]
-            f"INSERT OR IGNORE INTO [{self.name}] "
+        else:
-            f"({columns_sql}) VALUES {row_placeholders_sql}"
+            pks = pk
-        )
+        self.last_pk = None
-        queries_and_params.append((insert_sql, flat_params))
+        for record_values in values:
-
+            record = dict(zip(all_columns, record_values))
-        # If there is nothing to update we are done.
+            placeholders = list(pks)
-        if not non_pk_cols:
+            # Need to populate not-null columns too, or INSERT OR IGNORE ignores
-            return queries_and_params
+            # them since it ignores the resulting integrity errors
-
+            if not_null:
-        # We can use UPDATE … FROM (VALUES …) on SQLite ≥ 3.33.0
+                placeholders.extend(not_null)
-        # Older SQLite versions will run this as one UPDATE per row
+            sql = "INSERT OR IGNORE INTO [{table}]({cols}) VALUES({placeholders});".format(
-        # – which is what sqlite-utils did prior to this refactor.
+                table=self.name,
-        alias_cols_sql = ", ".join(pk_cols + non_pk_cols)
+                cols=", ".join(["[{}]".format(p) for p in placeholders]),
-
+                placeholders=", ".join(["?" for p in placeholders]),
-        assignments = []
+            )
-        for c in non_pk_cols:
+            queries_and_params.append(
-            if c in conversions:
+                (sql, [record[col] for col in pks] + ["" for _ in (not_null or [])])
-                assignments.append(f"[{c}] = {conversions[c].replace('?', f'v.[{c}]')}")
+            )
-            else:
+            # UPDATE [book] SET [name] = 'Programming' WHERE [id] = 1001;
-                assignments.append(f"[{c}] = v.[{c}]")
+            set_cols = [col for col in all_columns if col not in pks]
-        assignments_sql = ", ".join(assignments)
+            if set_cols:
-
+                sql2 = "UPDATE [{table}] SET {pairs} WHERE {wheres}".format(
-        update_sql = (
+                    table=self.name,
-            f"UPDATE [{self.name}] AS m SET {assignments_sql} "
+                    pairs=", ".join(
-            f"FROM (VALUES {row_placeholders_sql}) "
+                        "[{}] = {}".format(col, conversions.get(col, "?"))
-            f"AS v({alias_cols_sql}) "
+                        for col in set_cols
-            f"WHERE " + " AND ".join(f"m.[{c}] = v.[{c}]" for c in pk_cols)
+                    ),
-        )
+                    wheres=" AND ".join("[{}] = ?".format(pk) for pk in pks),
-
+                )
-        # Parameters for the UPDATE – pk cols first then non-pk cols
+                queries_and_params.append(
-        update_params = []
+                    (
-        for row in values:
+                        sql2,
-            row_dict = dict(zip(all_columns, row))
+                        [record[col] for col in set_cols] + [record[pk] for pk in pks],
-            ordered = [row_dict[c] for c in pk_cols + non_pk_cols]
+                    )
-            update_params.extend(ordered)
+                )
-
+            # We can populate .last_pk right here
-        queries_and_params.append((update_sql, update_params))
+            if num_records_processed == 1:
+                self.last_pk = tuple(record[pk] for pk in pks)
+                if len(self.last_pk) == 1:
+                    self.last_pk = self.last_pk[0]
         return queries_and_params
 
     def insert_chunk(

tests/test_create.py

@@ -173,14 +173,16 @@ def test_create_table_from_example_with_compound_primary_keys(fresh_db):
 @pytest.mark.parametrize(
     "method_name", ("insert", "upsert", "insert_all", "upsert_all")
 )
-def test_create_table_with_custom_columns(fresh_db, method_name):
+@pytest.mark.parametrize("use_old_upsert", (False, True))
-    table = fresh_db["dogs"]
+def test_create_table_with_custom_columns(method_name, use_old_upsert):
+    db = Database(memory=True, use_old_upsert=use_old_upsert)
+    table = db["dogs"]
     method = getattr(table, method_name)
     record = {"id": 1, "name": "Cleo", "age": "5"}
     if method_name.endswith("_all"):
         record = [record]
     method(record, pk="id", columns={"age": int, "weight": float})
-    assert ["dogs"] == fresh_db.table_names()
+    assert ["dogs"] == db.table_names()
     expected_columns = [
         {"name": "id", "type": "INTEGER"},
         {"name": "name", "type": "TEXT"},

tests/test_upsert.py

@@ -1,9 +1,12 @@
 from sqlite_utils.db import PrimaryKeyRequired
+from sqlite_utils import Database
 import pytest
 
 
-def test_upsert(fresh_db):
+@pytest.mark.parametrize("use_old_upsert", (False, True))
-    table = fresh_db["table"]
+def test_upsert(use_old_upsert):
+    db = Database(memory=True, use_old_upsert=use_old_upsert)
+    table = db["table"]
     table.insert({"id": 1, "name": "Cleo"}, pk="id")
     table.upsert({"id": 1, "age": 5}, pk="id", alter=True)
     assert list(table.rows) == [{"id": 1, "name": "Cleo", "age": 5}]