orchestrata_ultimate.tools.ast_query_cli

Reference › orchestrata_ultimate › tools › ast_query_cli

This module is the ast-query command line over ast_query, built with typer. No other module imports it, and tests/tools/test_ast_query_cli.py covers it.

The commands documented in its module docstring are summary, stubs, find, search, modules, largest, connected, inheritance, protocols, abstract, pydantic, dataclasses, enums, diff, unmapped, coverage and stats. Shared options select the snapshot path (default plans/ast-ultimate.json), an optional comparison file and JSON output. _get_ast builds the ASTQuery, _warn_if_stale prints a one-line staleness warning to standard error when the source tree is newer than the snapshot, and _output and _print_dict print results as JSON or as indented text. main runs the typer app, which prints help when called with no arguments.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/tools/ast_query_cli.py

"""
CLI for the AST Query Engine.

Usage::

    ast-query summary
    ast-query stubs
    ast-query find BaseAgent
    ast-query search 'Agent$'
    ast-query modules
    ast-query largest 10
    ast-query largest --top 10
    ast-query connected
    ast-query inheritance BaseAgent
    ast-query protocols
    ast-query abstract
    ast-query pydantic
    ast-query dataclasses
    ast-query enums
    ast-query diff BaseAgent
    ast-query unmapped
    ast-query coverage
    ast-query stats
"""

from __future__ import annotations

import json
import sys

import typer

from orchestrata_ultimate.tools.ast_query import ASTQuery

app = typer.Typer(
    name="ast-query",
    help="Query the serialized AST JSON for orchestrata-ultimate.",
    no_args_is_help=True,
)

# ── shared options ────────────────────────────────────────────────────

_ast_path_opt = typer.Option(
    "plans/ast-ultimate.json",
    "--ast",
    help="Path to ultimate AST JSON",
)

_compare_path_opt = typer.Option(
    None,
    "--compare",
    help="Path to deep-comparison JSON (enables diff/coverage/stubs-from-source)",
)

_json_opt = typer.Option(False, "--json", help="Output as JSON")


def _get_ast(ast_path: str, compare_path: str | None) -> ASTQuery:
    return ASTQuery(ast_path=ast_path, compare_path=compare_path)


def _warn_if_stale(ast: ASTQuery) -> None:
    """Print a one-line staleness WARNING to stderr if the snapshot is stale.

    The snapshot is stale when a ``.py`` file under its recorded source root
    has been modified since it was serialized. This only ever writes to
    stderr — stdout output (parsed by consumers) is never touched.
    """
    msg = ast.staleness_warning()
    if msg:
        print(msg, file=sys.stderr)


def _output(data: object, as_json: bool = False) -> None:
    if as_json:
        # Convert ClassInfo objects to dicts
        print(
            json.dumps(
                data,
                indent=2,
                default=lambda o: o.__dict__ if hasattr(o, "__dict__") else str(o),
            )
        )
    elif isinstance(data, list):
        for item in data:
            if isinstance(item, dict):
                _print_dict(item)
            else:
                print(item)
    elif isinstance(data, dict):
        _print_dict(data)
    else:
        print(data)


def _print_dict(d: dict, indent: int = 0) -> None:
    prefix = "  " * indent
    for k, v in d.items():
        if isinstance(v, dict):
            print(f"{prefix}{k}:")
            _print_dict(v, indent + 1)
        elif isinstance(v, list) and len(v) > 10:
            print(f"{prefix}{k}: [{len(v)} items]")
            for item in v[:5]:
                print(f"{prefix}  - {item}")
            print(f"{prefix}  ... and {len(v) - 5} more")
        elif isinstance(v, list):
            print(f"{prefix}{k}: {v}")
        else:
            print(f"{prefix}{k}: {v}")


# ── commands ──────────────────────────────────────────────────────────


@app.command()
def summary(
    ast_path: str = _ast_path_opt,
    json_output: bool = _json_opt,
) -> None:
    """Print top-level project summary."""
    ast = _get_ast(ast_path, None)
    _warn_if_stale(ast)
    _output(ast.summary(), json_output)


@app.command()
def modules(
    ast_path: str = _ast_path_opt,
    json_output: bool = _json_opt,
) -> None:
    """List all modules with LOC, class, and function counts."""
    ast = _get_ast(ast_path, None)
    _warn_if_stale(ast)
    _output(ast.module_summary(), json_output)


@app.command()
def stubs(
    ast_path: str = _ast_path_opt,
    compare_path: str | None = _compare_path_opt,
    min_methods: int = typer.Option(
        3, "--min-methods", help="Minimum methods in source to consider"
    ),
    max_ratio: float = typer.Option(
        0.3, "--max-ratio", help="Maximum method ratio to flag as stub"
    ),
    json_output: bool = _json_opt,
) -> None:
    """List stub classes (<30% methods ported)."""
    ast = _get_ast(ast_path, compare_path)
    result = ast.stubs(min_methods_in_source=min_methods, max_ratio=max_ratio)
    if not result:
        print("No stubs found.")
        return
    _output(result, json_output)


@app.command()
def find(
    name: str = typer.Argument(..., help="Class name to find"),
    ast_path: str = _ast_path_opt,
    compare_path: str | None = _compare_path_opt,
    json_output: bool = _json_opt,
) -> None:
    """Find a class by name."""
    ast = _get_ast(ast_path, compare_path)
    result = ast.find_class(name)
    if result is None:
        print(f"Class '{name}' not found.")
        raise typer.Exit(1)
    _output(result, json_output)


@app.command()
def search(
    pattern: str = typer.Argument(..., help="Regex pattern to search class names"),
    ast_path: str = _ast_path_opt,
    case_sensitive: bool = typer.Option(False, "--case", help="Case-sensitive match"),
    json_output: bool = _json_opt,
) -> None:
    """Search classes by regex pattern."""
    ast = _get_ast(ast_path, None)
    results = ast.search_classes(pattern, case_sensitive=case_sensitive)
    if not results:
        print(f"No classes matching '{pattern}'.")
        return
    for ci in results:
        print(f"  {ci.name:40s}  {ci.module_path:50s}  {ci.method_count} methods")
    if json_output:
        _output([ci.__dict__ for ci in results], True)


@app.command()
def largest(
    n: int = typer.Argument(20, help="Number of classes to show"),
    top: int | None = typer.Option(
        None, "--top", help="Number of classes to show (overrides positional N)"
    ),
    ast_path: str = _ast_path_opt,
    json_output: bool = _json_opt,
) -> None:
    """Show largest classes by method count."""
    count = top if top is not None else n
    ast = _get_ast(ast_path, None)
    results = ast.largest_classes(count)
    for r in results:
        print(
            f"  {r['name']:40s}  {r['module']:50s}  {r['methods']} methods  (line {r.get('line', '?')})"
        )
    if json_output:
        _output(results, True)


@app.command()
def connected(
    n: int = typer.Argument(20, help="Number of classes to show"),
    top: int | None = typer.Option(
        None, "--top", help="Number of classes to show (overrides positional N)"
    ),
    ast_path: str = _ast_path_opt,
    json_output: bool = _json_opt,
) -> None:
    """Show classes with the most subclasses."""
    count = top if top is not None else n
    ast = _get_ast(ast_path, None)
    results = ast.most_connected_classes(count)
    for r in results:
        print(f"  {r['class']:60s}  {r['subclasses']} subclasses")
    if json_output:
        _output(results, True)


@app.command()
def inheritance(
    class_name: str = typer.Argument(..., help="Class name to trace"),
    ast_path: str = _ast_path_opt,
    json_output: bool = _json_opt,
) -> None:
    """Show inheritance tree for a class."""
    ast = _get_ast(ast_path, None)
    result = ast.inheritance_tree(class_name)
    if result is None:
        print(f"Class '{class_name}' not found in inheritance graph.")
        raise typer.Exit(1)
    _output(result, json_output)


@app.command()
def protocols(
    ast_path: str = _ast_path_opt,
    json_output: bool = _json_opt,
) -> None:
    """List all Protocol classes."""
    ast = _get_ast(ast_path, None)
    results = ast.protocols()
    for ci in results:
        print(f"  {ci.name:40s}  {ci.module_path}")
    print(f"\n  Total: {len(results)} Protocol classes")
    if json_output:
        _output([ci.__dict__ for ci in results], True)


@app.command()
def abstract(
    ast_path: str = _ast_path_opt,
    json_output: bool = _json_opt,
) -> None:
    """List all abstract classes."""
    ast = _get_ast(ast_path, None)
    results = ast.abstract_classes()
    for ci in results:
        bases = ", ".join(ci.base_classes[:3]) if ci.base_classes else "none"
        print(f"  {ci.name:40s}  {ci.module_path:50s}  extends: {bases}")
    print(f"\n  Total: {len(results)} abstract classes")
    if json_output:
        _output([ci.__dict__ for ci in results], True)


@app.command()
def pydantic(
    ast_path: str = _ast_path_opt,
    json_output: bool = _json_opt,
) -> None:
    """List all Pydantic BaseModel classes."""
    ast = _get_ast(ast_path, None)
    results = ast.pydantic_models()
    for ci in results:
        print(f"  {ci.name:40s}  {ci.module_path}")
    print(f"\n  Total: {len(results)} Pydantic models")
    if json_output:
        _output([ci.__dict__ for ci in results], True)


@app.command()
def dataclasses(
    ast_path: str = _ast_path_opt,
    json_output: bool = _json_opt,
) -> None:
    """List all dataclass classes."""
    ast = _get_ast(ast_path, None)
    results = ast.dataclasses()
    for ci in results:
        print(f"  {ci.name:40s}  {ci.module_path}")
    print(f"\n  Total: {len(results)} dataclasses")
    if json_output:
        _output([ci.__dict__ for ci in results], True)


@app.command()
def enums(
    ast_path: str = _ast_path_opt,
    json_output: bool = _json_opt,
) -> None:
    """List all Enum classes."""
    ast = _get_ast(ast_path, None)
    results = ast.enums()
    for ci in results:
        print(f"  {ci.name:40s}  {ci.module_path}")
    print(f"\n  Total: {len(results)} Enum classes")
    if json_output:
        _output([ci.__dict__ for ci in results], True)


@app.command()
def interfaces(
    ast_path: str = _ast_path_opt,
    json_output: bool = _json_opt,
) -> None:
    """List all Interface declarations (JS/TS)."""
    ast = _get_ast(ast_path, None)
    results = ast.interfaces()
    for ci in results:
        print(f"  {ci.name:40s}  {ci.module_path}")
    print(f"\n  Total: {len(results)} interfaces")
    if json_output:
        _output([ci.__dict__ for ci in results], True)


@app.command()
def type_aliases(
    ast_path: str = _ast_path_opt,
    json_output: bool = _json_opt,
) -> None:
    """List all Type alias declarations (JS/TS)."""
    ast = _get_ast(ast_path, None)
    results = ast.type_aliases()
    for ci in results:
        print(f"  {ci.name:40s}  {ci.module_path}")
    print(f"\n  Total: {len(results)} type aliases")
    if json_output:
        _output([ci.__dict__ for ci in results], True)


@app.command()
def exported(
    ast_path: str = _ast_path_opt,
    json_output: bool = _json_opt,
) -> None:
    """List all exported classes/interfaces (JS/TS)."""
    ast = _get_ast(ast_path, None)
    results = ast.exported_classes()
    for ci in results:
        print(f"  {ci.name:40s}  {ci.module_path}")
    print(f"\n  Total: {len(results)} exported declarations")
    if json_output:
        _output([ci.__dict__ for ci in results], True)


@app.command()
def diff(
    class_name: str = typer.Argument(..., help="Class name to diff"),
    ast_path: str = _ast_path_opt,
    compare_path: str = typer.Option(
        ..., "--compare", help="Path to deep-comparison JSON (required)"
    ),
    json_output: bool = _json_opt,
) -> None:
    """Show method-level diff for a class (requires --compare)."""
    ast = _get_ast(ast_path, compare_path)
    result = ast.method_diff(class_name)
    if result is None:
        print(f"No comparison data for '{class_name}'.")
        raise typer.Exit(1)
    _output(result, json_output)


@app.command()
def unmapped(
    ast_path: str = _ast_path_opt,
    compare_path: str = typer.Option(
        ..., "--compare", help="Path to deep-comparison JSON (required)"
    ),
    json_output: bool = _json_opt,
) -> None:
    """List unported orchestrata classes and ultimate-only classes (requires --compare)."""
    ast = _get_ast(ast_path, compare_path)
    result = ast.unmapped_classes()
    if not result:
        print("No comparison data available.")
        return
    orchestrata_only = result.get("orchestrata_only", [])
    ultimate_only = result.get("ultimate_only", [])
    print(f"Orchestrata-only (not yet ported): {len(orchestrata_only)}")
    for c in orchestrata_only[:30]:
        print(f"  - {c}")
    if len(orchestrata_only) > 30:
        print(f"  ... and {len(orchestrata_only) - 30} more")
    print(f"\nUltimate-only (new additions): {len(ultimate_only)}")
    for c in ultimate_only[:30]:
        print(f"  - {c}")
    if len(ultimate_only) > 30:
        print(f"  ... and {len(ultimate_only) - 30} more")
    if json_output:
        _output(result, True)


@app.command()
def coverage(
    ast_path: str = _ast_path_opt,
    compare_path: str = typer.Option(
        ..., "--compare", help="Path to deep-comparison JSON (required)"
    ),
    json_output: bool = _json_opt,
) -> None:
    """Show porting coverage by tier (requires --compare)."""
    ast = _get_ast(ast_path, compare_path)
    result = ast.coverage_by_tier()
    if not result:
        print("No tier coverage data. Does the comparison file contain 'tier_summary'?")
        return
    print(
        f"{'Tier':<6} {'Packages':<10} {'Orch LOC':<12} {'Ult LOC':<12} {'LOC %':<8} {'Class %':<8}"
    )
    print("-" * 58)
    for tier, data in sorted(result.items()):
        print(
            f"  {tier:<4}  {data.get('packages', 0):<8}  "
            f"{data.get('orchestrata_loc', 0):<10}  "
            f"{data.get('ultimate_loc', 0):<10}  "
            f"{data.get('coverage', 0):<6.1f}  "
            f"{data.get('class_coverage', 0):<6.1f}"
        )
    if json_output:
        _output(result, True)


@app.command()
def stats(
    ast_path: str = _ast_path_opt,
    json_output: bool = _json_opt,
) -> None:
    """Show internal index statistics."""
    ast = _get_ast(ast_path, None)
    _warn_if_stale(ast)
    _output(ast.index_stats(), json_output)


def main() -> None:
    app()


if __name__ == "__main__":
    main()