"""separate master table per dropdown list; station staff; QA photos Revision ID: e1f2a3b4c5d6 Revises: c7d8e9f0a1b2 Create Date: 2026-10-02 10:00:00.000000 1. Splits lookup_value into one master table per list (17 tables) plus master_list_registry, copying every value (including hidden ones) across, then drops lookup_value. 2. station_user_assignment: which user works at which station as OPERATOR or QA. 3. qa_response_photo: photos attached to a QA checkpoint during an inspection. Safe to re-run: every step checks what already exists. """ from typing import Sequence, Union from alembic import op import sqlalchemy as sa revision: str = 'e1f2a3b4c5d6' down_revision: Union[str, None] = 'c7d8e9f0a1b2' branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None # Kept here (not imported from app code) so the migration never changes after release. TABLES = { 'PRODUCTION_LINE': 'production_line_master', 'AC_MAKE': 'ac_make_master', 'INTERIOR_WORK': 'interior_work_master', 'YARD_LOCATION': 'yard_location_master', 'MACHINE_TYPE': 'machine_type_master', 'INCOMING_MATERIAL_CATEGORY': 'incoming_material_category_master', 'RENTAL_VENDOR': 'rental_vendor_master', 'RENTAL_EQUIPMENT_TYPE': 'rental_equipment_type_master', 'QA_VERIFICATION_METHOD': 'qa_verification_method_master', 'PRODUCTION_HOLD_TYPE': 'production_hold_type_master', 'PAYMENT_TERMS': 'payment_terms_master', 'REGION': 'region_master', 'ENQUIRY_TYPE': 'enquiry_type_master', 'ENQUIRY_SOURCE': 'enquiry_source_master', 'CHASSIS_DOCUMENT_TYPE': 'chassis_document_type_master', 'GATE_PASS_ITEM': 'gate_pass_item_master', 'THERMAL_CHECK_ITEM': 'thermal_check_item_master', } def _exists(conn, table: str) -> bool: return conn.execute(sa.text("SELECT to_regclass(:t) IS NOT NULL"), {'t': f'public.{table}'}).scalar() def upgrade() -> None: conn = op.get_bind() conn.execute(sa.text(""" CREATE TABLE IF NOT EXISTS master_list_registry ( list_code VARCHAR(60) PRIMARY KEY, table_name VARCHAR(80) NOT NULL UNIQUE, label VARCHAR(150) NOT NULL, description TEXT, sort_order INTEGER NOT NULL DEFAULT 0 ) """)) has_lookup = _exists(conn, 'lookup_value') for code, table in TABLES.items(): extra = '' if code == 'GATE_PASS_ITEM': extra = """ default_qty INTEGER NOT NULL DEFAULT 1, print_row INTEGER NOT NULL DEFAULT 1,""" conn.execute(sa.text(f""" CREATE TABLE IF NOT EXISTS {table} ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), code VARCHAR(120) NOT NULL UNIQUE, label VARCHAR(300) NOT NULL, sort_order INTEGER NOT NULL DEFAULT 0,{extra} is_active BOOLEAN NOT NULL DEFAULT TRUE, created_at TIMESTAMPTZ DEFAULT now(), updated_at TIMESTAMPTZ DEFAULT now() ) """)) if has_lookup: # Registry row (name and description were stored under LOOKUP_CATEGORY). conn.execute(sa.text(""" INSERT INTO master_list_registry (list_code, table_name, label, description, sort_order) SELECT :code, :table, label, meta->>'description', sort_order FROM lookup_value WHERE category = 'LOOKUP_CATEGORY' AND code = :code ON CONFLICT (list_code) DO NOTHING """), {'code': code, 'table': table}) if code == 'GATE_PASS_ITEM': conn.execute(sa.text(f""" INSERT INTO {table} (id, code, label, sort_order, default_qty, print_row, is_active, created_at, updated_at) SELECT lookup_id, code, label, sort_order, COALESCE((meta->>'default_qty')::int, 1), COALESCE((meta->>'row')::int, 1), is_active, created_at, updated_at FROM lookup_value WHERE category = :code ON CONFLICT (code) DO NOTHING """), {'code': code}) else: conn.execute(sa.text(f""" INSERT INTO {table} (id, code, label, sort_order, is_active, created_at, updated_at) SELECT lookup_id, code, label, sort_order, is_active, created_at, updated_at FROM lookup_value WHERE category = :code ON CONFLICT (code) DO NOTHING """), {'code': code}) if has_lookup: conn.execute(sa.text("DROP TABLE lookup_value")) conn.execute(sa.text(""" CREATE TABLE IF NOT EXISTS station_user_assignment ( assignment_id UUID PRIMARY KEY DEFAULT gen_random_uuid(), user_id UUID NOT NULL REFERENCES login_users(user_id) ON DELETE CASCADE, station_id UUID NOT NULL REFERENCES station_master(station_id) ON DELETE CASCADE, role VARCHAR(20) NOT NULL CONSTRAINT ck_station_user_role CHECK (role IN ('OPERATOR', 'QA')), is_active BOOLEAN NOT NULL DEFAULT TRUE, created_by UUID REFERENCES login_users(user_id), created_at TIMESTAMPTZ DEFAULT now(), CONSTRAINT uq_station_user_role UNIQUE (user_id, station_id, role) ) """)) conn.execute(sa.text("CREATE INDEX IF NOT EXISTS ix_station_user_assignment_user_id ON station_user_assignment (user_id)")) conn.execute(sa.text("CREATE INDEX IF NOT EXISTS ix_station_user_assignment_station_id ON station_user_assignment (station_id)")) conn.execute(sa.text(""" CREATE TABLE IF NOT EXISTS qa_response_photo ( photo_id UUID PRIMARY KEY DEFAULT gen_random_uuid(), instance_id UUID NOT NULL REFERENCES qa_checksheet_instance(instance_id) ON DELETE CASCADE, item_id UUID NOT NULL REFERENCES qa_checkpoint_item(item_id), photo_url TEXT NOT NULL, uploaded_by UUID REFERENCES login_users(user_id), uploaded_at TIMESTAMPTZ DEFAULT now() ) """)) conn.execute(sa.text("CREATE INDEX IF NOT EXISTS ix_qa_response_photo_instance_id ON qa_response_photo (instance_id)")) def downgrade() -> None: conn = op.get_bind() conn.execute(sa.text("DROP TABLE IF EXISTS qa_response_photo")) conn.execute(sa.text("DROP TABLE IF EXISTS station_user_assignment")) conn.execute(sa.text(""" CREATE TABLE IF NOT EXISTS lookup_value ( lookup_id UUID PRIMARY KEY DEFAULT gen_random_uuid(), category VARCHAR(60) NOT NULL, code VARCHAR(120) NOT NULL, label VARCHAR(300) NOT NULL, sort_order INTEGER NOT NULL DEFAULT 0, meta JSONB, is_active BOOLEAN NOT NULL DEFAULT TRUE, created_at TIMESTAMPTZ DEFAULT now(), updated_at TIMESTAMPTZ DEFAULT now(), CONSTRAINT uq_lookup_value_category_code UNIQUE (category, code) ) """)) conn.execute(sa.text("CREATE INDEX IF NOT EXISTS ix_lookup_value_category ON lookup_value (category)")) conn.execute(sa.text(""" INSERT INTO lookup_value (category, code, label, sort_order, meta) SELECT 'LOOKUP_CATEGORY', list_code, label, sort_order, jsonb_build_object('description', description) FROM master_list_registry ON CONFLICT DO NOTHING """)) for code, table in TABLES.items(): if not _exists(conn, table): continue meta = "jsonb_build_object('default_qty', default_qty, 'row', print_row)" if code == 'GATE_PASS_ITEM' else 'NULL' conn.execute(sa.text(f""" INSERT INTO lookup_value (lookup_id, category, code, label, sort_order, meta, is_active, created_at, updated_at) SELECT id, :code, code, label, sort_order, {meta}, is_active, created_at, updated_at FROM {table} ON CONFLICT DO NOTHING """), {'code': code}) conn.execute(sa.text(f"DROP TABLE {table}")) conn.execute(sa.text("DROP TABLE IF EXISTS master_list_registry"))