"""Static safety and inventory checks for the model_operations V0.1 DDL.""" from __future__ import annotations import re from pathlib import Path BASE = Path(__file__).parent EXPECTED_TABLES = { "ops_model_categories", "ops_banks", "ops_model_instances", "ops_model_versions", "ops_monitor_batches", "ops_monitor_results", "ops_monitor_feature_metrics", "ops_monitor_distributions", "ops_rule_versions", "ops_rule_items", "ops_monitor_evaluations", "ops_monitor_reviews", "ops_outbox_events", "ops_consumer_inbox", "ops_report_template_versions", "ops_prompt_versions", "ops_prompt_regression_runs", "ops_reports", "ops_report_revisions", "ops_bank_report_configs", "ops_workflows", "ops_workflow_stages", "ops_documents", "ops_usage_events", } def main() -> None: sql_files = sorted(BASE.glob("*.sql")) combined = "\n".join(path.read_text(encoding="utf-8") for path in sql_files) tables = { match.lower() for match in re.findall( r"CREATE\s+TABLE\s+IF\s+NOT\s+EXISTS\s+([a-zA-Z0-9_]+)", combined, flags=re.IGNORECASE, ) } missing = EXPECTED_TABLES - tables extra = tables - EXPECTED_TABLES assert not missing, f"missing tables: {sorted(missing)}" assert not extra, f"unexpected tables: {sorted(extra)}" assert len(re.findall(r"CREATE\s+OR\s+REPLACE\s+VIEW", combined, re.I)) == 1 assert not re.search(r"\b(DROP|TRUNCATE|DELETE\s+FROM)\b", combined, re.I) assert not re.search( r"\b(?:FROM|JOIN|UPDATE|INTO|REFERENCES|TABLE)\s+model_platform\.", combined, re.I, ), "cross-database SQL object reference found" for path in sql_files: line_count = len(path.read_text(encoding="utf-8").splitlines()) assert line_count <= 500, f"{path.name} exceeds 500 lines" print( f"validated {len(sql_files)} SQL files: " f"{len(tables)} tables, 1 view, no destructive statements" ) if __name__ == "__main__": main()