Run black and cog for formatting and CLI reference update

This commit is contained in:
Claude 2026-01-21 15:00:37 +00:00
commit 5ce545b592
No known key found for this signature in database
4 changed files with 59 additions and 19 deletions

View file

@ -474,6 +474,8 @@ See :ref:`cli_transform_table`.
Add a foreign key constraint from a column to Add a foreign key constraint from a column to
another table with another column another table with another column
--drop-foreign-key TEXT Drop foreign key constraint for this column --drop-foreign-key TEXT Drop foreign key constraint for this column
--update-incoming-fks Update foreign keys in other tables that
reference renamed columns
--sql Output SQL without executing it --sql Output SQL without executing it
--load-extension TEXT Path to SQLite extension, with optional --load-extension TEXT Path to SQLite extension, with optional
:entrypoint :entrypoint

View file

@ -1877,9 +1877,7 @@ class Table(Queryable):
if fk.other_table == self.name and fk.other_column in rename: if fk.other_table == self.name and fk.other_column in rename:
# This FK needs updating # This FK needs updating
needs_update = True needs_update = True
new_fks.append( new_fks.append((fk.column, fk.other_table, rename[fk.other_column]))
(fk.column, fk.other_table, rename[fk.other_column])
)
else: else:
new_fks.append((fk.column, fk.other_table, fk.other_column)) new_fks.append((fk.column, fk.other_table, fk.other_column))

View file

@ -1828,7 +1828,9 @@ def test_transform_update_incoming_fks_cli(db_path):
"transform", "transform",
db_path, db_path,
"authors", "authors",
"--rename", "id", "author_pk", "--rename",
"id",
"author_pk",
"--update-incoming-fks", "--update-incoming-fks",
], ],
) )

View file

@ -682,7 +682,9 @@ def test_transform_update_incoming_fks_on_column_rename(fresh_db):
# Verify initial FK # Verify initial FK
assert fresh_db["books"].foreign_keys == [ assert fresh_db["books"].foreign_keys == [
ForeignKey(table="books", column="author_id", other_table="authors", other_column="id") ForeignKey(
table="books", column="author_id", other_table="authors", other_column="id"
)
] ]
# Rename authors.id to authors.author_pk with update_incoming_fks=True # Rename authors.id to authors.author_pk with update_incoming_fks=True
@ -697,12 +699,19 @@ def test_transform_update_incoming_fks_on_column_rename(fresh_db):
# Verify books FK was updated to point to new column name # Verify books FK was updated to point to new column name
assert fresh_db["books"].foreign_keys == [ assert fresh_db["books"].foreign_keys == [
ForeignKey(table="books", column="author_id", other_table="authors", other_column="author_pk") ForeignKey(
table="books",
column="author_id",
other_table="authors",
other_column="author_pk",
)
] ]
# Verify data integrity # Verify data integrity
assert list(fresh_db["authors"].rows) == [{"author_pk": 1, "name": "Alice"}] assert list(fresh_db["authors"].rows) == [{"author_pk": 1, "name": "Alice"}]
assert list(fresh_db["books"].rows) == [{"id": 1, "title": "Book A", "author_id": 1}] assert list(fresh_db["books"].rows) == [
{"id": 1, "title": "Book A", "author_id": 1}
]
# Verify FK enforcement still works # Verify FK enforcement still works
assert fresh_db.execute("PRAGMA foreign_keys").fetchone()[0] == 1 assert fresh_db.execute("PRAGMA foreign_keys").fetchone()[0] == 1
@ -749,13 +758,28 @@ def test_transform_update_incoming_fks_multiple_tables(fresh_db):
# Verify all FKs were updated # Verify all FKs were updated
assert fresh_db["books"].foreign_keys == [ assert fresh_db["books"].foreign_keys == [
ForeignKey(table="books", column="author_id", other_table="authors", other_column="author_pk") ForeignKey(
table="books",
column="author_id",
other_table="authors",
other_column="author_pk",
)
] ]
assert fresh_db["articles"].foreign_keys == [ assert fresh_db["articles"].foreign_keys == [
ForeignKey(table="articles", column="writer_id", other_table="authors", other_column="author_pk") ForeignKey(
table="articles",
column="writer_id",
other_table="authors",
other_column="author_pk",
)
] ]
assert fresh_db["quotes"].foreign_keys == [ assert fresh_db["quotes"].foreign_keys == [
ForeignKey(table="quotes", column="speaker_id", other_table="authors", other_column="author_pk") ForeignKey(
table="quotes",
column="speaker_id",
other_table="authors",
other_column="author_pk",
)
] ]
# Verify FK enforcement still works # Verify FK enforcement still works
@ -770,22 +794,31 @@ def test_transform_update_incoming_fks_self_referential(fresh_db):
fresh_db.execute("PRAGMA foreign_keys=ON") fresh_db.execute("PRAGMA foreign_keys=ON")
# Create employees table with self-referential FK (manager_id -> id) # Create employees table with self-referential FK (manager_id -> id)
fresh_db.execute(""" fresh_db.execute(
"""
CREATE TABLE employees ( CREATE TABLE employees (
id INTEGER PRIMARY KEY, id INTEGER PRIMARY KEY,
name TEXT, name TEXT,
manager_id INTEGER REFERENCES employees(id) manager_id INTEGER REFERENCES employees(id)
) )
""") """
fresh_db["employees"].insert_all([ )
{"id": 1, "name": "CEO", "manager_id": None}, fresh_db["employees"].insert_all(
{"id": 2, "name": "VP", "manager_id": 1}, [
{"id": 3, "name": "Dev", "manager_id": 2}, {"id": 1, "name": "CEO", "manager_id": None},
]) {"id": 2, "name": "VP", "manager_id": 1},
{"id": 3, "name": "Dev", "manager_id": 2},
]
)
# Verify initial FK # Verify initial FK
assert fresh_db["employees"].foreign_keys == [ assert fresh_db["employees"].foreign_keys == [
ForeignKey(table="employees", column="manager_id", other_table="employees", other_column="id") ForeignKey(
table="employees",
column="manager_id",
other_table="employees",
other_column="id",
)
] ]
# Rename employees.id to employees.emp_id with update_incoming_fks=True # Rename employees.id to employees.emp_id with update_incoming_fks=True
@ -800,7 +833,12 @@ def test_transform_update_incoming_fks_self_referential(fresh_db):
# Verify self-referential FK was updated # Verify self-referential FK was updated
assert fresh_db["employees"].foreign_keys == [ assert fresh_db["employees"].foreign_keys == [
ForeignKey(table="employees", column="manager_id", other_table="employees", other_column="emp_id") ForeignKey(
table="employees",
column="manager_id",
other_table="employees",
other_column="emp_id",
)
] ]
# Verify data integrity # Verify data integrity