"""Build the deployable DAX lineage catalog from one dbt manifest.json. Example (run from ``server/`` after checking the exact dbt revision): /app/.venv/bin/python scripts/build_data_access_lineage_catalog.py \ /tmp/manifest.json \ app/domains/data_access/lineage_catalogs/gcp-digital-bi-dbt.json \ --catalog-id gcp-digital-bi-dbt --source-version dbt-main- """ from __future__ import annotations import argparse import json from pathlib import Path from app.domains.data_access.lineage_catalog import build_dbt_lineage_catalog def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("manifest", type=Path) parser.add_argument("output", type=Path) parser.add_argument("--catalog-id", required=True) parser.add_argument("--source-version", required=True) args = parser.parse_args() manifest = json.loads(args.manifest.read_text(encoding="utf-8")) if not isinstance(manifest, dict): raise ValueError("dbt manifest must be a JSON object") catalog = build_dbt_lineage_catalog( manifest, catalog_id=args.catalog_id, source_version=args.source_version, ) args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text( json.dumps(catalog, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) if __name__ == "__main__": main()