Fix bug with upsert_all() and single column tables, closes #271

Author: simonw

sqlite_utils/db.py

@@ -1732,20 +1732,22 @@ def build_insert_queries_and_params(
                 queries_and_params.append((sql, [record[col] for col in pks]))
                 # UPDATE [book] SET [name] = 'Programming' WHERE [id] = 1001;
                 set_cols = [col for col in all_columns if col not in pks]
-                sql2 = "UPDATE [{table}] SET {pairs} WHERE {wheres}".format(
-                    table=self.name,
-                    pairs=", ".join(
-                        "[{}] = {}".format(col, conversions.get(col, "?"))
-                        for col in set_cols
-                    ),
-                    wheres=" AND ".join("[{}] = ?".format(pk) for pk in pks),
-                )
-                queries_and_params.append(
-                    (
-                        sql2,
-                        [record[col] for col in set_cols] + [record[pk] for pk in pks],
+                if set_cols:
+                    sql2 = "UPDATE [{table}] SET {pairs} WHERE {wheres}".format(
+                        table=self.name,
+                        pairs=", ".join(
+                            "[{}] = {}".format(col, conversions.get(col, "?"))
+                            for col in set_cols
+                        ),
+                        wheres=" AND ".join("[{}] = ?".format(pk) for pk in pks),
+                    )
+                    queries_and_params.append(
+                        (
+                            sql2,
+                            [record[col] for col in set_cols]
+                            + [record[pk] for pk in pks],
+                        )
                     )
-                )
                 # We can populate .last_pk right here
                 if num_records_processed == 1:
                     self.last_pk = tuple(record[pk] for pk in pks)

tests/test_create.py

@@ -991,6 +991,13 @@ def test_insert_all_empty_list(fresh_db):
     assert 1 == fresh_db["t"].count
 
 
+def test_insert_all_single_column(fresh_db):
+    table = fresh_db["table"]
+    table.insert_all([{"name": "Cleo"}], pk="name")
+    assert [{"name": "Cleo"}] == list(table.rows)
+    assert table.pks == ["name"]
+
+
 def test_create_with_a_null_column(fresh_db):
     record = {"name": "Name", "description": None}
     fresh_db["t"].insert(record)

tests/test_upsert.py

@@ -21,6 +21,13 @@ def test_upsert_all(fresh_db):
     assert table.last_pk is None
 
 
+def test_upsert_all_single_column(fresh_db):
+    table = fresh_db["table"]
+    table.upsert_all([{"name": "Cleo"}], pk="name")
+    assert [{"name": "Cleo"}] == list(table.rows)
+    assert table.pks == ["name"]
+
+
 def test_upsert_error_if_no_pk(fresh_db):
     table = fresh_db["table"]
     with pytest.raises(PrimaryKeyRequired):