scripts.ast_api

Reference › scripts › ast_api

A FastAPI application that exposes the AST toolkit over HTTP, with documentation generated automatically from the route signatures. It is a service rather than a command-line script: it defines no main, and the resident ast_daemon loads it as a module (_load_ast_api) and serves its routes. No tests import it directly.

Each route wraps one toolkit command and takes a source query parameter, a comma-separated list of directories or files that _sources splits. The routes are healthz, symbols, refs, source, unused, complexity, nesting, audit, branches, who_imports, untraced, check_deps, interface, docs_md, diagram, long_params, cycles, call_graph, dead_params, untyped, gen_tests, gof_pattern and gen_dto. The numeric ones take thresholds: complexity 10, nesting 4, long parameters 5. Responses use pydantic models (Symbol, ComplexityRow, NestingRow, LongParamRow, Violation, Health) where the shape is fixed. _load_tool imports the toolkit script lazily, so the module itself stays light. All routes are read-only analysis; none writes to the analysed tree.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Classes

Symbol (pydantic)

Bases: BaseModel · Line 60

ComplexityRow (pydantic)

Bases: BaseModel · Line 68

NestingRow (pydantic)

Bases: BaseModel · Line 75

LongParamRow (pydantic)

Bases: BaseModel · Line 82

Violation (pydantic)

Bases: BaseModel · Line 89

Health (pydantic)

Bases: BaseModel · Line 96

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/ast_api.py

"""FastAPI service exposing the AST toolkit over HTTP — with auto-generated
OpenAPI/Swagger docs.

The read-only analysis + generation capabilities of
``scripts/serialize_codebase_ast.py`` are surfaced as JSON/text endpoints so
they can be driven by other tools, dashboards, or CI. Mutating operations
(``rename --apply``) are intentionally NOT exposed.

Run:
    uv run uvicorn ast_api:app --reload      # from the scripts/ directory
Then open:
    /docs          interactive Swagger UI
    /redoc         ReDoc
    /openapi.json  the OpenAPI 3 schema (the "Swagger")

All endpoints take a ``source`` query param (comma-separated dirs/files,
default ``src/orchestrata_ultimate``) so callers can scope the analysis.
"""

from __future__ import annotations

import importlib.util
from pathlib import Path
from typing import Literal

from fastapi import FastAPI
from fastapi.responses import PlainTextResponse
from pydantic import BaseModel

_TOOL_PATH = Path(__file__).resolve().parent / "serialize_codebase_ast.py"


def _load_tool():
    spec = importlib.util.spec_from_file_location("serialize_codebase_ast", _TOOL_PATH)
    if spec is None or spec.loader is None:
        raise ImportError(f"cannot load AST toolkit from {_TOOL_PATH}")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


_tool = _load_tool()
_DEFAULT_SOURCE = "src/orchestrata_ultimate"

# Module-level Literal aliases (a value assignment, NOT an inline annotation):
# keeping the quoted members here stops the formatter from "unquoting" them under
# `from __future__ import annotations`, which would turn Literal["singleton"] into
# the undefined bare name `singleton`.
PatternName = Literal["singleton", "factory", "observer", "strategy", "decorator"]
DtoKind = Literal["dataclass", "pydantic"]


def _sources(source: str) -> list[str]:
    return [s.strip() for s in source.split(",") if s.strip()]


# --- response models (drive a rich, typed OpenAPI/Swagger schema) ---


class Symbol(BaseModel):
    kind: str
    name: str
    signature: str
    file: str
    line: int


class ComplexityRow(BaseModel):
    name: str
    complexity: int
    file: str
    line: int


class NestingRow(BaseModel):
    name: str
    depth: int
    file: str
    line: int


class LongParamRow(BaseModel):
    name: str
    params: int
    file: str
    line: int


class Violation(BaseModel):
    file: str
    line: int
    rule: str
    detail: str


class Health(BaseModel):
    status: str
    tool: str


app = FastAPI(
    title="Orchestrata AST Toolkit API",
    version="1.0.0",
    description=(
        "Query the codebase AST on live source: symbols, references, "
        "cyclomatic complexity, nesting depth, long parameter lists, import "
        "cycles, dead parameters, missing type hints, Key-Rule compliance, "
        "logging/exception audit, undeclared dependencies, plus interface, "
        "API-doc, Mermaid, branch-test, DTO, and GoF-pattern generation."
    ),
)


@app.get("/healthz", tags=["meta"], response_model=Health)
def healthz() -> Health:
    """Liveness probe."""
    return Health(status="ok", tool=str(_TOOL_PATH.name))


@app.get("/symbols", tags=["navigate"], response_model=list[Symbol])
def symbols(
    source: str = _DEFAULT_SOURCE, pattern: str | None = None, kind: str | None = None
):
    """List classes/functions/methods with signatures."""
    return _tool.list_symbols(_sources(source), pattern=pattern, kind=kind)


@app.get("/refs", tags=["navigate"])
def refs(name: str, source: str = _DEFAULT_SOURCE) -> dict:
    """Every reference to *name*, split into definitions/uses/imports."""
    return _tool.find_references(name, _sources(source))


@app.get("/source", tags=["navigate"])
def source(name: str, source: str = _DEFAULT_SOURCE, residue: bool = False) -> dict:
    """Resolve *name* (qualified symbol or module) to its exact source bytes
    — the same payload the CLI ``source`` subcommand prints, unchanged, so a
    daemon-delegated answer is byte-identical to a local one."""
    return _tool.resolve_source_payload(name, _sources(source), residue=residue)


@app.get("/unused", tags=["health"])
def unused(source: str = _DEFAULT_SOURCE) -> list[dict]:
    """Module-level functions/classes never referenced."""
    return _tool.find_unused(_sources(source))


@app.get("/complexity", tags=["health"], response_model=list[ComplexityRow])
def complexity(source: str = _DEFAULT_SOURCE, threshold: int = 10):
    """Cyclomatic complexity per function, descending."""
    return _tool.measure_complexity(_sources(source), threshold=threshold)


@app.get("/nesting", tags=["health"], response_model=list[NestingRow])
def nesting(source: str = _DEFAULT_SOURCE, threshold: int = 4):
    """Max nesting depth per function, descending."""
    return _tool.measure_nesting(_sources(source), threshold=threshold)


@app.get("/audit", tags=["health"], response_model=list[Violation])
def audit(source: str = _DEFAULT_SOURCE):
    """Logging + exception-handling findings (advisory)."""
    return _tool.audit_error_handling(_sources(source))


@app.get("/branches", tags=["navigate"])
def branches(source: str = _DEFAULT_SOURCE, name: str | None = None) -> list[dict]:
    """Decision points (branches) per function."""
    return _tool.list_branches(name, _sources(source))


@app.get("/who-imports", tags=["navigate"])
def who_imports(target: str, source: str = _DEFAULT_SOURCE) -> list[dict]:
    """Reverse import graph: files that import a module path or symbol."""
    return _tool.who_imports(target, _sources(source))


@app.get("/untraced", tags=["health"])
def untraced(source: str = _DEFAULT_SOURCE) -> list[dict]:
    """Public functions/methods lacking an @traces decorator."""
    return _tool.find_untraced(_sources(source))


@app.get("/check-deps", tags=["health"])
def check_deps(
    source: str = _DEFAULT_SOURCE,
    pyproject: str = "pyproject.toml",
    include_tests: bool = False,
) -> dict:
    """Third-party imports not declared in pyproject."""
    srcs = _sources(source)
    # Relative paths resolve against the served checkout when the toolkit has
    # one configured (the resident daemon), else against the cwd as before.
    pyproject_path = _tool._locate(pyproject)
    repo_root = pyproject_path.resolve().parent
    if include_tests:
        srcs = [*srcs, str(repo_root / "tests")]
    return _tool.find_undeclared_dependencies(
        [_tool._locate(s) for s in srcs], pyproject_path, repo_root
    )


@app.get("/interface/{class_name}", tags=["generate"], response_class=PlainTextResponse)
def interface(class_name: str, source: str = _DEFAULT_SOURCE) -> str:
    """Generate a typing.Protocol for a class's public API."""
    code = _tool.generate_interface(class_name, _sources(source))
    return code or f"# class {class_name!r} not found"


@app.get("/docs-md", tags=["generate"], response_class=PlainTextResponse)
def docs_md(source: str = _DEFAULT_SOURCE) -> str:
    """Generate a markdown API reference."""
    return _tool.generate_docs(_sources(source))


@app.get("/diagram", tags=["generate"], response_class=PlainTextResponse)
def diagram(source: str = _DEFAULT_SOURCE, kind: str = "class") -> str:
    """Generate a Mermaid diagram (class inheritance or module deps)."""
    return _tool.generate_mermaid(_sources(source), kind=kind)


@app.get("/long-params", tags=["health"], response_model=list[LongParamRow])
def long_params(source: str = _DEFAULT_SOURCE, threshold: int = 5):
    """Functions/methods whose parameter count (excl. self/cls) exceeds *threshold*."""
    return _tool.find_long_param_lists(_sources(source), threshold=threshold)


@app.get("/cycles", tags=["health"])
def cycles(source: str = _DEFAULT_SOURCE) -> list[list[str]]:
    """First-party import cycles, one closed chain per cycle."""
    return _tool.find_import_cycles(_sources(source))


@app.get("/call-graph/{func_name}", tags=["navigate"])
def call_graph(func_name: str, source: str = _DEFAULT_SOURCE) -> dict | None:
    """Callers and callees of a function (forward + reverse), type-resolved.

    ``None`` (200, not 404 — mirrors ``/interface``'s not-found convention of
    a still-successful response) when *func_name* isn't found under *source*.
    """
    return _tool.build_call_graph(func_name, _sources(source))


@app.get("/dead-params", tags=["health"])
def dead_params(source: str = _DEFAULT_SOURCE) -> list[dict]:
    """Parameters never referenced in their body (contract params skipped)."""
    return _tool.find_dead_params(_sources(source))


@app.get("/untyped", tags=["health"])
def untyped(source: str = _DEFAULT_SOURCE) -> list[dict]:
    """Public functions/methods missing parameter or return type hints."""
    return _tool.find_untyped(_sources(source))


@app.get("/gen-tests/{func_name}", tags=["generate"], response_class=PlainTextResponse)
def gen_tests(func_name: str, source: str = _DEFAULT_SOURCE) -> str:
    """Generate runnable pytest tests covering each branch of a function."""
    return _tool.generate_branch_tests(func_name, _sources(source))


@app.get("/pattern/{class_name}", tags=["generate"], response_class=PlainTextResponse)
def gof_pattern(
    class_name: str,
    pattern: PatternName,
    source: str = _DEFAULT_SOURCE,
) -> str:
    """Scaffold a Gang-of-Four pattern around *class_name* (reviewable code)."""
    code = _tool.generate_pattern(class_name, pattern, _sources(source))
    return code or f"# class {class_name!r} not found"


@app.get("/gen-dto/{func_name}", tags=["generate"], response_class=PlainTextResponse)
def gen_dto(
    func_name: str,
    source: str = _DEFAULT_SOURCE,
    kind: DtoKind = "dataclass",
) -> str:
    """Generate a DTO (dataclass or pydantic model) from a function's parameters."""
    code = _tool.generate_dto(func_name, _sources(source), kind=kind)
    return code or f"# function {func_name!r} not found"