数据迁移bug修复

WonderQ-Admin/alembic/versions/0009_remove_product_domain.py

@@ -40,6 +40,20 @@ def _drop_column_with_foreign_keys(table_name: str, column_name: str) -> None:
     op.drop_column(table_name, column_name)
 
 
+def _drop_referencing_foreign_keys(table_name: str) -> None:
+    bind = op.get_bind()
+    inspector = inspect(bind)
+    for current_table in inspector.get_table_names():
+        if current_table == table_name:
+            continue
+        for foreign_key in inspector.get_foreign_keys(current_table):
+            if foreign_key.get("referred_table") != table_name:
+                continue
+            constraint_name = foreign_key.get("name")
+            if constraint_name:
+                op.drop_constraint(constraint_name, current_table, type_="foreignkey")
+
+
 def upgrade() -> None:
     bind = op.get_bind()
     existing_tables = set(inspect(bind).get_table_names())
@@ -49,6 +63,7 @@ def upgrade() -> None:
 
     for table_name in (*PRODUCT_TABLES[:3], "RouteSection", "DemandRecommendation", *PRODUCT_TABLES[3:]):
         if table_name in existing_tables:
+            _drop_referencing_foreign_keys(table_name)
             op.drop_table(table_name)
 
 

WonderQ-Admin/tests/test_remove_product_domain_migration.py

@@ -0,0 +1,67 @@
+from importlib.util import module_from_spec, spec_from_file_location
+from pathlib import Path
+
+
+MIGRATION_PATH = Path(__file__).parents[1] / "alembic" / "versions" / "0009_remove_product_domain.py"
+SPEC = spec_from_file_location("remove_product_domain_migration", MIGRATION_PATH)
+assert SPEC and SPEC.loader
+MIGRATION = module_from_spec(SPEC)
+SPEC.loader.exec_module(MIGRATION)
+
+
+class LegacySchemaInspector:
+    def get_table_names(self) -> list[str]:
+        return ["Product", "MiniProgramFavorite", "MiniProgramHistory"]
+
+    def get_foreign_keys(self, table_name: str) -> list[dict[str, object]]:
+        foreign_keys = {
+            "MiniProgramFavorite": [
+                {
+                    "name": "MiniProgramFavorite_productId_fkey",
+                    "constrained_columns": ["productId"],
+                    "referred_table": "Product",
+                    "referred_columns": ["id"],
+                }
+            ],
+            "MiniProgramHistory": [
+                {
+                    "name": "MiniProgramHistory_productId_fkey",
+                    "constrained_columns": ["productId"],
+                    "referred_table": "Product",
+                    "referred_columns": ["id"],
+                }
+            ],
+        }
+        return foreign_keys.get(table_name, [])
+
+
+def test_upgrade_drops_legacy_product_foreign_keys_before_product_table(monkeypatch):
+    events: list[tuple[str, ...]] = []
+    inspector = LegacySchemaInspector()
+
+    monkeypatch.setattr(MIGRATION, "inspect", lambda _bind: inspector)
+    monkeypatch.setattr(MIGRATION.op, "get_bind", lambda: object())
+    monkeypatch.setattr(
+        MIGRATION.op,
+        "drop_constraint",
+        lambda name, table_name, type_: events.append(("drop_constraint", name, table_name, type_)),
+    )
+    monkeypatch.setattr(MIGRATION.op, "drop_table", lambda table_name: events.append(("drop_table", table_name)))
+
+    MIGRATION.upgrade()
+
+    assert events == [
+        (
+            "drop_constraint",
+            "MiniProgramFavorite_productId_fkey",
+            "MiniProgramFavorite",
+            "foreignkey",
+        ),
+        (
+            "drop_constraint",
+            "MiniProgramHistory_productId_fkey",
+            "MiniProgramHistory",
+            "foreignkey",
+        ),
+        ("drop_table", "Product"),
+    ]