mirror of
https://github.com/simonw/sqlite-utils.git
synced 2026-08-04 08:24:20 +02:00
Fix mypy type errors in update_incoming_fks implementation
- Add type annotation for incoming_fk_sqls: List[str] - Use self.db.table() instead of self.db[] to get Table type - Use setattr/getattr for _skip_fk_validation to avoid attr-defined error
This commit is contained in:
parent
5ce545b592
commit
c727ceab07
1 changed files with 7 additions and 7 deletions
|
|
@ -1867,7 +1867,7 @@ class Table(Queryable):
|
||||||
if other_table_name == self.name:
|
if other_table_name == self.name:
|
||||||
continue
|
continue
|
||||||
|
|
||||||
other_table = self.db[other_table_name]
|
other_table = self.db.table(other_table_name)
|
||||||
other_fks = other_table.foreign_keys
|
other_fks = other_table.foreign_keys
|
||||||
|
|
||||||
# Check if any FK references a column being renamed
|
# Check if any FK references a column being renamed
|
||||||
|
|
@ -1927,26 +1927,26 @@ class Table(Queryable):
|
||||||
assert self.exists(), "Cannot transform a table that doesn't exist yet"
|
assert self.exists(), "Cannot transform a table that doesn't exist yet"
|
||||||
|
|
||||||
# Collect SQL for updating incoming FKs if needed
|
# Collect SQL for updating incoming FKs if needed
|
||||||
incoming_fk_sqls = []
|
incoming_fk_sqls: List[str] = []
|
||||||
if update_incoming_fks and rename:
|
if update_incoming_fks and rename:
|
||||||
tables_needing_update = self._get_incoming_fks_needing_update(rename)
|
tables_needing_update = self._get_incoming_fks_needing_update(rename)
|
||||||
for other_table_name, new_fks in tables_needing_update:
|
for other_table_name, new_fks in tables_needing_update:
|
||||||
other_table = self.db[other_table_name]
|
other_table = self.db.table(other_table_name)
|
||||||
# Generate transform SQL for the other table with updated FKs
|
# Generate transform SQL for the other table with updated FKs
|
||||||
# Skip FK validation since the new column doesn't exist yet
|
# Skip FK validation since the new column doesn't exist yet
|
||||||
try:
|
try:
|
||||||
self.db._skip_fk_validation = True
|
setattr(self.db, "_skip_fk_validation", True)
|
||||||
incoming_fk_sqls.extend(
|
incoming_fk_sqls.extend(
|
||||||
other_table.transform_sql(foreign_keys=new_fks)
|
other_table.transform_sql(foreign_keys=new_fks)
|
||||||
)
|
)
|
||||||
finally:
|
finally:
|
||||||
self.db._skip_fk_validation = False
|
setattr(self.db, "_skip_fk_validation", False)
|
||||||
|
|
||||||
# Skip FK validation for main transform if update_incoming_fks is True
|
# Skip FK validation for main transform if update_incoming_fks is True
|
||||||
# because self-referential FKs will reference the new column name
|
# because self-referential FKs will reference the new column name
|
||||||
# that only exists after the transform completes
|
# that only exists after the transform completes
|
||||||
if update_incoming_fks and rename:
|
if update_incoming_fks and rename:
|
||||||
self.db._skip_fk_validation = True
|
setattr(self.db, "_skip_fk_validation", True)
|
||||||
try:
|
try:
|
||||||
sqls = self.transform_sql(
|
sqls = self.transform_sql(
|
||||||
types=types,
|
types=types,
|
||||||
|
|
@ -1963,7 +1963,7 @@ class Table(Queryable):
|
||||||
)
|
)
|
||||||
finally:
|
finally:
|
||||||
if update_incoming_fks and rename:
|
if update_incoming_fks and rename:
|
||||||
self.db._skip_fk_validation = False
|
setattr(self.db, "_skip_fk_validation", False)
|
||||||
|
|
||||||
pragma_foreign_keys_was_on = self.db.execute("PRAGMA foreign_keys").fetchone()[
|
pragma_foreign_keys_was_on = self.db.execute("PRAGMA foreign_keys").fetchone()[
|
||||||
0
|
0
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue