Implement CollapsPath

Author: DarthMax

graphdatascience/procedure_surface/api/catalog/relationships_endpoints.py

@@ -208,6 +208,52 @@ def to_undirected(
 
     pass
 
+    @abstractmethod
+    def collapse_path(
+        self,
+        G: GraphV2,
+        path_templates: list[list[str]],
+        mutate_relationship_type: str,
+        *,
+        allow_self_loops: bool = False,
+        concurrency: int | None = None,
+        job_id: str | None = None,
+        sudo: bool = False,
+        log_progress: bool = True,
+        username: str | None = None,
+    ) -> CollapsePathResult:
+        """
+        Collapse a path in the graph catalog.
+
+        Parameters
+        ----------
+
+        G : GraphV2
+            Name of the generated graph.
+        path_templates : list[list[str]]
+            A path template is an ordered list of relationship types used for the traversal. The same relationship type can be added multiple times, in order to traverse them as indicated. And, you may specify several path templates to process in one go.
+        mutate_relationship_type : str
+            The name of the new relationship type to be created.
+        allow_self_loops : bool, default=False
+            Whether nodes in the graph can have relationships where start and end nodes are the same.
+        concurrency : int | None, default=None:
+            Number of concurrent threads to use. Defaults to None.
+        job_id : str | None, default=None
+            Unique identifier for the job associated with the graph generation.
+        sudo : bool | None, default=None
+            Override memory estimation limits
+        log_progress : bool | None, default=None
+            Whether to log progress during graph generation.
+        username : str | None, default=None
+            Username of the individual requesting the graph generation.
+
+        Returns
+        -------
+        CollapsePathResult: meta data about the generated relationships.
+        """
+
+    pass
+
 
 class RelationshipsWriteResult(BaseResult):
     graph_name: str
@@ -251,6 +297,14 @@ class RelationshipsToUndirectedResult(RelationshipsInverseIndexResult):
     relationships_written: int
 
 
+class CollapsePathResult(BaseResult):
+    preProcessingMillis: int
+    computeMillis: int
+    mutateMillis: int
+    relationshipsWritten: int
+    configuration: dict[str, Any]
+
+
 class Aggregation(str, Enum):
     NONE = "NONE"
     SINGLE = "SINGLE"

graphdatascience/procedure_surface/arrow/catalog/relationship_arrow_endpoints.py

@@ -7,6 +7,7 @@
 from graphdatascience.procedure_surface.api.catalog.graph_api import GraphV2
 from graphdatascience.procedure_surface.api.catalog.relationships_endpoints import (
     Aggregation,
+    CollapsePathResult,
     RelationshipsDropResult,
     RelationshipsEndpoints,
     RelationshipsInverseIndexResult,
@@ -197,3 +198,35 @@ def to_undirected(
         )
         result = JobClient.get_summary(self._arrow_client, job_id)
         return RelationshipsToUndirectedResult(**result)
+
+    def collapse_path(
+        self,
+        G: GraphV2,
+        path_templates: list[list[str]],
+        mutate_relationship_type: str,
+        *,
+        allow_self_loops: bool = False,
+        concurrency: int | None = None,
+        job_id: str | None = None,
+        sudo: bool = False,
+        log_progress: bool = True,
+        username: str | None = None,
+    ) -> CollapsePathResult:
+        config = ConfigConverter.convert_to_gds_config(
+            graph_name=G.name(),
+            path_templates=path_templates,
+            mutate_relationship_type=mutate_relationship_type,
+            allow_self_loops=allow_self_loops,
+            concurrency=concurrency,
+            job_id=job_id,
+            sudo=sudo,
+            log_progress=log_progress,
+            username=username,
+        )
+
+        show_progress = self._show_progress and log_progress
+        job_id = JobClient.run_job_and_wait(
+            self._arrow_client, "v2/graph.relationships.collapsePath", config, show_progress=show_progress
+        )
+
+        return CollapsePathResult(**JobClient.get_summary(self._arrow_client, job_id))

graphdatascience/procedure_surface/cypher/catalog/relationship_cypher_endpoints.py

@@ -5,6 +5,7 @@
 from graphdatascience.procedure_surface.api.catalog.graph_api import GraphV2
 from graphdatascience.procedure_surface.api.catalog.relationships_endpoints import (
     Aggregation,
+    CollapsePathResult,
     RelationshipsDropResult,
     RelationshipsEndpoints,
     RelationshipsInverseIndexResult,
@@ -224,3 +225,36 @@ def to_undirected(
         ).squeeze()
 
         return RelationshipsToUndirectedResult(**result.to_dict())
+
+    def collapse_path(
+        self,
+        G: GraphV2,
+        path_templates: list[list[str]],
+        mutate_relationship_type: str,
+        *,
+        allow_self_loops: bool = False,
+        concurrency: int | None = None,
+        job_id: str | None = None,
+        sudo: bool = False,
+        log_progress: bool = True,
+        username: str | None = None,
+    ) -> CollapsePathResult:
+        config = ConfigConverter.convert_to_gds_config(
+            path_templates=path_templates,
+            mutate_relationship_type=mutate_relationship_type,
+            allow_self_loops=allow_self_loops,
+            concurrency=concurrency,
+            job_id=job_id,
+            sudo=sudo,
+            log_progress=log_progress,
+            username=username,
+        )
+
+        params = CallParameters(
+            graph_name=G.name(),
+            config=config,
+        )
+        params.ensure_job_id_in_config()
+
+        result = self._query_runner.call_procedure(endpoint="gds.collapsePath.mutate", params=params).squeeze()
+        return CollapsePathResult(**result.to_dict())

graphdatascience/tests/integrationV2/procedure_surface/arrow/catalog/test_relationship_arrow_endpoints.py

@@ -231,3 +231,16 @@ def test_to_undirected_with_property_aggregation(
     assert result.mutate_millis >= 0
     assert result.input_relationships == 3
     assert result.relationships_written == 6
+
+
+def test_collapse_path(relationship_endpoints: RelationshipArrowEndpoints, sample_graph: GraphV2) -> None:
+    result = relationship_endpoints.collapse_path(
+        G=sample_graph,
+        path_templates=[["REL", "REL"]],
+        mutate_relationship_type="FoF",
+    )
+
+    assert result.relationshipsWritten == 3
+    assert result.mutateMillis >= 0
+    assert result.preProcessingMillis >= 0
+    assert result.computeMillis >= 0

graphdatascience/tests/integrationV2/procedure_surface/cypher/catalog/test_relationship_cypher_endpoints.py

@@ -252,3 +252,16 @@ def test_to_undirected_with_property_aggregation(
     assert result.mutate_millis >= 0
     assert result.input_relationships == 3
     assert result.relationships_written == 6
+
+
+def test_collapse_path(relationship_endpoints: RelationshipCypherEndpoints, sample_graph: GraphV2) -> None:
+    result = relationship_endpoints.collapse_path(
+        G=sample_graph,
+        path_templates=[["REL", "REL"]],
+        mutate_relationship_type="FoF",
+    )
+
+    assert result.relationshipsWritten == 3
+    assert result.mutateMillis >= 0
+    assert result.preProcessingMillis >= 0
+    assert result.computeMillis >= 0