"""add arrival remark and fix template nullability Revision ID: ae19ae1b21e3 Revises: 2fe975d2227b Create Date: 2026-08-12 14:05:00.385967 """ from typing import Sequence, Union from alembic import op import sqlalchemy as sa from sqlalchemy.dialects import postgresql # revision identifiers, used by Alembic. revision: str = 'ae19ae1b21e3' down_revision: Union[str, None] = '2fe975d2227b' branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None def upgrade() -> None: bind = op.get_bind() # Legacy tables from a prior schema revision may already be absent depending on # how the branch was merged. Guard these destructive operations so the migration # can be applied safely across partially-migrated local databases. for table_name in ['chassis_operation_log', 'tracking_phase_master', 'station_operation_master']: if sa.inspect(bind).has_table(table_name): op.drop_table(table_name) if not sa.inspect(bind).has_table('chassis_master'): raise RuntimeError('chassis_master table is required for this migration') for column_name in ['arrival_remark']: if not column_name in {c['name'] for c in sa.inspect(bind).get_columns('chassis_master')}: op.add_column('chassis_master', sa.Column(column_name, sa.Text(), nullable=True)) for column_name in ['current_phase', 'current_operation_id', 'current_sub_station_id']: if column_name in {c['name'] for c in sa.inspect(bind).get_columns('chassis_master')}: op.drop_column('chassis_master', column_name) if sa.inspect(bind).has_table('chassis_station_qa_result'): for column_name in ['item_id', 'is_ok', 'remark', 'inspected_by']: if not column_name in {c['name'] for c in sa.inspect(bind).get_columns('chassis_station_qa_result')}: if column_name == 'item_id': op.add_column('chassis_station_qa_result', sa.Column(column_name, sa.UUID(), nullable=False)) elif column_name == 'is_ok': op.add_column('chassis_station_qa_result', sa.Column(column_name, sa.Boolean(), nullable=False)) elif column_name == 'remark': op.add_column('chassis_station_qa_result', sa.Column(column_name, sa.String(length=255), nullable=True)) else: op.add_column('chassis_station_qa_result', sa.Column(column_name, sa.UUID(), nullable=True)) if 'inspected_at' in {c['name'] for c in sa.inspect(bind).get_columns('chassis_station_qa_result')}: op.alter_column('chassis_station_qa_result', 'inspected_at', existing_type=postgresql.TIMESTAMP(timezone=True), nullable=True) # Drop the legacy foreign key only if it exists. inspector = sa.inspect(bind) fks = inspector.get_foreign_keys('chassis_station_qa_result') foreign_key_names = {fk.get('name') for fk in fks if fk.get('name')} if 'chassis_station_qa_result_inspector_id_fkey' in foreign_key_names: op.drop_constraint(op.f('chassis_station_qa_result_inspector_id_fkey'), 'chassis_station_qa_result', type_='foreignkey') if not sa.inspect(bind).has_table('station_qa_checklist_item_master'): raise RuntimeError('station_qa_checklist_item_master table is required for this migration') if not sa.inspect(bind).has_table('login_users'): raise RuntimeError('login_users table is required for this migration') op.create_foreign_key(None, 'chassis_station_qa_result', 'station_qa_checklist_item_master', ['item_id'], ['item_id']) op.create_foreign_key(None, 'chassis_station_qa_result', 'login_users', ['inspected_by'], ['user_id']) for column_name in ['status', 'remarks', 'inspector_id', 'operation_id']: if column_name in {c['name'] for c in sa.inspect(bind).get_columns('chassis_station_qa_result')}: op.drop_column('chassis_station_qa_result', column_name) if sa.inspect(bind).has_table('chassis_status_log'): for column_name in ['sub_station_code', 'operation_id', 'sub_station_id', 'operation_code', 'phase_code']: if column_name in {c['name'] for c in sa.inspect(bind).get_columns('chassis_status_log')}: op.drop_column('chassis_status_log', column_name) if sa.inspect(bind).has_table('station_master') and 'phase_code' in {c['name'] for c in sa.inspect(bind).get_columns('station_master')}: op.drop_column('station_master', 'phase_code') # Nullability fixes for sales_vehicle_template if sa.inspect(bind).has_table('sales_vehicle_template'): for column_name in ['model_id', 'body_type_id', 'segment_id']: if column_name in {c['name'] for c in sa.inspect(bind).get_columns('sales_vehicle_template')}: op.alter_column('sales_vehicle_template', column_name, existing_type=sa.UUID(), nullable=True) def downgrade() -> None: bind = op.get_bind() if sa.inspect(bind).has_table('station_master') and 'phase_code' not in {c['name'] for c in sa.inspect(bind).get_columns('station_master')}: op.add_column('station_master', sa.Column('phase_code', sa.VARCHAR(length=50), autoincrement=False, nullable=True)) if sa.inspect(bind).has_table('chassis_status_log'): for column_name, column in { 'phase_code': sa.Column('phase_code', sa.VARCHAR(length=50), autoincrement=False, nullable=True), 'operation_code': sa.Column('operation_code', sa.VARCHAR(length=50), autoincrement=False, nullable=True), 'sub_station_id': sa.Column('sub_station_id', sa.UUID(), autoincrement=False, nullable=True), 'operation_id': sa.Column('operation_id', sa.UUID(), autoincrement=False, nullable=True), 'sub_station_code': sa.Column('sub_station_code', sa.VARCHAR(length=50), autoincrement=False, nullable=True), }.items(): if column_name not in {c['name'] for c in sa.inspect(bind).get_columns('chassis_status_log')}: op.add_column('chassis_status_log', column) if sa.inspect(bind).has_table('chassis_station_qa_result'): for column_name, column in { 'operation_id': sa.Column('operation_id', sa.UUID(), autoincrement=False, nullable=True), 'inspector_id': sa.Column('inspector_id', sa.UUID(), autoincrement=False, nullable=True), 'remarks': sa.Column('remarks', sa.TEXT(), autoincrement=False, nullable=True), 'status': sa.Column('status', sa.VARCHAR(length=20), autoincrement=False, nullable=False), }.items(): if column_name not in {c['name'] for c in sa.inspect(bind).get_columns('chassis_station_qa_result')}: op.add_column('chassis_station_qa_result', column) if 'inspected_by' in {c['name'] for c in sa.inspect(bind).get_columns('chassis_station_qa_result')}: op.drop_column('chassis_station_qa_result', 'inspected_by') if 'remark' in {c['name'] for c in sa.inspect(bind).get_columns('chassis_station_qa_result')}: op.drop_column('chassis_station_qa_result', 'remark') if 'is_ok' in {c['name'] for c in sa.inspect(bind).get_columns('chassis_station_qa_result')}: op.drop_column('chassis_station_qa_result', 'is_ok') if 'item_id' in {c['name'] for c in sa.inspect(bind).get_columns('chassis_station_qa_result')}: op.drop_column('chassis_station_qa_result', 'item_id') if 'chassis_station_qa_result_inspector_id_fkey' in {fk.get('name') for fk in sa.inspect(bind).get_foreign_keys('chassis_station_qa_result') if fk.get('name')}: op.drop_constraint(None, 'chassis_station_qa_result', type_='foreignkey') if sa.inspect(bind).has_table('chassis_master'): for column_name in ['current_sub_station_id', 'current_operation_id', 'current_phase']: if column_name not in {c['name'] for c in sa.inspect(bind).get_columns('chassis_master')}: if column_name == 'current_sub_station_id': op.add_column('chassis_master', sa.Column('current_sub_station_id', sa.UUID(), autoincrement=False, nullable=True)) elif column_name == 'current_operation_id': op.add_column('chassis_master', sa.Column('current_operation_id', sa.UUID(), autoincrement=False, nullable=True)) else: op.add_column('chassis_master', sa.Column('current_phase', sa.VARCHAR(length=50), autoincrement=False, nullable=True)) if 'arrival_remark' in {c['name'] for c in sa.inspect(bind).get_columns('chassis_master')}: op.drop_column('chassis_master', 'arrival_remark') if sa.inspect(bind).has_table('sales_vehicle_template'): for column_name in ['segment_id', 'body_type_id', 'model_id']: if column_name in {c['name'] for c in sa.inspect(bind).get_columns('sales_vehicle_template')}: op.alter_column('sales_vehicle_template', column_name, existing_type=sa.UUID(), nullable=False)