railiance-fabric/railiance_fabric/cli.py

from __future__ import annotations

import argparse
import sys
from pathlib import Path

from .graph import FabricGraph, build_graph
from .validation import validate_roots


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="railiance-fabric",
        description="Load and validate Railiance Fabric declarations.",
    )
    sub = parser.add_subparsers(dest="command", required=True)

    validate = sub.add_parser(
        "validate",
        help="Validate one or more repo roots or declaration files.",
    )
    validate.add_argument(
        "paths",
        nargs="+",
        type=Path,
        help="Repo root, fabric directory, or declaration YAML file.",
    )
    validate.add_argument(
        "--warnings-as-errors",
        action="store_true",
        help="Exit non-zero when warnings are present.",
    )

    providers = sub.add_parser("providers", help="List providers for a capability type or id.")
    providers.add_argument("capability", help="Capability type or capability id.")
    providers.add_argument("paths", nargs="*", type=Path, default=[Path(".")])

    consumers = sub.add_parser("consumers", help="List consumers of a capability or interface.")
    consumers.add_argument("target", help="Capability/interface type or declaration id.")
    consumers.add_argument("paths", nargs="*", type=Path, default=[Path(".")])

    dependency_path = sub.add_parser("dependency-path", help="Show dependency path for a service.")
    dependency_path.add_argument("service_id", help="Service declaration id.")
    dependency_path.add_argument("paths", nargs="*", type=Path, default=[Path(".")])

    unresolved = sub.add_parser("unresolved", help="Show missing or unresolved dependencies.")
    unresolved.add_argument("paths", nargs="*", type=Path, default=[Path(".")])

    blast = sub.add_parser("blast-radius", help="Show consumers affected by an interface change.")
    blast.add_argument("interface", help="Interface type or interface declaration id.")
    blast.add_argument("paths", nargs="*", type=Path, default=[Path(".")])

    export = sub.add_parser("export", help="Export graph as JSON or Mermaid.")
    export.add_argument("paths", nargs="*", type=Path, default=[Path(".")])
    export.add_argument("--format", choices=["json", "mermaid"], default="json")
    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)

    if args.command == "validate":
        report = validate_roots(args.paths)
        for diagnostic in report.diagnostics:
            print(diagnostic.format())
        print(report.summary())
        if report.errors:
            return 1
        if args.warnings_as_errors and report.warnings:
            return 1
        return 0

    if args.command == "providers":
        graph = _load_graph_or_exit(args.paths)
        _print_providers(graph, args.capability)
        return 0

    if args.command == "consumers":
        graph = _load_graph_or_exit(args.paths)
        _print_consumers(graph, args.target)
        return 0

    if args.command == "dependency-path":
        graph = _load_graph_or_exit(args.paths)
        print("\n".join(graph.dependency_path_lines(args.service_id)))
        return 0

    if args.command == "unresolved":
        graph = _load_graph_or_exit(args.paths)
        _print_unresolved(graph)
        return 0

    if args.command == "blast-radius":
        graph = _load_graph_or_exit(args.paths)
        _print_consumers(graph, args.interface, matches=graph.blast_radius(args.interface))
        return 0

    if args.command == "export":
        graph = _load_graph_or_exit(args.paths)
        print(graph.to_mermaid() if args.format == "mermaid" else graph.to_json())
        return 0

    parser.error(f"unknown command {args.command!r}")
    return 2

def _load_graph_or_exit(paths: list[Path]) -> FabricGraph:
    graph = build_graph(paths)
    if graph.load_errors:
        for path, message in graph.load_errors:
            print(f"ERROR {path}: {message}", file=sys.stderr)
        raise SystemExit(1)
    return graph


def _print_providers(graph: FabricGraph, capability: str) -> None:
    providers = graph.providers(capability)
    if not providers:
        print(f"no providers found for {capability}")
        return
    print("provider_id\tservice_id\tlifecycle\tenvironments\tinterfaces")
    for provider in providers:
        spec = provider.spec
        print(
            "\t".join(
                [
                    provider.id,
                    str(spec.get("service_id", "")),
                    str(spec.get("lifecycle", "")),
                    ",".join(spec.get("environments", [])),
                    ",".join(spec.get("interface_ids", [])),
                ]
            )
        )


def _print_consumers(
    graph: FabricGraph,
    target: str,
    matches: object | None = None,
) -> None:
    consumer_matches = graph.consumers(target) if matches is None else list(matches)
    if not consumer_matches:
        print(f"no consumers found for {target}")
        return
    print("consumer_service_id\tdependency_id\trequires\tprovider_capability_id\tprovider_interface_id\tstatus")
    for match in consumer_matches:
        print(
            "\t".join(
                [
                    match.consumer_service_id,
                    match.dependency_id,
                    match.required_capability_type,
                    match.provider_capability_id,
                    match.provider_interface_id,
                    match.status,
                ]
            )
        )


def _print_unresolved(graph: FabricGraph) -> None:
    unresolved = graph.unresolved_dependencies()
    if not unresolved:
        print("no unresolved dependencies")
        return
    print("dependency_id\tconsumer_service_id\trequires")
    for dependency in unresolved:
        spec = dependency.spec
        requires = spec.get("requires", {})
        print(
            "\t".join(
                [
                    dependency.id,
                    str(spec.get("consumer_service_id", "")),
                    str(requires.get("capability_id") or requires.get("capability_type", "")),
                ]
            )
        )


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