Function export_graph_data

code_review_graph/visualization.py:102–164

Export all graph nodes and edges as a JSON-serializable dict. Returns ``{"nodes": [...], "edges": [...], "stats": {...}, "flows": [...], "communities": [...]}``.

(store: GraphStore)

def export_graph_data(store: GraphStore) -> dict:
    """Export all graph nodes and edges as a JSON-serializable dict.

    Returns ``{"nodes": [...], "edges": [...], "stats": {...},
    "flows": [...], "communities": [...]}``.
    """
    nodes = []
    seen_qn: set[str] = set()

    # Preload community_id mapping from DB (column may not exist in old schemas)
    community_map = store.get_all_community_ids()

    for file_path in store.get_all_files():
        for gnode in store.get_nodes_by_file(file_path):
            if gnode.qualified_name in seen_qn:
                continue
            seen_qn.add(gnode.qualified_name)
            d = node_to_dict(gnode)
            d["params"] = gnode.params
            d["return_type"] = gnode.return_type
            d["community_id"] = community_map.get(gnode.qualified_name)
            nodes.append(d)

    name_index = _build_name_index(nodes, seen_qn)

    all_edges = [edge_to_dict(e) for e in store.get_all_edges()]

    # Resolve short/unqualified edge targets to full qualified names,
    # then drop edges that still can't be resolved (external/stdlib calls).
    edges = []
    for e in all_edges:
        src = _resolve_target(e["source"], e["source"], seen_qn, name_index)
        tgt = _resolve_target(e["target"], e["source"], seen_qn, name_index)
        if src and tgt:
            e["source"] = src
            e["target"] = tgt
            edges.append(e)

    stats = store.get_stats()

    # Include flows (graceful fallback if table doesn't exist)
    try:
        from code_review_graph.flows import get_flows
        flows = get_flows(store, limit=100)
    except (ImportError, sqlite3.OperationalError) as exc:
        logger.debug("flows unavailable for export: %s", exc)
        flows = []

    # Include communities (graceful fallback if table doesn't exist)
    try:
        from code_review_graph.communities import get_communities
        communities = get_communities(store)
    except (ImportError, sqlite3.OperationalError) as exc:
        logger.debug("communities unavailable for export: %s", exc)
        communities = []

    return {
        "nodes": nodes,