Node Management Routes

This section documents the node management API endpoints.

backend.routes.nodes

Classes

Functions

set_node_nbh(user_id: str, graph_id: str, node_id: str, nbh: str)

Set the active morph (neighborhood) for a polynode. This determines which morph state the node is currently in.

Source code in backend/routes/nodes.py

@router.put("/users/{user_id}/graphs/{graph_id}/set_nbh/{node_id}")
def set_node_nbh(user_id: str, graph_id: str, node_id: str, nbh: str):
    """
    Set the active morph (neighborhood) for a polynode.
    This determines which morph state the node is currently in.
    """
    try:
        with graph_transaction(user_id, graph_id, "set_node_nbh") as backup_dir:
            # Load the node
            node_path = Path(f"graph_data/users/{user_id}/nodes/{node_id}.json")
            if not node_path.exists():
                raise HTTPException(status_code=404, detail="Node not found")

            node_data = load_json_file(node_path)

            # Verify the morph exists
            morph_exists = False
            for morph in node_data.get("morphs", []):
                if morph.get("morph_id") == nbh:
                    morph_exists = True
                    break

            if not morph_exists:
                raise HTTPException(status_code=404, detail=f"Morph {nbh} not found in node {node_id}")

            # Set the active morph
            node_data["nbh"] = nbh

            # Save the updated node
            atomic_node_save(user_id, node_id, node_data)

            return {
                "status": "Active morph set successfully",
                "node_id": node_id,
                "active_morph": nbh,
                "available_morphs": [m.get("morph_id") for m in node_data.get("morphs", [])]
            }

    except AtomicityError as e:
        raise HTTPException(status_code=500, detail=f"Atomic operation failed: {str(e)}")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Failed to set active morph: {str(e)}")

get_node_registry(user_id: str)

Return the user's canonical node_registry.json for fast node lookup and autocomplete.

Source code in backend/routes/nodes.py

@router.get("/users/{user_id}/node_registry")
def get_node_registry(user_id: str):
    """
    Return the user's canonical node_registry.json for fast node lookup and autocomplete.
    """
    registry_path = os.path.join("graph_data", "users", user_id, "node_registry.json")
    if not os.path.exists(registry_path):
        return {}
    with open(registry_path, "r") as f:
        return json.load(f)

get_cross_graph_reuse_stats(user_id: str, graph_id: str)

Get statistics about cross-graph node reuse for a specific graph.

Source code in backend/routes/nodes.py

@router.get("/users/{user_id}/graphs/{graph_id}/cross_graph_reuse")
def get_cross_graph_reuse_stats(user_id: str, graph_id: str):
    """Get statistics about cross-graph node reuse for a specific graph."""
    registry = load_node_registry(user_id)
    reuse_count = 0
    reused_nodes = []

    for node_id, entry in registry.items():
        graphs = entry.get("graphs", [])
        if graph_id in graphs and len(graphs) > 1:
            # This node is used in multiple graphs, including the current one
            reuse_count += 1
            reused_nodes.append({
                "id": node_id,
                "name": entry.get("name", node_id),
                "graphs": graphs
            })

    return {
        "cross_graph_reuse_count": reuse_count,
        "reused_nodes": reused_nodes
    }