scripts.graph_browser

Reference › scripts › graph_browser

scripts/graph_browser.py is an interactive graph browser in the style of Neo4j Browser for the project's graphs. A human starts it; it is a FastAPI app served by uvicorn. main takes --host (default 127.0.0.1) and --port (default 8090), and runs uvicorn with the h11 HTTP implementation, because the default auto mode imports httptools, which re-enables the GIL process-wide (SPEC-023); ast_daemon carries the same pin. See also the graph browser page.

The app serves one self-contained HTML page at / (index) and a JSON API: api_meta returns node labels and relationship types with counts and assigned colours, api_nodes lists nodes with an optional substring filter on the display name, and api_expand returns a node's neighbours and relationships, capped by limit. Each request names a graph kind; _normalize_graph_kind folds the deprecated neo4j alias onto kg, _resolve_memgraph_uri and _get_driver obtain a driver through orchestrata_ultimate.core.graph_factory, and _open_session binds a session to the database the graph kind names. _jsonable and _trim_props coerce Bolt property values to JSON-safe, shortened values. The script is read-only against the graph.

At a glance

Class diagram

Dependency diagram

API

Classes

LabelMeta (pydantic)

Bases: BaseModel · Line 186

RelMeta (pydantic)

Bases: BaseModel · Line 192

MetaResponse (pydantic)

Bases: BaseModel · Line 197

NodeSummary (pydantic)

Bases: BaseModel · Line 202

EdgeSummary (pydantic)

Bases: BaseModel · Line 209

ExpandResponse (pydantic)

Bases: BaseModel · Line 216

Functions

Source

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

"""Interactive graph browser (Neo4j-Browser-style) for the project's graphs.

Serves a self-contained d3-force HTML app plus a small read-only JSON API
over the Memgraph semantic graph (default) or the operational knowledge graph
(``?graph=kg``). Connects via ``core.graph_factory`` only -- direct
``GraphDatabase.driver`` calls are banned by ``scripts/architecture_checker.py``.

``?graph=neo4j`` is accepted as a DEPRECATED alias for ``?graph=kg`` (the
knowledge graph this browser reads has run on the orchgraph/Memgraph cluster
since the 2026-07-27 Neo4j decommission; the query-param name is kept working
so existing bookmarks/links do not break).

Run:
    uv run python scripts/graph_browser.py [--host 127.0.0.1] [--port 8090]
Then open http://127.0.0.1:8090/ in a browser.
"""

from __future__ import annotations

import argparse
import os
from collections.abc import Mapping
from pathlib import Path
from typing import Any, Literal

import httpx
from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

from orchestrata_ultimate.core.graph_factory import (
    NO_AUTH,
    create_graph,
    resolve_kg_database,
)

_HERE = Path(__file__).resolve().parent
_STATIC_DIR = _HERE / "static"
_HTML_PATH = _STATIC_DIR / "graph_browser.html"

GraphKind = Literal["memgraph", "kg", "neo4j"]


def _normalize_graph_kind(graph: GraphKind) -> Literal["memgraph", "kg"]:
    """Fold the deprecated ``neo4j`` alias onto the canonical ``kg`` value."""
    return "kg" if graph == "neo4j" else graph


# A stable categorical palette (Tableau10-ish) assigned to labels in
# first-seen/alphabetical order so colors are consistent across requests.
_PALETTE = [
    "#4e79a7",
    "#f28e2b",
    "#e15759",
    "#76b7b2",
    "#59a14f",
    "#edc948",
    "#b07aa1",
    "#ff9da7",
    "#9c755f",
    "#bab0ac",
]

# Display-name property candidates, in priority order.
_NAME_PROPS = ("name", "title", "id", "path", "key", "label")

# Properties considered "key" for the compact node summary returned by
# /api/nodes and /api/expand (kept small -- full properties live behind the
# node's own detail, but we return everything we have since payloads are tiny).
_MAX_PROPS = 12


def _resolve_memgraph_uri(env: Mapping[str, str] | None = None) -> str:
    """Discover the current Memgraph MAIN via the HA controller, falling
    back to the first URI in MEMGRAPH_URIS, then MEMGRAPH_URI."""
    env_vars = os.environ if env is None else env
    controller_url = env_vars.get("MEMGRAPH_HA_CONTROLLER_URL")
    if controller_url:
        try:
            resp = httpx.get(f"{controller_url.rstrip('/')}/main", timeout=3.0)
            resp.raise_for_status()
            data = resp.json()
            uri = data.get("uri")
            if uri:
                return str(uri)
        except httpx.HTTPError, ValueError:
            pass
    uris_env = env_vars.get("MEMGRAPH_URIS")
    if uris_env:
        first = uris_env.split(",")[0].strip()
        if first:
            return first
    uri = env_vars.get("MEMGRAPH_URI")
    if not uri:
        # No literal fallback (RUL-003). `bolt://localhost:7687` was worse
        # than useless here: nothing in this project runs Memgraph on
        # localhost (the graphs are Incus containers on 10.53.1.x), so the
        # default could only ever turn a missing-config bug into a confusing
        # connection error against a host that was never the target.
        raise RuntimeError(
            "cannot resolve a Memgraph URI: none of MEMGRAPH_HA_CONTROLLER_URL, "
            "MEMGRAPH_URIS or MEMGRAPH_URI is set in the environment"
        )
    return uri


def _get_driver(graph: GraphKind) -> Any:
    if _normalize_graph_kind(graph) == "kg":
        return create_graph()
    return create_graph(uri=_resolve_memgraph_uri(), auth=NO_AUTH)


def _open_session(driver: Any, graph: GraphKind) -> Any:
    """Open a session bound to the database *graph* actually names.

    Since the 2026-07-26 cutover both graphs live on the ONE orchgraph
    cluster, so the URI no longer distinguishes them — the ``database=``
    argument does. A bare ``driver.session()`` lands on the cluster default
    (the shared semantic graph), which is how this browser silently served
    ``Chunk``/``ADR``/``Page`` counts when asked for the operational KG.
    ``?graph=kg`` (or its deprecated ``?graph=neo4j`` alias) therefore has to
    name ``KG_DATABASE`` explicitly; ``?graph=memgraph`` keeps the default,
    which IS the semantic graph.
    """
    if _normalize_graph_kind(graph) == "kg":
        return driver.session(database=resolve_kg_database())
    return driver.session()


def _known_labels(driver: Any, graph: GraphKind) -> set[str]:
    """The set of node labels actually present in the database.

    Used to validate any client-supplied label BEFORE it is interpolated
    into a Cypher query string (labels cannot be parameterized in Cypher).
    """
    with _open_session(driver, graph) as session:
        rows = session.run("MATCH (n) UNWIND labels(n) AS label RETURN DISTINCT label")
        return {row["label"] for row in rows}


def _display_name(props: dict[str, Any], node_id: int, labels: list[str]) -> str:
    for key in _NAME_PROPS:
        val = props.get(key)
        if isinstance(val, str) and val:
            return val
    label = labels[0] if labels else "Node"
    return f"{label}#{node_id}"


def _jsonable(value: Any) -> Any:
    """Coerce a Bolt property value into something the JSON serializer accepts.

    The driver hands back native temporal/spatial objects (``neo4j.time.DateTime``,
    ``Date``, ``Time``, ``Duration``, ``neo4j.spatial.Point``) for temporal and
    point properties. Pydantic cannot serialize those and raises
    ``PydanticSerializationError: Unable to serialize unknown type`` — so any
    node carrying a timestamp used to blow up ``/api/nodes`` and ``/api/expand``
    with a 500. This browser is a read-only viewer: the ISO string (or the
    object's own repr for the exotic types) is exactly what should be shown.
    """
    if isinstance(value, (str, int, float, bool)) or value is None:
        return value
    if isinstance(value, (bytes, bytearray)):
        return value.hex()
    if isinstance(value, (list, tuple)):
        return [_jsonable(v) for v in value]
    if isinstance(value, dict):
        return {str(k): _jsonable(v) for k, v in value.items()}
    for method in ("isoformat", "iso_format"):
        fn = getattr(value, method, None)
        if callable(fn):
            return fn()
    return str(value)


def _trim_props(props: dict[str, Any]) -> dict[str, Any]:
    trimmed = {}
    for i, (k, v) in enumerate(props.items()):
        if i >= _MAX_PROPS:
            break
        trimmed[k] = _jsonable(v)
    return trimmed


class LabelMeta(BaseModel):
    label: str
    count: int
    color: str


class RelMeta(BaseModel):
    type: str
    count: int


class MetaResponse(BaseModel):
    labels: list[LabelMeta]
    relationships: list[RelMeta]


class NodeSummary(BaseModel):
    id: int
    labels: list[str]
    name: str
    properties: dict[str, Any]


class EdgeSummary(BaseModel):
    id: int
    source: int
    target: int
    type: str


class ExpandResponse(BaseModel):
    nodes: list[NodeSummary]
    edges: list[EdgeSummary]
    truncated: bool


app = FastAPI(
    title="Graph Browser", description="Interactive graph explorer for Memgraph/Neo4j"
)
if _STATIC_DIR.is_dir():
    app.mount("/static", StaticFiles(directory=str(_STATIC_DIR)), name="static")


@app.get("/", response_class=HTMLResponse)
def index() -> str:
    """Serve the self-contained graph-browser HTML app."""
    if not _HTML_PATH.is_file():
        raise HTTPException(status_code=500, detail=f"missing {_HTML_PATH}")
    return _HTML_PATH.read_text(encoding="utf-8")


@app.get("/api/meta", response_model=MetaResponse)
def api_meta(
    graph: GraphKind = Query(
        "memgraph",
        description="'memgraph' or 'kg' ('neo4j' accepted as a deprecated alias for 'kg')",
    ),
) -> MetaResponse:
    """Node labels with counts + assigned colors, relationship types with counts."""
    driver = _get_driver(graph)
    try:
        with _open_session(driver, graph) as session:
            label_rows = list(
                session.run(
                    "MATCH (n) UNWIND labels(n) AS label "
                    "RETURN label, count(*) AS count ORDER BY label"
                )
            )
            rel_rows = list(
                session.run(
                    "MATCH ()-[r]->() RETURN type(r) AS type, count(*) AS count ORDER BY type"
                )
            )
    finally:
        driver.close()

    labels = [
        LabelMeta(
            label=row["label"], count=row["count"], color=_PALETTE[i % len(_PALETTE)]
        )
        for i, row in enumerate(label_rows)
    ]
    relationships = [RelMeta(type=row["type"], count=row["count"]) for row in rel_rows]
    return MetaResponse(labels=labels, relationships=relationships)


@app.get("/api/nodes", response_model=list[NodeSummary])
def api_nodes(
    label: str = Query(..., description="Node label to seed from"),
    limit: int = Query(50, ge=1, le=500),
    q: str | None = Query(None, description="Substring filter on the display name"),
    graph: GraphKind = Query(
        "memgraph",
        description="'memgraph' or 'kg' ('neo4j' accepted as a deprecated alias for 'kg')",
    ),
) -> list[NodeSummary]:
    """Seed nodes for a given label, optionally filtered by a name substring."""
    driver = _get_driver(graph)
    try:
        known = _known_labels(driver, graph)
        if label not in known:
            raise HTTPException(status_code=400, detail=f"unknown label: {label!r}")
        with _open_session(driver, graph) as session:
            cypher = f"MATCH (n:`{label}`) "
            params: dict[str, Any] = {"limit": limit}
            if q:
                # `toString` REJECTS list values outright — it accepts only
                # scalars/temporals — so filtering a label whose nodes carry
                # any list property (every chunk label has `embedding`, a
                # list of floats) raised a ClientError and returned a 500
                # instead of a result set. Restricting the scan to string
                # properties is also the only thing a substring filter can
                # meaningfully match: an embedding vector rendered as text is
                # not something a user searches for.
                # `valueType()`, not the `IS :: STRING` type predicate — the
                # latter is not parsed by this Memgraph build (verified live
                # 2026-07-26).
                cypher += (
                    "WHERE any(prop IN keys(n) WHERE "
                    "valueType(n[prop]) = 'STRING' AND "
                    "toLower(n[prop]) CONTAINS toLower($q)) "
                )
                params["q"] = q
            cypher += "RETURN id(n) AS id, labels(n) AS labels, properties(n) AS props LIMIT $limit"
            rows = list(session.run(cypher, params))
    finally:
        driver.close()

    return [
        NodeSummary(
            id=row["id"],
            labels=row["labels"],
            name=_display_name(row["props"], row["id"], row["labels"]),
            properties=_trim_props(row["props"]),
        )
        for row in rows
    ]


@app.get("/api/expand/{node_id}", response_model=ExpandResponse)
def api_expand(
    node_id: int,
    limit: int = Query(100, ge=1, le=500),
    graph: GraphKind = Query(
        "memgraph",
        description="'memgraph' or 'kg' ('neo4j' accepted as a deprecated alias for 'kg')",
    ),
) -> ExpandResponse:
    """Neighbors + relationships of a node, capped at ``limit`` edges."""
    driver = _get_driver(graph)
    try:
        with _open_session(driver, graph) as session:
            exists_row = session.run(
                "MATCH (n) WHERE id(n) = $id RETURN n IS NOT NULL AS exists",
                {"id": node_id},
            ).single()
            if exists_row is None:
                raise HTTPException(status_code=404, detail=f"node {node_id} not found")

            rows = list(
                session.run(
                    "MATCH (n)-[r]-(m) WHERE id(n) = $id "
                    "RETURN id(r) AS rid, type(r) AS rtype, "
                    "startNode(r) = n AS outgoing, "
                    "id(m) AS mid, labels(m) AS mlabels, properties(m) AS mprops "
                    "LIMIT $limit",
                    {"id": node_id, "limit": limit + 1},
                )
            )
            total_row = session.run(
                "MATCH (n)-[r]-() WHERE id(n) = $id RETURN count(r) AS total",
                {"id": node_id},
            ).single()
    finally:
        driver.close()

    total = total_row["total"] if total_row is not None else 0
    truncated = total > limit
    rows = rows[:limit]

    seen_nodes: dict[int, NodeSummary] = {}
    edges: list[EdgeSummary] = []
    for row in rows:
        mid = row["mid"]
        if mid not in seen_nodes:
            seen_nodes[mid] = NodeSummary(
                id=mid,
                labels=row["mlabels"],
                name=_display_name(row["mprops"], mid, row["mlabels"]),
                properties=_trim_props(row["mprops"]),
            )
        source, target = (node_id, mid) if row["outgoing"] else (mid, node_id)
        edges.append(
            EdgeSummary(id=row["rid"], source=source, target=target, type=row["rtype"])
        )

    return ExpandResponse(
        nodes=list(seen_nodes.values()), edges=edges, truncated=truncated
    )


def main() -> None:
    import uvicorn

    parser = argparse.ArgumentParser(
        description="Launch the interactive graph browser."
    )
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8090)
    args = parser.parse_args()
    # http="h11": "auto" imports httptools, which re-enables the GIL
    # process-wide (SPEC-023). Same pin as scripts/ast_daemon.py.
    uvicorn.run(app, host=args.host, port=args.port, http="h11")


if __name__ == "__main__":
    main()