teff.observability.topology

teff.observability.topology

Graph topology snapshot — the node/edge shape for the dashboard.

Functions:

Name	Description
topology_from_graph	Capture {nodes, edges} from a compiled :class:~teff.graph.Graph.

topology_from_graph

topology_from_graph(graph)

Capture {nodes, edges} from a compiled :class:~teff.graph.Graph.

Each node is {"id", "type"}; each edge is {"source", "target", "condition"} (condition omitted when unconditional).

Source code in teff/observability/topology.py

def topology_from_graph(graph) -> GraphTopology:
    """Capture ``{nodes, edges}`` from a compiled :class:`~teff.graph.Graph`.

    Each node is ``{"id", "type"}``; each edge is ``{"source", "target",
    "condition"}`` (``condition`` omitted when unconditional).
    """
    nodes = [
        {"id": node_id, "type": node.type} for node_id, node in graph.nodes.items()
    ]
    edges: list[dict[str, Any]] = []
    for edge in graph.edges:
        item: dict[str, Any] = {"source": edge.source_id, "target": edge.target_id}
        if getattr(edge, "condition", None):
            item["condition"] = edge.condition
        edges.append(item)
    return GraphTopology(nodes=nodes, edges=edges)