transpile_project.py

"""Transpile a dbt project from one SQL dialect to another.

Usage:
    uv run python scripts/transpile_project.py --src <models_dir> --dst <output_dir> \
        [--from snowflake] [--to databricks] [--workers N]

Exit code: 0 if all files succeeded, 1 if any file failed.
"""
from __future__ import annotations

import argparse
import json
import os
import sys
from concurrent.futures import ProcessPoolExecutor, as_completed
from pathlib import Path

# Allow running from the repo root without installing the package.
sys.path.insert(0, str(Path(__file__).parent.parent))

from sqlglot.dbt.transpiler import DbtTranspiler, _fix_snowflake_syntax_in_output


def _transpile_one(
    src_file: Path,
    dst_file: Path,
    src_dialect: str,
    dst_dialect: str,
) -> dict:
    """Transpile a single file and return its report data (picklable for multiprocessing)."""
    is_macro = "/macros/" in str(src_file) or "\\macros\\" in str(src_file)
    dst_file.parent.mkdir(parents=True, exist_ok=True)

    if is_macro:
        content = src_file.read_text(encoding="utf-8")
        content = _fix_snowflake_syntax_in_output(content)
        dst_file.write_text(content, encoding="utf-8")
        result_data = {
            "path": str(src_file),
            "success": True,
            "errors": [],
            "warnings": [],
            "certification": "UNCERTIFIED",
            "is_macro_file": True,
            "output": content,
        }
    else:
        transpiler = DbtTranspiler(
            source_dialect=src_dialect,
            target_dialect=dst_dialect,
            allow_opaque=True,
        )
        result = transpiler.transpile_file(src_file, dst_file)
        result_data = {
            "path": result.path,
            "success": result.success,
            "errors": result.errors,
            "warnings": result.warnings,
            "certification": result.certification,
            "is_macro_file": result.is_macro_file,
            "output": result.output,
        }

    report_path = dst_file.parent / f"{dst_file.stem}_transpile_report.json"
    report_path.write_text(
        json.dumps({k: v for k, v in result_data.items() if k != "output"}, indent=2),
        encoding="utf-8",
    )

    return result_data


def _copy_one(src_file: Path, dst_file: Path) -> None:
    dst_file.parent.mkdir(parents=True, exist_ok=True)
    dst_file.write_text(src_file.read_text(encoding="utf-8"), encoding="utf-8")


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--src", required=True, type=Path, help="Source models directory")
    parser.add_argument("--dst", required=True, type=Path, help="Destination directory (created if absent)")
    parser.add_argument("--from", dest="src_dialect", default="snowflake", help="Source SQL dialect (default: snowflake)")
    parser.add_argument("--to", dest="dst_dialect", default="databricks", help="Target SQL dialect (default: databricks)")
    parser.add_argument("--workers", type=int, default=os.cpu_count() or 4, help="Parallel workers (default: cpu_count)")
    args = parser.parse_args()

    if not args.src.exists():
        print(f"ERROR: source directory does not exist: {args.src}", file=sys.stderr)
        return 1

    args.dst.mkdir(parents=True, exist_ok=True)

    sql_files = list(args.src.rglob("*.sql"))
    yml_files = list(args.src.rglob("*.yml")) + list(args.src.rglob("*.yaml"))
    md_files = list(args.src.rglob("*.md"))

    total = succeeded = failed = 0
    failed_results: list[dict] = []

    with ProcessPoolExecutor(max_workers=args.workers) as executor:
        futures = {
            executor.submit(
                _transpile_one,
                src_file,
                args.dst / src_file.relative_to(args.src),
                args.src_dialect,
                args.dst_dialect,
            ): src_file
            for src_file in sql_files
        }

        for future in as_completed(futures):
            result_data = future.result()
            total += 1
            if result_data["success"]:
                succeeded += 1
            else:
                failed += 1
                failed_results.append(result_data)

    for src_file in yml_files + md_files:
        dst_file = args.dst / src_file.relative_to(args.src)
        _copy_one(src_file, dst_file)

    print(f"Transpiled {total} files: {succeeded} succeeded, {failed} failed")

    if failed:
        print("Failed files:")
        for r in failed_results:
            print(f"  {r['path']}: {'; '.join(r['errors'])}")
        return 1

    return 0


if __name__ == "__main__":
    sys.exit(main())