diff --git a/docs/changelog.rst b/docs/changelog.rst index a68a9b6..2b2e5cf 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -11,6 +11,7 @@ Unreleased - JSON output from the command-line tool no longer escapes non-ASCII characters, so ``sqlite-utils data.db "select '日本語' as text"`` now outputs ``[{"text": "日本語"}]``. This matches how values were already stored by ``insert`` and how CSV/TSV output already behaved. A new ``--ascii`` option restores the previous behavior of escaping non-ASCII characters, for output destinations that cannot handle UTF-8 - see :ref:`cli_query_json_ascii`. The option is available on the ``query``, ``rows``, ``search``, ``tables``, ``views``, ``triggers``, ``indexes`` and ``memory`` commands. The ``convert --multi --dry-run`` preview and ``plugins`` output also no longer escape non-ASCII characters. (:issue:`625`) - ``table.insert_all(..., pk=...)`` now raises ``InvalidColumns`` if ``pk=`` names columns that do not exist in an existing table. Previously this behaved inconsistently, with single-row inserts raising a ``KeyError`` while other row counts succeeded. (:issue:`732`) +- Fixed an ``IndexError`` from ``table.insert(..., pk=..., ignore=True)`` when an ignored insert followed writes to another table on the same connection. ``last_pk`` is now populated from the explicit primary key value instead of looking up a stale ``lastrowid``. (:issue:`554`) .. _v4_0rc3: diff --git a/sqlite_utils/db.py b/sqlite_utils/db.py index 68aee3e..916d2f0 100644 --- a/sqlite_utils/db.py +++ b/sqlite_utils/db.py @@ -4386,18 +4386,54 @@ 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[resolve_casing(pk, row)] + 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( + resolve_casing(pk, column_names) + ) + self.last_pk = first_record_list[pk_index] + elif pk: + self.last_pk = tuple( + first_record_list[ + column_names.index(resolve_casing(p, column_names)) + ] + for p in pk + ) else: - self.last_pk = tuple(row[resolve_casing(p, row)] 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[ + resolve_casing(pk, first_record_dict) + ] + elif pk: + self.last_pk = tuple( + first_record_dict[resolve_casing(p, first_record_dict)] + 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[resolve_casing(pk, row)] + else: + self.last_pk = tuple( + row[resolve_casing(p, row)] for p in pk + ) + else: + self.last_pk = self.last_rowid else: # For an upsert use first_record from earlier if list_mode: diff --git a/tests/test_create.py b/tests/test_create.py index 7fbd7d5..2788920 100644 --- a/tests/test_create.py +++ b/tests/test_create.py @@ -953,6 +953,24 @@ def test_insert_ignore(fresh_db): assert rows == [{"id": 1, "bar": 2}] +def test_insert_ignore_with_pk_after_other_table_insert(fresh_db): + # https://github.com/simonw/sqlite-utils/issues/554 + 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