76 lines
2.2 KiB
Python
76 lines
2.2 KiB
Python
"""reset spifox tenant data for new seed
|
|
|
|
Revision ID: g9b0c1d2e3f4
|
|
Revises: f8a9b0c1d2e3
|
|
Create Date: 2026-02-10 20:00:00.000000
|
|
|
|
"""
|
|
|
|
from typing import Sequence, Union
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
revision: str = "g9b0c1d2e3f4"
|
|
down_revision: Union[str, None] = "f8a9b0c1d2e3"
|
|
branch_labels: Union[str, Sequence[str], None] = None
|
|
depends_on: Union[str, Sequence[str], None] = None
|
|
|
|
|
|
def upgrade() -> None:
|
|
conn = op.get_bind()
|
|
|
|
machine_ids = conn.execute(
|
|
sa.text("SELECT id FROM machines WHERE tenant_id = 'spifox'")
|
|
).fetchall()
|
|
mid_list = [str(r[0]) for r in machine_ids]
|
|
|
|
if not mid_list:
|
|
return
|
|
|
|
part_ids = conn.execute(
|
|
sa.text("SELECT id FROM equipment_parts WHERE tenant_id = 'spifox'")
|
|
).fetchall()
|
|
pid_list = [str(r[0]) for r in part_ids]
|
|
|
|
conn.execute(sa.text("DELETE FROM alarms WHERE tenant_id = 'spifox'"))
|
|
|
|
if pid_list:
|
|
conn.execute(sa.text("DELETE FROM part_counters WHERE tenant_id = 'spifox'"))
|
|
conn.execute(
|
|
sa.text(
|
|
"DELETE FROM part_replacement_logs WHERE equipment_part_id = ANY(:pids)"
|
|
),
|
|
{"pids": pid_list},
|
|
)
|
|
conn.execute(
|
|
sa.text(
|
|
"UPDATE inspection_template_items SET equipment_part_id = NULL "
|
|
"WHERE equipment_part_id = ANY(:pids)"
|
|
),
|
|
{"pids": pid_list},
|
|
)
|
|
|
|
conn.execute(
|
|
sa.text(
|
|
"DELETE FROM inspection_records WHERE session_id IN "
|
|
"(SELECT id FROM inspection_sessions WHERE tenant_id = 'spifox')"
|
|
)
|
|
)
|
|
conn.execute(sa.text("DELETE FROM inspection_sessions WHERE tenant_id = 'spifox'"))
|
|
|
|
conn.execute(
|
|
sa.text(
|
|
"DELETE FROM inspection_template_items WHERE template_id IN "
|
|
"(SELECT id FROM inspection_templates WHERE tenant_id = 'spifox')"
|
|
)
|
|
)
|
|
conn.execute(sa.text("DELETE FROM inspection_templates WHERE tenant_id = 'spifox'"))
|
|
|
|
conn.execute(sa.text("DELETE FROM equipment_parts WHERE tenant_id = 'spifox'"))
|
|
conn.execute(sa.text("DELETE FROM machines WHERE tenant_id = 'spifox'"))
|
|
|
|
|
|
def downgrade() -> None:
|
|
pass
|