remote/sqlite-utils
1
0
Fork 0
mirror of https://github.com/simonw/sqlite-utils.git synced 2026-07-23 09:24:31 +02:00
Code Issues Projects Releases Packages Wiki Activity

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

Browse source
This commit is contained in:
Simon Willison 2021-06-12 19:57:21 -07:00
commit b9629099ab
3 changed files with 29 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

28
sqlite_utils/db.py
View file

@ -1732,20 +1732,22 @@ class Table(Queryable):
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)

7
tests/test_create.py
View file

@ -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)

7
tests/test_upsert.py
View file

@ -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):