Node grouping API

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]["dependencies"] = set()
+            grouped_nodes[key]["nodes"].append(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"].update(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,88 @@ def test_node_dependencies(self, complex_pipeline):
         }
         assert actual == expected
 
+    @pytest.mark.parametrize(
+        "pipeline_name, expected",
+        [
+            ("pipeline_with_namespace_simple", ["namespace_1", "namespace_2"]),
+            (
+                "pipeline_with_namespace_partial",
+                ["namespace_1", "node_3", "namespace_2", "node_6"],
+            ),
+        ],
+    )
+    def test_node_grouping_by_namespace_name_type(
+        self, request, pipeline_name, expected
+    ):
+        p = request.getfixturevalue(pipeline_name)
+        grouped = p.grouped_nodes_by_namespace
+        assert set(grouped.keys()) == set(expected)
+        for key in expected:
+            assert grouped[key]["name"] == key
+            assert key.startswith(grouped[key]["type"])
+
+    @pytest.mark.parametrize(
+        "pipeline_name, expected",
+        [
+            (
+                "pipeline_with_namespace_simple",
+                {
+                    "namespace_1": [
+                        "namespace_1.node_1",
+                        "namespace_1.node_2",
+                        "namespace_1.node_3",
+                    ],
+                    "namespace_2": [
+                        "namespace_2.node_4",
+                        "namespace_2.node_5",
+                        "namespace_2.node_6",
+                    ],
+                },
+            ),
+            (
+                "pipeline_with_namespace_partial",
+                {
+                    "namespace_1": ["namespace_1.node_1", "namespace_1.node_2"],
+                    "node_3": ["node_3"],
+                    "namespace_2": ["namespace_2.node_4", "namespace_2.node_5"],
+                    "node_6": ["node_6"],
+                },
+            ),
+        ],
+    )
+    def test_node_grouping_by_namespace_nodes(self, request, pipeline_name, expected):
+        p = request.getfixturevalue(pipeline_name)
+        grouped = p.grouped_nodes_by_namespace
+        for key, value in grouped.items():
+            names = [node.name for node in value["nodes"]]
+            assert set(names) == set(expected[key])
+
+    @pytest.mark.parametrize(
+        "pipeline_name, expected",
+        [
+            (
+                "pipeline_with_namespace_simple",
+                {"namespace_1": set(), "namespace_2": {"namespace_1"}},
+            ),
+            (
+                "pipeline_with_namespace_partial",
+                {
+                    "namespace_1": set(),
+                    "node_3": {"namespace_1"},
+                    "namespace_2": {"node_3"},
+                    "node_6": {"namespace_2"},
+                },
+            ),
+        ],
+    )
+    def test_node_grouping_by_namespace_dependencies(
+        self, request, pipeline_name, expected
+    ):
+        p = request.getfixturevalue(pipeline_name)
+        grouped = p.grouped_nodes_by_namespace
+        for key, value in grouped.items():
+            assert set(value["dependencies"]) == set(expected[key])
+
 
 @pytest.fixture
 def pipeline_with_circle():
@@ -758,6 +840,34 @@ def pipeline_with_namespaces():
     )
 
 
+@pytest.fixture
+def pipeline_with_namespace_simple():
+    return modular_pipeline(
+        [
+            node(identity, "A", "B", name="node_1", namespace="namespace_1"),
+            node(identity, "B", "C", name="node_2", namespace="namespace_1"),
+            node(identity, "C", "D", name="node_3", namespace="namespace_1"),
+            node(identity, "D", "E", name="node_4", namespace="namespace_2"),
+            node(identity, "E", "F", name="node_5", namespace="namespace_2"),
+            node(identity, "F", "G", name="node_6", namespace="namespace_2"),
+        ]
+    )
+
+
+@pytest.fixture
+def pipeline_with_namespace_partial():
+    return modular_pipeline(
+        [
+            node(identity, "A", "B", name="node_1", namespace="namespace_1"),
+            node(identity, "B", "C", name="node_2", namespace="namespace_1"),
+            node(identity, "C", "D", name="node_3"),
+            node(identity, "D", "E", name="node_4", namespace="namespace_2"),
+            node(identity, "E", "F", name="node_5", namespace="namespace_2"),
+            node(identity, "F", "G", name="node_6"),
+        ]
+    )
+
+
 class TestPipelineFilter:
     def test_no_filters(self, complex_pipeline):
         filtered_pipeline = complex_pipeline.filter()