upsert() now requires a value for every primary key column

A record missing a primary key value - or with None for one - can
never match an existing row, because a NULL primary key never
satisfies ON CONFLICT. Such records were previously inserted as
brand new rows (silently, for multi-record upsert_all calls) or
triggered a KeyError after the insert had already happened. Both
cases, including upserting an empty record, now raise
PrimaryKeyRequired before any SQL is executed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UnLnhsH25Nnv7LHhekUfPd
This commit is contained in:
Claude 2026-07-04 18:43:37 +00:00
commit d6753a9f57
No known key found for this signature in database
4 changed files with 64 additions and 0 deletions

View file

@ -49,6 +49,42 @@ def test_upsert_error_if_no_pk(fresh_db):
table.upsert({"id": 1, "name": "Cleo"})
@pytest.mark.parametrize("use_old_upsert", (False, True))
def test_upsert_empty_record_errors(use_old_upsert):
db = Database(memory=True, use_old_upsert=use_old_upsert)
table = db["table"]
table.insert({"id": 1, "name": "Cleo"}, pk="id")
with pytest.raises(PrimaryKeyRequired):
table.upsert({}, pk="id")
with pytest.raises(PrimaryKeyRequired):
table.upsert_all([{}, {}], pk="id")
# No rows can have been inserted
assert table.count == 1
@pytest.mark.parametrize("use_old_upsert", (False, True))
def test_upsert_missing_pk_value_errors(use_old_upsert):
db = Database(memory=True, use_old_upsert=use_old_upsert)
table = db["table"]
table.insert({"id": 1, "name": "Cleo"}, pk="id")
# Records that omit the pk column entirely
with pytest.raises(PrimaryKeyRequired):
table.upsert_all([{"name": "Pancakes"}, {"name": "Marnie"}], pk="id")
# A record with an explicit None pk value can never conflict
with pytest.raises(PrimaryKeyRequired):
table.upsert({"id": None, "name": "Pancakes"}, pk="id")
assert list(table.rows) == [{"id": 1, "name": "Cleo"}]
def test_upsert_missing_compound_pk_value_errors(fresh_db):
table = fresh_db["table"]
table.insert({"a": "x", "b": "y", "v": 1}, pk=("a", "b"))
# Missing one component of the detected compound primary key
with pytest.raises(PrimaryKeyRequired):
table.upsert({"a": "x", "v": 2})
assert list(table.rows) == [{"a": "x", "b": "y", "v": 1}]
def test_upsert_error_if_existing_table_has_no_pk(fresh_db):
table = fresh_db.create_table("table", {"id": int, "name": str})
with pytest.raises(PrimaryKeyRequired):