Table.create() now stores configuration in _defaults

When configuration parameters like pk, foreign_keys, not_null, etc.
are passed to Table.create(), they are now stored in self._defaults
so that subsequent operations on the same Table instance (like insert,
upsert) will use those settings automatically.

Previously, calling db["table"].insert({...}, pk="id") would create
the table correctly but subsequent calls to insert() or upsert() on
the same table object would not know about the pk setting, causing
issues like upsert() failing with "requires a pk" error.

Closes #655

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Simon Willison 2025-11-23 14:51:12 -08:00
commit fafa966300
2 changed files with 139 additions and 9 deletions

View file

@ -1703,19 +1703,19 @@ class Table(Queryable):
def create(
self,
columns: Dict[str, Any],
pk: Optional[Any] = None,
foreign_keys: Optional[ForeignKeysType] = None,
column_order: Optional[List[str]] = None,
not_null: Optional[Iterable[str]] = None,
defaults: Optional[Dict[str, Any]] = None,
hash_id: Optional[str] = None,
hash_id_columns: Optional[Iterable[str]] = None,
extracts: Optional[Union[Dict[str, str], List[str]]] = None,
pk: Optional[Any] = DEFAULT,
foreign_keys: Optional[ForeignKeysType] = DEFAULT,
column_order: Optional[List[str]] = DEFAULT,
not_null: Optional[Iterable[str]] = DEFAULT,
defaults: Optional[Dict[str, Any]] = DEFAULT,
hash_id: Optional[str] = DEFAULT,
hash_id_columns: Optional[Iterable[str]] = DEFAULT,
extracts: Optional[Union[Dict[str, str], List[str]]] = DEFAULT,
if_not_exists: bool = False,
replace: bool = False,
ignore: bool = False,
transform: bool = False,
strict: bool = False,
strict: bool = DEFAULT,
) -> "Table":
"""
Create a table with the specified columns.
@ -1737,6 +1737,38 @@ class Table(Queryable):
:param transform: If table already exists transform it to fit the specified schema
:param strict: Apply STRICT mode to table
"""
# Resolve defaults from _defaults (issue #655)
pk = self.value_or_default("pk", pk)
foreign_keys = self.value_or_default("foreign_keys", foreign_keys)
column_order = self.value_or_default("column_order", column_order)
not_null = self.value_or_default("not_null", not_null)
defaults = self.value_or_default("defaults", defaults)
hash_id = self.value_or_default("hash_id", hash_id)
hash_id_columns = self.value_or_default("hash_id_columns", hash_id_columns)
extracts = self.value_or_default("extracts", extracts)
strict = self.value_or_default("strict", strict)
# Store configuration in _defaults for subsequent operations (issue #655)
# Don't store pk if hash_id is set, since pk is derived from hash_id in that case
if pk is not None and hash_id is None:
self._defaults["pk"] = pk
if foreign_keys is not None:
self._defaults["foreign_keys"] = foreign_keys
if column_order is not None:
self._defaults["column_order"] = column_order
if not_null is not None:
self._defaults["not_null"] = not_null
if defaults is not None:
self._defaults["defaults"] = defaults
if hash_id is not None:
self._defaults["hash_id"] = hash_id
if hash_id_columns is not None:
self._defaults["hash_id_columns"] = hash_id_columns
if extracts is not None:
self._defaults["extracts"] = extracts
if strict:
self._defaults["strict"] = strict
columns = {name: value for (name, value) in columns.items()}
with self.db.conn:
self.db.create_table(

View file

@ -1380,3 +1380,101 @@ def test_bad_table_and_view_exceptions(fresh_db):
with pytest.raises(NoView) as ex2:
fresh_db.view("t")
assert ex2.value.args[0] == "View t does not exist"
# Tests for issue #655: Table configuration should be stored in _defaults
# after table creation, so subsequent operations use the same settings.
def test_pk_persists_after_insert_655(fresh_db):
"""When pk is passed to insert(), subsequent inserts should use it."""
table = fresh_db["users"]
table.insert({"id": 1, "name": "Alice"}, pk="id")
# Second insert should use pk="id" from _defaults
table.insert({"id": 2, "name": "Bob"})
assert table.pks == ["id"]
# Verify both rows exist (not overwritten due to missing pk)
assert table.count == 2
def test_pk_persists_after_insert_all_655(fresh_db):
"""When pk is passed to insert_all(), subsequent inserts should use it."""
table = fresh_db["users"]
table.insert_all([{"id": 1, "name": "Alice"}], pk="id")
# Second insert_all should use pk="id" from _defaults
table.insert_all([{"id": 2, "name": "Bob"}])
assert table.pks == ["id"]
assert table.count == 2
def test_pk_persists_after_create_655(fresh_db):
"""When pk is passed to create(), it should be stored in _defaults."""
table = fresh_db["users"]
table.create({"id": int, "name": str}, pk="id")
assert table._defaults["pk"] == "id"
# Subsequent insert should use the pk
table.insert({"id": 1, "name": "Alice"})
table.insert({"id": 2, "name": "Bob"})
assert table.count == 2
def test_foreign_keys_persist_after_create_655(fresh_db):
"""When foreign_keys is passed to create(), it should be stored in _defaults."""
fresh_db["authors"].insert({"id": 1, "name": "Alice"}, pk="id")
table = fresh_db["books"]
table.create(
{"id": int, "title": str, "author_id": int},
pk="id",
foreign_keys=[("author_id", "authors", "id")],
)
assert table._defaults["pk"] == "id"
assert table._defaults["foreign_keys"] == [("author_id", "authors", "id")]
def test_not_null_persists_after_create_655(fresh_db):
"""When not_null is passed to create(), it should be stored in _defaults."""
table = fresh_db["users"]
table.create({"id": int, "name": str}, pk="id", not_null=["name"])
assert table._defaults["not_null"] == ["name"]
def test_defaults_persist_after_create_655(fresh_db):
"""When defaults is passed to create(), it should be stored in _defaults."""
table = fresh_db["users"]
table.create({"id": int, "score": int}, pk="id", defaults={"score": 0})
assert table._defaults["defaults"] == {"score": 0}
def test_strict_persists_after_create_655(fresh_db):
"""When strict is passed to create(), it should be stored in _defaults."""
table = fresh_db["users"]
table.create({"id": int, "name": str}, pk="id", strict=True)
assert table._defaults["strict"] == True
def test_upsert_uses_pk_from_prior_insert_655(fresh_db):
"""After insert with pk, upsert should use the same pk."""
table = fresh_db["users"]
table.insert({"id": 1, "name": "Alice"}, pk="id")
# Upsert should work without specifying pk again
table.upsert({"id": 1, "name": "Alice Updated"})
assert table.count == 1
assert list(table.rows)[0]["name"] == "Alice Updated"
def test_upsert_all_uses_pk_from_prior_insert_655(fresh_db):
"""After insert with pk, upsert_all should use the same pk."""
table = fresh_db["users"]
table.insert({"id": 1, "name": "Alice"}, pk="id")
# Upsert_all should work without specifying pk again
table.upsert_all([{"id": 1, "name": "Alice Updated"}, {"id": 2, "name": "Bob"}])
assert table.count == 2
rows = {row["id"]: row["name"] for row in table.rows}
assert rows == {1: "Alice Updated", 2: "Bob"}
def test_chained_create_sets_pks(fresh_db):
table = fresh_db.table("dogs3", pk="id").create(
{"id": int, "name": str, "color": str}
)
assert table.pks == ["id"]