scripts/gen_routing_pb2.py

Rama main · réplica de lectura del repositorio original.

"""Generate the routing API protobuf bindings from its ``routing.proto``.

The routing API ships a protobuf schema
(``omnigent/api/routing/v1/routing.proto``); the runtime imports the generated
``omnigent.api.routing.v1.routing_pb2`` module. Generated
code is checked in (so a plain ``pip install`` / editor / type checker sees it without a
build step), and this script is the one blessed way to regenerate it — run it
whenever ``routing.proto`` changes and commit the result.

It shells out to ``grpc_tools.protoc`` (the ``grpcio-tools`` dev dependency),
which bundles both the compiler and the well-known-type protos, so the
``import "google/protobuf/struct.proto"`` in the schema resolves with no system
``protoc`` install.

Pass ``--check`` to verify the committed bindings are up to date without
writing: it regenerates into a temp dir, compares, and exits non-zero (printing
the drift) if ``routing.proto`` was edited without regenerating. The
``routing-pb2-fresh`` pre-commit hook runs it in this mode.

Usage::

    python scripts/gen_routing_pb2.py            # (re)generate in place
    python scripts/gen_routing_pb2.py --check     # verify freshness (CI / hook)
"""

from __future__ import annotations

import argparse
import filecmp
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

# Repo root (this file lives in ``scripts/``). ``--proto_path`` is rooted here so
# the generated package path is ``omnigent/api/routing/v1/routing_pb2`` —
# matching the import the runtime uses (and the proto's ``package`` line).
_REPO_ROOT = Path(__file__).resolve().parent.parent
_PROTO = Path("omnigent/api/routing/v1/routing.proto")
# The two files protoc emits for the schema, relative to the repo root.
_OUTPUTS = (
    Path("omnigent/api/routing/v1/routing_pb2.py"),
    Path("omnigent/api/routing/v1/routing_pb2.pyi"),
)


def _run_protoc(out_dir: Path) -> None:
    """Compile ``routing.proto`` into ``out_dir`` (rooted like the repo)."""
    cmd = [
        sys.executable,
        "-m",
        "grpc_tools.protoc",
        f"--proto_path={_REPO_ROOT}",
        f"--python_out={out_dir}",
        f"--pyi_out={out_dir}",
        str(_PROTO),
    ]
    try:
        subprocess.run(cmd, check=True, cwd=_REPO_ROOT)
    except FileNotFoundError:
        raise SystemExit(
            "grpc_tools.protoc not found. Install the dev dependencies first:\n"
            "  uv sync --group lint"
        ) from None


def _generate() -> int:
    _run_protoc(_REPO_ROOT)
    print(f"Generated {', '.join(str(p) for p in _OUTPUTS)}")
    return 0


def _check() -> int:
    with tempfile.TemporaryDirectory() as tmp:
        tmp_dir = Path(tmp)
        _run_protoc(tmp_dir)
        stale = [
            p
            for p in _OUTPUTS
            if not (tmp_dir / p).exists()
            or not (_REPO_ROOT / p).exists()
            or not filecmp.cmp(tmp_dir / p, _REPO_ROOT / p, shallow=False)
        ]
    if stale:
        names = ", ".join(str(p) for p in stale)
        print(
            f"Routing protobuf bindings are out of date: {names}\n"
            "Regenerate and commit them:\n"
            "  python scripts/gen_routing_pb2.py",
            file=sys.stderr,
        )
        return 1
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--check",
        action="store_true",
        help="verify the committed bindings match the schema without writing",
    )
    args = parser.parse_args()
    if shutil.which(sys.executable) is None:  # pragma: no cover - defensive
        raise SystemExit(f"python interpreter not found: {sys.executable}")
    return _check() if args.check else _generate()


if __name__ == "__main__":
    raise SystemExit(main())