from __future__ import annotations from sqlalchemy import inspect, text from sqlalchemy.engine import Engine MASTER_FORMAT_COLUMNS: dict[str, dict[str, str]] = { "users": { "functionality_access": "JSON NULL", }, "requisitions": { "requisition_date": "DATE NULL", "received_from": "VARCHAR(160) NULL", "received_from_district": "VARCHAR(120) NULL", "received_from_range": "VARCHAR(120) NULL", "sender_name": "VARCHAR(160) NULL", "sender_designation": "VARCHAR(160) NULL", "requisition_items": "JSON NULL", "item_description": "TEXT NULL", "quantity": "INT NULL", "estimated_cost": "DECIMAL(14,2) NULL", "procurement_mode": "VARCHAR(120) NULL", "procurement_mode_other": "VARCHAR(255) NULL", "budget_head": "VARCHAR(255) NULL", "administrative_approval_date": "DATE NULL", "administrative_approval_quantity": "INT NULL", "lifecycle_stage": "VARCHAR(80) NULL", }, "procurement_cases": { "procurement_mode_other": "VARCHAR(255) NULL", }, "tender_bids": { "serial_no": "INT NULL", "procurement_case_id": "INT NULL", "tender_date": "DATE NULL", "tender_items": "JSON NULL", "item_name": "TEXT NULL", "quantity": "INT NULL", "estimated_cost": "DECIMAL(14,2) NULL", "budget_head": "VARCHAR(255) NULL", "bid_published_date": "DATE NULL", "bid_closed_date": "DATE NULL", "technical_bid_opening_date": "DATE NULL", "product_demo_date": "DATE NULL", "technical_evaluation_date": "DATE NULL", "technical_qualified_bidders": "INT NULL", "financial_bid_opening_date": "DATE NULL", "l1_bidder_name": "VARCHAR(160) NULL", "contract_number": "VARCHAR(80) NULL", "contract_date": "DATE NULL", "amount": "DECIMAL(14,2) NULL", "remarks": "TEXT NULL", }, "emds": { "procurement_case_id": "INT NULL", "serial_no": "INT NULL", "emd_unique_id": "VARCHAR(80) NULL", "entry_date": "DATE NULL", "tender_number": "VARCHAR(80) NULL", "tender_date": "DATE NULL", "item_name": "TEXT NULL", "emd_form": "VARCHAR(80) NULL", "emd_no": "VARCHAR(80) NULL", "emd_date": "DATE NULL", "issuing_branch": "VARCHAR(160) NULL", "bank_confirmation_date": "DATE NULL", "expiry_date": "DATE NULL", "returned_date": "DATE NULL", "return_details": "TEXT NULL", "remarks": "TEXT NULL", }, "sample_registers": { "serial_no": "INT NULL", "procurement_case_id": "INT NULL", "tender_number": "VARCHAR(80) NULL", "tender_date": "DATE NULL", "sample_items": "JSON NULL", "item_name": "TEXT NULL", "quantity": "INT NULL", "demonstration_date": "DATE NULL", "make": "VARCHAR(120) NULL", "model": "VARCHAR(120) NULL", "make_model": "TEXT NULL", "receipt_date": "DATE NULL", "recipient_name": "VARCHAR(160) NULL", "return_date": "DATE NULL", "remarks": "TEXT NULL", }, "purchase_orders": { "procurement_case_id": "INT NULL", "procurement_mode": "VARCHAR(120) NULL", "procurement_mode_other": "VARCHAR(255) NULL", "tender_reference_no": "VARCHAR(80) NULL", "tender_reference_date": "DATE NULL", "estimated_cost": "DECIMAL(14,2) NULL", "process_initiate_date": "DATE NULL", "budget_head": "VARCHAR(255) NULL", "item_description": "TEXT NULL", "quantity": "INT NULL", "unit": "VARCHAR(80) NULL", }, "gem_procurement_cases": { "delivery_delay_reason": "TEXT NULL", "delivery_delay_document_path": "VARCHAR(255) NULL", }, "item_database": { "item_type": "VARCHAR(80) NOT NULL DEFAULT 'Equipment'", "is_active": "BOOLEAN NOT NULL DEFAULT TRUE", "name_en": "VARCHAR(160) NULL", "name_or": "VARCHAR(160) NULL", "item_type_en": "VARCHAR(80) NULL", "item_type_or": "VARCHAR(80) NULL", "description_en": "TEXT NULL", "description_or": "TEXT NULL", }, "budget_heads": { "display_name_en": "VARCHAR(255) NULL", "display_name_or": "VARCHAR(255) NULL", "category_name_en": "VARCHAR(160) NULL", "category_name_or": "VARCHAR(160) NULL", "expenditure_head_en": "VARCHAR(160) NULL", "expenditure_head_or": "VARCHAR(160) NULL", "item_component_en": "VARCHAR(160) NULL", "item_component_or": "VARCHAR(160) NULL", }, "ofes_establishments": { "name_en": "VARCHAR(160) NULL", "name_or": "VARCHAR(160) NULL", "district_en": "VARCHAR(120) NULL", "district_or": "VARCHAR(120) NULL", "range_name_en": "VARCHAR(120) NULL", "range_name_or": "VARCHAR(120) NULL", "display_name_en": "VARCHAR(255) NULL", "display_name_or": "VARCHAR(255) NULL", }, "performance_securities": { "procurement_case_id": "INT NULL", "entry_date": "DATE NULL", "contract_number": "VARCHAR(80) NULL", "contract_date": "DATE NULL", "item_name": "TEXT NULL", "quantity": "INT NULL", "bidder_name": "VARCHAR(160) NULL", "security_form": "VARCHAR(80) NULL", "security_number": "VARCHAR(80) NULL", "security_date": "DATE NULL", "issuing_branch": "VARCHAR(160) NULL", "bank_confirmation_date": "DATE NULL", "expiry_date": "DATE NULL", "warranty_period": "VARCHAR(120) NULL", "return_date": "DATE NULL", "return_details": "TEXT NULL", "remarks": "TEXT NULL", }, "technical_evaluations": { "procurement_case_id": "INT NULL", }, "financial_evaluations": { "procurement_case_id": "INT NULL", }, "delivery_acceptances": { "procurement_case_id": "INT NULL", }, "payment_tracking": { "procurement_case_id": "INT NULL", }, } NULLABLE_EXISTING_COLUMNS: dict[str, dict[str, str]] = { "requisitions": { "demand_id": "INT NULL", "requested_by": "VARCHAR(120) NULL", "budget_head": "VARCHAR(255) NULL", }, "item_database": { "item_type": "VARCHAR(80) NOT NULL DEFAULT 'Equipment'", "is_active": "BOOLEAN NOT NULL DEFAULT TRUE", }, "tender_bids": { "requisition_id": "INT NULL", "title": "VARCHAR(200) NULL", "budget_head": "VARCHAR(255) NULL", }, "emds": { "tender_id": "INT NULL", "amount": "DECIMAL(14,2) NULL", "instrument_no": "VARCHAR(80) NULL", }, "sample_registers": { "tender_id": "INT NULL", "sample_description": "TEXT NULL", }, "purchase_orders": { "tender_id": "INT NULL", "po_value": "DECIMAL(14,2) NULL", "budget_head": "VARCHAR(255) NULL", }, "performance_securities": { "po_id": "INT NULL", "vendor_name": "VARCHAR(160) NULL", "amount": "DECIMAL(14,2) NULL", "instrument_no": "VARCHAR(80) NULL", }, "technical_evaluations": { "tender_id": "INT NULL", }, "financial_evaluations": { "tender_id": "INT NULL", }, "delivery_acceptances": { "po_id": "INT NULL", }, "payment_tracking": { "po_id": "INT NULL", }, } MINIMUM_STRING_LENGTHS: tuple[tuple[str, str, int, str], ...] = ( ("procurement_cases", "procurement_case_no", 255, "VARCHAR(255) NOT NULL"), ) INDEXES: tuple[tuple[str, str, str], ...] = ( ("tender_bids", "idx_tenders_serial_no", "serial_no"), ("tender_bids", "idx_tenders_procurement_case_id", "procurement_case_id"), ("emds", "idx_emds_procurement_case_id", "procurement_case_id"), ("emds", "idx_emds_serial_no", "serial_no"), ("emds", "idx_emds_tender_number", "tender_number"), ("sample_registers", "idx_samples_tender_number", "tender_number"), ("sample_registers", "idx_samples_procurement_case_id", "procurement_case_id"), ("sample_registers", "idx_samples_serial_no", "serial_no"), ("technical_evaluations", "idx_tech_eval_procurement_case_id", "procurement_case_id"), ("financial_evaluations", "idx_fin_eval_procurement_case_id", "procurement_case_id"), ("purchase_orders", "idx_pos_procurement_case_id", "procurement_case_id"), ("performance_securities", "idx_ps_procurement_case_id", "procurement_case_id"), ("performance_securities", "idx_ps_contract_number", "contract_number"), ("delivery_acceptances", "idx_delivery_procurement_case_id", "procurement_case_id"), ("payment_tracking", "idx_payments_procurement_case_id", "procurement_case_id"), ) def apply_master_format_migrations(engine: Engine) -> None: if engine.dialect.name != "mysql": return inspector = inspect(engine) existing_tables = set(inspector.get_table_names()) with engine.begin() as connection: for table, columns in MASTER_FORMAT_COLUMNS.items(): if table not in existing_tables: continue existing_columns = {column["name"] for column in inspector.get_columns(table)} for column_name, definition in columns.items(): if column_name not in existing_columns: connection.execute( text(f"ALTER TABLE `{table}` ADD COLUMN `{column_name}` {definition}") ) for table, columns in NULLABLE_EXISTING_COLUMNS.items(): if table not in existing_tables: continue existing_columns = {column["name"]: column for column in inspector.get_columns(table)} for column_name, definition in columns.items(): column = existing_columns.get(column_name) desired_nullable = " NOT NULL" not in definition.upper() if column is not None and bool(column.get("nullable", True)) != desired_nullable: connection.execute( text(f"ALTER TABLE `{table}` MODIFY COLUMN `{column_name}` {definition}") ) for table, column_name, minimum_length, definition in MINIMUM_STRING_LENGTHS: if table not in existing_tables: continue column = next( (item for item in inspector.get_columns(table) if item["name"] == column_name), None, ) current_length = getattr(column.get("type"), "length", None) if column else None if column is not None and current_length is not None and current_length < minimum_length: connection.execute( text(f"ALTER TABLE `{table}` MODIFY COLUMN `{column_name}` {definition}") ) existing_indexes = { table: {index["name"] for index in inspector.get_indexes(table)} for table in existing_tables } for table, index_name, column_name in INDEXES: if table in existing_tables and index_name not in existing_indexes.get(table, set()): connection.execute( text(f"CREATE INDEX `{index_name}` ON `{table}` (`{column_name}`)") )