remote/sqlite-utils
1
0
Fork 0
mirror of https://github.com/simonw/sqlite-utils.git synced 2026-09-19 06:54:25 +02:00
Code Issues Projects Releases Packages Wiki Activity

fix: handle ignored inserts with explicit primary keys

Browse source
This commit is contained in:
Kairos Ops 2026-06-21 14:06:41 +08:00
commit 40a4a0b771
2 changed files with 51 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

44
sqlite_utils/db.py
View file

@ -3604,18 +3604,42 @@ class Table(Queryable):
if num_records_processed == 1:
# For an insert we need to use result.lastrowid
if not upsert and result is not None:
self.last_rowid = result.lastrowid
if (hash_id or pk) and self.last_rowid:
# Set self.last_pk to the pk(s) for that rowid
row = list(self.rows_where("rowid = ?", [self.last_rowid]))[0]
if hash_id:
self.last_pk = row[hash_id]
elif isinstance(pk, str):
self.last_pk = row[pk]
ignored_insert = ignore and result.rowcount == 0
if ignored_insert:
if list_mode:
first_record_list = cast(Sequence[Any], first_record)
if hash_id:
pass
elif isinstance(pk, str):
pk_index = column_names.index(pk)
self.last_pk = first_record_list[pk_index]
elif pk:
self.last_pk = tuple(
first_record_list[column_names.index(p)] for p in pk
)
else:
self.last_pk = tuple(row[p] for p in pk)
first_record_dict = cast(Dict[str, Any], first_record)
if hash_id:
self.last_pk = hash_record(
first_record_dict, hash_id_columns
)
elif isinstance(pk, str):
self.last_pk = first_record_dict[pk]
elif pk:
self.last_pk = tuple(first_record_dict[p] for p in pk)
else:
self.last_pk = self.last_rowid
self.last_rowid = result.lastrowid
if (hash_id or pk) and self.last_rowid:
# Set self.last_pk to the pk(s) for that rowid
row = list(self.rows_where("rowid = ?", [self.last_rowid]))[0]
if hash_id:
self.last_pk = row[hash_id]
elif isinstance(pk, str):
self.last_pk = row[pk]
else:
self.last_pk = tuple(row[p] for p in pk)
else:
self.last_pk = self.last_rowid
else:
# For an upsert use first_record from earlier
if list_mode:

17
tests/test_create.py
View file

@ -937,6 +937,23 @@ def test_insert_ignore(fresh_db):
assert rows == [{"id": 1, "bar": 2}]
def test_insert_ignore_with_pk_after_other_table_insert(fresh_db):
user = {"id": "abc", "name": "david"}
fresh_db["users"].insert(user, pk="id")
fresh_db["comments"].insert_all(
[
{"id": "def", "text": "ok"},
{"id": "ghi", "text": "great"},
],
)
table = fresh_db["users"].insert(user, pk="id", ignore=True)
assert table.last_pk == "abc"
assert list(fresh_db["users"].rows) == [user]
def test_insert_hash_id(fresh_db):
dogs = fresh_db["dogs"]
id = dogs.insert({"name": "Cleo", "twitter": "cleopaws"}, hash_id="id").last_pk