add code generation, ingest process resource

Author: a-s-g93

servers/mcp-neo4j-data-modeling/CHANGELOG.md

@@ -5,7 +5,8 @@
 ### Changed
 
 ### Added
-
+* Code generation tools for ingestion queries
+* Resource that explains the recommended process of ingesting data into Neo4j
 
 ## v0.1.0
 

servers/mcp-neo4j-data-modeling/claude_desktop_config.json

@@ -170,7 +170,7 @@ def get_mermaid_config_str(self) -> str:
         "Get the Mermaid configuration string for the node."
         props = [f"<br/>{self.key_property.name}: {self.key_property.type} | KEY"]
         props.extend([f"<br/>{p.name}: {p.type}" for p in self.properties])
-        return f'{self.label}["{self.label}{''.join(props)}"]'
+        return f'{self.label}["{self.label}{"".join(props)}"]'
 
     @classmethod
     def from_arrows(cls, arrows_node_dict: dict[str, Any]) -> "Node":
@@ -214,6 +214,25 @@ def to_arrows(
             "caption": self.metadata.get("caption", ""),
         }
 
+    def get_cypher_ingest_query_for_many_records(self) -> str:
+        """
+        Generate a Cypher query to ingest a list of Node records into a Neo4j database.
+        This query takes a parameter $records that is a list of dictionaries, each representing a Node record.
+        """
+        formatted_props = ", ".join(
+            [f"{p.name}: record.{p.name}" for p in self.properties]
+        )
+        return f"""UNWIND $records as record
+MERGE (n: {self.label} {{{self.key_property.name}: record.{self.key_property.name}}})
+SET n += {{{formatted_props}}}"""
+
+    def get_cypher_constraint_query(self) -> str:
+        """
+        Generate a Cypher query to create a NODE KEY constraint on the node.
+        This creates a range index on the key property of the node and enforces uniqueness and existence of the key property.
+        """
+        return f"CREATE CONSTRAINT {self.label}_constraint IF NOT EXISTS FOR (n:{self.label}) REQUIRE (n.{self.key_property.name}) IS NODE KEY"
+
 
 class Relationship(BaseModel):
     "A Neo4j Relationship."
@@ -347,6 +366,41 @@ def to_arrows(self) -> dict[str, Any]:
             "style": self.metadata.get("style", {}),
         }
 
+    def get_cypher_ingest_query_for_many_records(
+        self, start_node_key_property_name: str, end_node_key_property_name: str
+    ) -> str:
+        """
+        Generate a Cypher query to ingest a list of Relationship records into a Neo4j database.
+        The sourceId and targetId properties are used to match the start and end nodes.
+        This query takes a parameter $records that is a list of dictionaries, each representing a Relationship record.
+        """
+        formatted_props = ", ".join(
+            [f"{p.name}: record.{p.name}" for p in self.properties]
+        )
+        key_prop = (
+            f" {{{self.key_property.name}: record.{self.key_property.name}}}"
+            if self.key_property
+            else ""
+        )
+        query = f"""UNWIND $records as record
+MATCH (start: {self.start_node_label} {{{start_node_key_property_name}: record.sourceId}})
+MATCH (end: {self.end_node_label} {{{end_node_key_property_name}: record.targetId}})
+MERGE (start)-[:{self.type}{key_prop}]->(end)"""
+        if formatted_props:
+            query += f"""
+SET end += {{{formatted_props}}}"""
+        return query
+
+    def get_cypher_constraint_query(self) -> str | None:
+        """
+        Generate a Cypher query to create a RELATIONSHIP KEY constraint on the relationship.
+        This creates a range index on the key property of the relationship and enforces uniqueness and existence of the key property.
+        """
+        if self.key_property:
+            return f"CREATE CONSTRAINT {self.type}_constraint IF NOT EXISTS FOR ()-[r:{self.type}]->() REQUIRE (r.{self.key_property.name}) IS RELATIONSHIP KEY"
+        else:
+            return None
+
 
 class DataModel(BaseModel):
     "A Neo4j Graph Data Model."
@@ -403,6 +457,16 @@ def validate_relationships(
 
         return relationships
 
+    @property
+    def nodes_dict(self) -> dict[str, Node]:
+        "Return a dictionary of the nodes of the data model. {node_label: node_dict}"
+        return {n.label: n for n in self.nodes}
+
+    @property
+    def relationships_dict(self) -> dict[str, Relationship]:
+        "Return a dictionary of the relationships of the data model. {relationship_pattern: relationship_dict}"
+        return {r.pattern: r for r in self.relationships}
+
     def add_node(self, node: Node) -> None:
         "Add a new node to the data model."
         if node.label in [n.label for n in self.nodes]:
@@ -520,3 +584,40 @@ def to_arrows_dict(self) -> dict[str, Any]:
     def to_arrows_json_str(self) -> str:
         "Convert the data model to an Arrows Data Model JSON string."
         return json.dumps(self.to_arrows_dict(), indent=2)
+
+    def get_node_cypher_ingest_query_for_many_records(self, node_label: str) -> str:
+        "Generate a Cypher query to ingest a list of Node records into a Neo4j database."
+        node = self.nodes_dict[node_label]
+        return node.get_cypher_ingest_query_for_many_records()
+
+    def get_relationship_cypher_ingest_query_for_many_records(
+        self,
+        relationship_type: str,
+        relationship_start_node_label: str,
+        relationship_end_node_label: str,
+    ) -> str:
+        "Generate a Cypher query to ingest a list of Relationship records into a Neo4j database."
+        pattern = _generate_relationship_pattern(
+            relationship_start_node_label,
+            relationship_type,
+            relationship_end_node_label,
+        )
+        relationship = self.relationships_dict[pattern]
+        start_node = self.nodes_dict[relationship.start_node_label]
+        end_node = self.nodes_dict[relationship.end_node_label]
+        return relationship.get_cypher_ingest_query_for_many_records(
+            start_node.key_property.name, end_node.key_property.name
+        )
+
+    def get_cypher_constraints_query(self) -> list[str]:
+        """
+        Generate a list of Cypher queries to create constraints on the data model.
+        This creates range indexes on the key properties of the nodes and relationships and enforces uniqueness and existence of the key properties.
+        """
+        node_queries = [n.get_cypher_constraint_query() + ";" for n in self.nodes]
+        relationship_queries = [
+            r.get_cypher_constraint_query() + ";"
+            for r in self.relationships
+            if r.key_property is not None
+        ]
+        return node_queries + relationship_queries

servers/mcp-neo4j-data-modeling/src/mcp_neo4j_data_modeling/data_model.py

@@ -4,14 +4,15 @@
 from typing import Any, Literal
 
 from mcp.server.fastmcp import FastMCP
-from pydantic import ValidationError
+from pydantic import Field, ValidationError
 
 from .data_model import (
     DataModel,
     Node,
     Property,
     Relationship,
 )
+from .static import DATA_INGEST_PROCESS
 
 logger = logging.getLogger("mcp_neo4j_data_modeling")
 
@@ -47,6 +48,12 @@ def data_model_schema() -> dict[str, Any]:
         logger.info("Getting the schema for a data model.")
         return DataModel.model_json_schema()
 
+    @mcp.resource("resource://static/neo4j_data_ingest_process")
+    def neo4j_data_ingest_process() -> str:
+        """Get the process for ingesting data into a Neo4j database."""
+        logger.info("Getting the process for ingesting data into a Neo4j database.")
+        return DATA_INGEST_PROCESS
+
     @mcp.tool()
     def validate_node(
         node: Node, return_validated: bool = False
@@ -152,6 +159,58 @@ def get_mermaid_config_str(data_model: DataModel) -> str:
             raise ValueError(f"Validation error: {e}")
         return dm_validated.get_mermaid_config_str()
 
+    @mcp.tool()
+    def get_node_cypher_ingest_query_for_many_records(
+        node: Node = Field(description="The node to get the Cypher query for."),
+    ) -> str:
+        """
+        Get the Cypher query to ingest a list of Node records into a Neo4j database.
+        This should be used to ingest data into a Neo4j database.
+        This is a parameterized Cypher query that takes a list of records as input to the $records parameter.
+        """
+        logger.info(
+            f"Getting the Cypher query to ingest a list of Node records into a Neo4j database for node {node.label}."
+        )
+        return node.get_cypher_ingest_query_for_many_records()
+
+    @mcp.tool()
+    def get_relationship_cypher_ingest_query_for_many_records(
+        data_model: DataModel = Field(
+            description="The data model snippet that contains the relationship, start node and end node."
+        ),
+        relationship_type: str = Field(
+            description="The type of the relationship to get the Cypher query for."
+        ),
+        relationship_start_node_label: str = Field(
+            description="The label of the relationship start node."
+        ),
+        relationship_end_node_label: str = Field(
+            description="The label of the relationship end node."
+        ),
+    ) -> str:
+        """
+        Get the Cypher query to ingest a list of Relationship records into a Neo4j database.
+        This should be used to ingest data into a Neo4j database.
+        This is a parameterized Cypher query that takes a list of records as input to the $records parameter.
+        The records must contain the Relationship properties, if any, as well as the sourceId and targetId properties of the start and end nodes respectively.
+        """
+        logger.info(
+            "Getting the Cypher query to ingest a list of Relationship records into a Neo4j database."
+        )
+        return data_model.get_relationship_cypher_ingest_query_for_many_records(
+            relationship_type,
+            relationship_start_node_label,
+            relationship_end_node_label,
+        )
+
+    @mcp.tool()
+    def get_constraints_cypher_queries(data_model: DataModel) -> list[str]:
+        "Get the Cypher queries to create constraints on the data model. This creates range indexes on the key properties of the nodes and relationships and enforces uniqueness and existence of the key properties."
+        logger.info(
+            "Getting the Cypher queries to create constraints on the data model."
+        )
+        return data_model.get_cypher_constraints_query()
+
     return mcp
 
 

servers/mcp-neo4j-data-modeling/src/mcp_neo4j_data_modeling/server.py

@@ -0,0 +1,6 @@
+DATA_INGEST_PROCESS = """
+Follow these steps when ingesting data into Neo4j.
+1. Create constraints before loading any data.
+2. Load all nodes before relationships.
+3. Then load relationships serially to avoid deadlocks.
+"""

servers/mcp-neo4j-data-modeling/src/mcp_neo4j_data_modeling/static.py

@@ -2,6 +2,8 @@
 
 import pytest
 
+from mcp_neo4j_data_modeling.data_model import DataModel, Node, Property, Relationship
+
 
 @pytest.fixture(scope="function")
 def arrows_data_model_dict() -> dict[str, Any]:
@@ -130,3 +132,36 @@ def arrows_data_model_dict() -> dict[str, Any]:
             },
         ],
     }
+
+
+@pytest.fixture(scope="function")
+def valid_data_model() -> DataModel:
+    "A simple valid data model with a Person node, a Place node, and a LIVES_IN relationship."
+    nodes = [
+        Node(
+            label="Person",
+            key_property=Property(
+                name="id", type="STRING", description="Unique identifier"
+            ),
+            properties=[
+                Property(name="name", type="STRING", description="Name of the person"),
+                Property(name="age", type="INTEGER", description="Age of the person"),
+            ],
+        ),
+        Node(
+            label="Place",
+            key_property=Property(
+                name="id", type="STRING", description="Unique identifier"
+            ),
+            properties=[
+                Property(name="name", type="STRING", description="Name of the place")
+            ],
+        ),
+    ]
+
+    relationship = Relationship(
+        type="LIVES_IN",
+        start_node_label="Person",
+        end_node_label="Place",
+    )
+    return DataModel(nodes=nodes, relationships=[relationship])