[Prototype] Node grouping

kedro/pipeline/pipeline.py

@@ -369,6 +369,31 @@ def grouped_nodes(self) -> list[list[Node]]:
 
         return [list(group) for group in self._toposorted_groups]
 
+    @property
+    def grouped_nodes_by_namespace(self) -> dict[str, dict[str, Any]]:
+        """Return a dictionary of the pipeline nodes grouped by namespace with
+        information about the nodes, their type, and dependencies."""
+        grouped_nodes: dict[str, dict[str, Any]] = defaultdict(dict)
+        for node in self.nodes:
+            key = node.namespace or node.name
+            if key not in grouped_nodes:
+                grouped_nodes[key] = {}
+                grouped_nodes[key]["name"] = key
+                grouped_nodes[key]["type"] = "namespace" if node.namespace else "node"
+            grouped_nodes[key]["nodes"] = [*grouped_nodes[key].get("nodes", []), node]
+            dependencies = set()
+            for parent in self.node_dependencies[node]:
+                if parent.namespace and parent.namespace != key:
+                    dependencies.add(parent.namespace)
+                elif parent.namespace and parent.namespace == key:
+                    continue
+                else:
+                    dependencies.add(parent.name)
+            grouped_nodes[key]["dependencies"] = (
+                grouped_nodes[key].get("dependencies", set()) | dependencies
+            )
+        return grouped_nodes
+
     def only_nodes(self, *node_names: str) -> Pipeline:
         """Create a new ``Pipeline`` which will contain only the specified
         nodes by name.

tests/pipeline/test_pipeline.py

@@ -376,6 +376,37 @@ def test_node_dependencies(self, complex_pipeline):
         }
         assert actual == expected
 
+    def test_node_grouping_by_namespace(self):
+        pipeline = modular_pipeline(
+            [
+                node(identity, "A", "B", name="node1", namespace="name_1"),
+                node(identity, "B", "C", name="node2", namespace="name_1"),
+                node(identity, "C", "D", name="node3", namespace="name_2"),
+                node(identity, "D", "E", name="node4", namespace="name_2"),
+                node(identity, "E", "G", name="node5"),
+                node(identity, "G", "H", name="node6"),
+            ]
+        )
+        grouped = pipeline.grouped_nodes_by_namespace
+        # Validate keys for namespace groups
+        for key in ["name_1", "name_2"]:
+            assert key in grouped
+            assert grouped[key]["name"] == key
+            assert grouped[key]["type"] == "namespace"
+            assert len(grouped[key]["nodes"]) == 2
+
+        # Validate dependencies for namespace groups
+        assert grouped["name_1"]["dependencies"] == set()
+        assert grouped["name_2"]["dependencies"] == {"name_1"}
+
+        # Validate nodes for namespace groups
+        assert grouped["node5"]["type"] == "node"
+        assert grouped["node5"]["name"] == "node5"
+        assert len(grouped["node5"]["nodes"]) == 1
+        assert grouped["node5"]["dependencies"] == {"name_2"}
+        # Validate when node depends on node
+        assert grouped["node6"]["dependencies"] == {"node5"}
+
 
 @pytest.fixture
 def pipeline_with_circle():