Repository: kedro-org/kedro

Issue:
Prototype for node grouping deployment solution using namespaces
## Description

Following the discussion in #4319, we decided to add helper functions to Kedro to simplify the process for plugin developers to implement node grouping based on namespaces.

One of possible solutions was proposed by @marrrcin in [this comment](https://github.com/kedro-org/kedro/issues/4319#issuecomment-2527689783).

This prototype will be used together with modification of the `kedro-airflow` deployment plugin developed in [that PR](https://github.com/kedro-org/kedro-plugins/issues/962)  and the final result will be demonstrated in the Tech Design session.

Discussion:
ankatiyar: > Am I correct in thinking that if we add this method in the framework there's then no longer a need to add methods into plugins as well, like you now did in airflow in https://github.com/kedro-org/kedro-plugins/pull/981/files?

There would still need to be _some_ code added to the plugins to call this method but the plugins wouldn't need to implement the logic to figure out the groups and the dependencies
 
> I was wondering what the "type" is used for?

I believe it was for plugins to determine if the group is a node to then call `kedro run --nodes <nodename>` or `kedro run --namespace <namespace>` but I think we could do without it too and simply pass a list of nodes to `kedro run --nodes <node_1>,<node_2>`

merelcht: > There would still need to be _some_ code added to the plugins to call this method but the plugins wouldn't need to implement the logic to figure out the groups and the dependencies

Yes that makes sense 👍 
 
> I believe it was for plugins to determine if the group is a node to then call `kedro run --nodes <nodename>` or `kedro run --namespace <namespace>` but I think we could do without it too and simply pass a list of nodes to `kedro run --nodes <node_1>,<node_2>`

Ah okay. I don't think it functionally makes a difference, but might be easier for the user to understand what's happening if we clearly differentiate between `--namespace` and `--nodes`. Or is this not something the user will see?

I've played around with the airflow prototype a bit, and one thing I feel like is important here is to make sure that the namespace grouping is actually executable. If a user makes a mistake in grouping and for example forgets to add a node in a namespace which is required for the rest to run, we should error out as early on as possible so that the user doesn't have to go through a whole deployment and only then discover the grouping was wrong. 


ElenaKhaustova: > I think it would be more beneficial to return a topologically sorted list (similar to how pipeline.nodes currently works) instead of a dictionary. This way, the list can be executed in order.

Since dictionaries are ordered by default starting from `python3.8` the resulting dictionary is topologically sorted as we build it by iterating topologically sorted nodes. Also, dictionary representation looks more convenient to me as the whole namespace can be accessed.
ElenaKhaustova: > I've played around with the airflow prototype a bit, and one thing I feel like is important here is to make sure that the namespace grouping is actually executable. If a user makes a mistake in grouping and for example forgets to add a node in a namespace which is required for the rest to run

That's a valid point
ankatiyar: @merelcht @DimedS @ElenaKhaustova 
About the validation of namespaces, I think it's worth doing separately. Namespacing on pipeline level might be fine but Kedro allows for weird namespacing on node level as well which might result in situations were there's circular dependencies  between the groups. Maybe we need to prevent that at pipeline creation time and not during deployment?
merelcht: > @merelcht @DimedS @ElenaKhaustova About the validation of namespaces, I think it's worth doing separately. Namespacing on pipeline level might be fine but Kedro allows for weird namespacing on node level as well which might result in situations were there's circular dependencies between the groups. Maybe we need to prevent that at pipeline creation time and not during deployment?

Yes completely agree on tackling this separately. I think we need to set clear expectations for namespace behaviour and make changes accordingly. 

Patch:
diff --git a/RELEASE.md b/RELEASE.md
index bd95ac47..e974dd5b 100644
--- a/RELEASE.md
+++ b/RELEASE.md
@@ -2,6 +2,7 @@
 
 ## Major features and improvements
 * Added `KedroDataCatalog.filter()` to filter datasets by name and type.
+* Added `Pipeline.grouped_nodes_by_namespace` property which returns a dictionary of nodes grouped by namespace, intended to be used by plugins to facilitate deployment of namespaced nodes together.
 
 ## Bug fixes and other changes
 * Updated `_LazyDataset` representation when printing `KedroDataCatalog`.
diff --git a/docs/source/notebooks_and_ipython/notebook-example/add_kedro_to_a_notebook.md b/docs/source/notebooks_and_ipython/notebook-example/add_kedro_to_a_notebook.md
index 31b5ba92..cf32ffd6 100644
--- a/docs/source/notebooks_and_ipython/notebook-example/add_kedro_to_a_notebook.md
+++ b/docs/source/notebooks_and_ipython/notebook-example/add_kedro_to_a_notebook.md
@@ -152,7 +152,7 @@ When writing exploratory code, it’s tempting to hard code values to save time,
 X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=3)
 ```
 
-[Good software engineering practice](https://towardsdatascience.com/five-software-engineering-principles-for-collaborative-data-science-ab26667a311) suggests that we extract *‘magic numbers’* into named constants. These could be defined at the top of a file or in a utility file, in a format such as yaml.
+[Good software engineering practice](https://medium.com/towards-data-science/five-software-engineering-principles-for-collaborative-data-science-ab26667a311) suggests that we extract *‘magic numbers’* into named constants. These could be defined at the top of a file or in a utility file, in a format such as yaml.
 
 
 ```yaml
diff --git a/kedro/pipeline/pipeline.py b/kedro/pipeline/pipeline.py
index 826acd1b..b434bae4 100644
--- a/kedro/pipeline/pipeline.py
+++ b/kedro/pipeline/pipeline.py
@@ -369,6 +369,35 @@ class Pipeline:
 
         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. The structure of the dictionary is:
+        {'node_name/namespace_name' : {'name': 'node_name/namespace_name','type': 'namespace' or 'node','nodes': [list of nodes],'dependencies': [list of dependencies]}}
+        This property is intended to be used by deployment plugins to group nodes by namespace.
+
+        """
+        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.


Test patch:
diff --git a/tests/pipeline/test_pipeline.py b/tests/pipeline/test_pipeline.py
index 746ea479..1315e759 100644
--- a/tests/pipeline/test_pipeline.py
+++ b/tests/pipeline/test_pipeline.py
@@ -376,6 +376,118 @@ class TestValidPipeline:
         }
         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
+    ):
+        """Test for pipeline.grouped_nodes_by_namespace which returns a dictionary with the following structure:
+        {
+            'node_name/namespace_name' : {
+                                            'name': 'node_name/namespace_name',
+                                            'type': 'namespace' or 'node',
+                                            'nodes': [list of nodes],
+                                            'dependencies': [list of dependencies]}
+        }
+        This test checks for the 'name' and 'type' keys in the dictionary.
+        """
+        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):
+        """Test for pipeline.grouped_nodes_by_namespace which returns a dictionary with the following structure:
+        {
+            'node_name/namespace_name' : {
+                                            'name': 'node_name/namespace_name',
+                                            'type': 'namespace' or 'node',
+                                            'nodes': [list of nodes],
+                                            'dependencies': [list of dependencies]}
+        }
+        This test checks for the 'nodes' key in the dictionary which should be a list of nodes.
+        """
+        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
+    ):
+        """Test for pipeline.grouped_nodes_by_namespace which returns a dictionary with the following structure:
+        {
+            'node_name/namespace_name' : {
+                                            'name': 'node_name/namespace_name',
+                                            'type': 'namespace' or 'node',
+                                            'nodes': [list of nodes],
+                                            'dependencies': [list of dependencies]}
+        }
+        This test checks for the 'dependencies' in the dictionary which is a list of nodes/namespaces the group depends on.
+        """
+        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 +870,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()