73 lines
2.3 KiB
Python
73 lines
2.3 KiB
Python
"""Remove the retired product domain and its foreign-key references.
|
|
|
|
Revision ID: 0009_remove_product_domain
|
|
Revises: 0008_demand_page_modules
|
|
Create Date: 2026-08-17
|
|
"""
|
|
|
|
from alembic import op
|
|
from sqlalchemy import inspect
|
|
|
|
|
|
revision = "0009_remove_product_domain"
|
|
down_revision = "0008_demand_page_modules"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
|
|
PRODUCT_TABLES = (
|
|
"CampaignProduct",
|
|
"RouteSectionProduct",
|
|
"DemandRecommendationProduct",
|
|
"ProductImage",
|
|
"Product",
|
|
)
|
|
|
|
|
|
def _drop_column_with_foreign_keys(table_name: str, column_name: str) -> None:
|
|
bind = op.get_bind()
|
|
inspector = inspect(bind)
|
|
if table_name not in inspector.get_table_names():
|
|
return
|
|
columns = {column["name"] for column in inspector.get_columns(table_name)}
|
|
if column_name not in columns:
|
|
return
|
|
for foreign_key in inspector.get_foreign_keys(table_name):
|
|
if column_name in foreign_key.get("constrained_columns", []):
|
|
constraint_name = foreign_key.get("name")
|
|
if constraint_name:
|
|
op.drop_constraint(constraint_name, table_name, type_="foreignkey")
|
|
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())
|
|
|
|
_drop_column_with_foreign_keys("Lead", "sourceProductId")
|
|
_drop_column_with_foreign_keys("Order", "productId")
|
|
|
|
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)
|
|
|
|
|
|
def downgrade() -> None:
|
|
# The retired product schema intentionally has no downgrade path. Restore it from a backup if needed.
|
|
pass
|