Examples of uploading Cytoscape formatted JSON data to Neo4j and downloading back as Cytoscape compatible dict

cytoscape2neo4j.py

from neo4j import GraphDatabase
from neo4j.graph import Relationship, Node
from typing import Optional, List
import json
import os
from dotenv import load_dotenv

load_dotenv()

def upload_cytoscape_to_neo4j(
    cytoscape_data: str | dict,
    neo4j_uri: Optional[str] = None,
    neo4j_user: Optional[str] = None,
    neo4j_password: Optional[str] = None,
):
    """
    Upload Cytoscape JSON data to Neo4j using the bolt driver
    Args:
        cytoscape_data (str | dict): Cytoscape formatted JSON data in either a string or dict representation
        neo4j_uri (str): URI for Neo4j database connection
        neo4j_user (str): Neo4j username
        neo4j_password (str): Neo4j password
    """
    print("Uploading Cytoscape data: ", cytoscape_data)
    if isinstance(cytoscape_data, str):
        cytoscape_data = json.loads(cytoscape_data)

    if neo4j_uri is None:
        neo4j_uri = os.getenv("NEO4J_URI")
    if neo4j_user is None:
        neo4j_user = os.getenv("NEO4J_USER", "neo4j")
    if neo4j_password is None:
        neo4j_password = os.getenv("NEO4J_PASSWORD")

    if neo4j_uri is None:
        raise ValueError("NEO4J_URI is required")
    if neo4j_password is None:
        raise ValueError("NEO4J_PASSWORD is required")

    try:
        # Connect to Neo4j using a context manager
        with GraphDatabase.driver(
            neo4j_uri, auth=(neo4j_user, neo4j_password)
        ) as driver:

            elements = cytoscape_data.get("elements")
            if not elements:
                raise ValueError("No elements found in Cytoscape data")
            if len(elements) != 2:
                raise ValueError(
                    "Expected exactly 2 elements in Cytoscape data. Got {}".format(
                        len(elements)
                    )
                )
            nodes = elements.get("nodes", [])
            if len(nodes) == 0:
                raise ValueError("No nodes found in Cytoscape data")
            edges = elements.get("edges", [])


            # Create nodes
            nodes_params = []
            for node in nodes:
                node_data = node.get("data", {})
                label = node_data.get("label")
                node_id = node_data.get("id")
                if label is None:
                    raise ValueError("Node 'label' missing")
                if node_id is None:
                    raise ValueError("Node 'id' missing")

                # Create properties string from remaining data
                properties = {k: v for k, v in node_data.items() if k != "label"}

                nodes_params.append({"label": label, "id": node_id, "properties": properties})

            # Use UNWIND statement to process list of data
            query = """
                UNWIND $nodes AS node 
                MERGE (n {id: node.id})
                WITH n, node
                CALL apoc.create.addLabels(n, [node.label]) YIELD node AS labeledNode
                SET labeledNode += node.properties
            """
            driver.execute_query(
                query_=query,
                parameters_={"nodes": nodes_params},
            )


            # Create relationships
            edges_params = []
            for edge in edges:
                edge_data = edge.get("data", {})
                source = edge_data.get("source")
                target = edge_data.get("target")
                rel_type = edge_data.get("label", "RELATED_TO").upper()

                # Create properties string from remaining data
                properties = {
                    k: v
                    for k, v in edge_data.items()
                    if k not in ["source", "target", "label"]
                }

                edges_params.append({"rel_type": rel_type, "source": source, "target": target, "properties": properties})

            query = """
            UNWIND $edges AS edge 
            MATCH (source {id: edge.source}), (target {id: edge.target})
            CALL apoc.create.relationship(source, edge.rel_type, edge.properties, target) YIELD rel
            RETURN rel
            """
            driver.execute_query(
                query_=query,
                parameters_={"edges": edges_params},
            )
            return True
    except Exception as e:
        print(f"Error uploading Cytoscape data to Neo4j: {str(e)}")
        return e


def download_neo4j_to_cytoscape(
    node_labels: List[str] = None,
    relationship_types: List[str] = None,
    neo4j_uri: Optional[str] = None,
    neo4j_user: Optional[str] = None,
    neo4j_password: Optional[str] = None,
):
    """
    Download data from Neo4j and convert it to Cytoscape JSON format
    Args:
        node_labels (List[str]): Labels of nodes to download
        relationship_types (List[str]): Types of relationships to download
        neo4j_uri (str): URI for Neo4j database connection
        neo4j_user (str): Neo4j username
        neo4j_password (str): Neo4j password

    Returns:
        str: Cytoscape JSON data
    """
    if neo4j_uri is None:
        neo4j_uri = os.getenv("NEO4J_URI")
    if neo4j_user is None:
        neo4j_user = os.getenv("NEO4J_USER", "neo4j")
    if neo4j_password is None:
        neo4j_password = os.getenv("NEO4J_PASSWORD")

    if neo4j_uri is None:
        raise ValueError("NEO4J_URI is required")
    if neo4j_password is None:
        raise ValueError("NEO4J_PASSWORD is required")
    try:
        # Connect to Neo4j using a context manager
        with GraphDatabase.driver(
            neo4j_uri, auth=(neo4j_user, neo4j_password)
        ) as driver:

            # Get all the node labels

            if node_labels is None:
                labels, _, _ = driver.execute_query("CALL db.labels()")
                print(f"Labels: {labels}")
                node_labels = [str(label.value("label")) for label in labels]
                print(f"Node Labels as strings: {node_labels}")


            if relationship_types is None:
                rel_types, _, _ = driver.execute_query("CALL db.relationshipTypes()")
                print(f"Relationship Types: {rel_types}")
                relationship_types = [str(rel_type.value("relationshipType")) for rel_type in rel_types]
                print(f"Relationship Types as strings: {relationship_types}")


            # Create a Cypher query to fetch data
            query = """
                MATCH (n)
                WHERE any(label IN labels(n) WHERE label IN $node_labels)
                OPTIONAL MATCH (n)-[r]->(m)
                WHERE type(r) IN $relationship_types
                RETURN n, r, m
            """
            params = {
                "node_labels": node_labels,
                "relationship_types": relationship_types,
            }

            print(f"Query: {query}")
            print(f"Params: {params}")

            # Execute the query - returns a tuple of (records, summary, keys)
            records, _, _ = driver.execute_query(query, params)

            # Convert Neo4j result to Cytoscape format
            nodes = []
            edges = []

            for record in records:

                rv = record.values()

                for i in rv:
                    if isinstance(i, Node):

                        # Set node to user assigned data
                        node_data = i._properties

                        # Update with required fields, overwriting id and label for consistency
                        node_data.update(
                            {
                                "id": i["id"],
                                "label": list(i.labels)[0],
                            }
                        )
                        nodes.append({"data": node_data})
                    elif isinstance(i, Relationship):
                        print("Relationship: ", i)
                        edge_data = i._properties
                        edge_data.update(
                            {
                                "id": i["id"],
                                "source": i.start_node["id"],
                                "target": i.end_node["id"],
                                "label": i.type,
                            }
                        )
                        edges.append({"data": edge_data})

            # # Return in Cytoscape format
            # # Additional key-values stubs added for compatibility with NetworkX for importing
            return {
                "data": [],
                "directed": False,
                "multigraph": False,
                "elements": {"nodes": nodes, "edges": edges},
            }
    except Exception as e:
        print(f"Error downloading data from Neo4j: {str(e)}")
        return e