Repository: kedro-org/kedro
Pull request: Add new function to specify namespace for `KedroSession` and `kedro run`
> **NOTE:** Kedro datasets are moving from `kedro.extras.datasets` to a separate `kedro-datasets` package in
> [`kedro-plugins` repository](https://github.com/kedro-org/kedro-plugins). Any changes to the dataset implementations
> should be done by opening a pull request in that repository.
## Description
<!-- Why was this PR created? -->
Close #2145

## Description

Currently there's a way to select different sections from a pipeline to run, but selecting by namespace is missing. We should add a way for users to run a specific subpipeline (namespace) by executing this in the terminal:

```
kedro run --namespace data_engineering.preprocessing
```

```
kedro run -ns data_engineering
```

## Development notes

<!-- What have you changed, and how has this been tested? -->
* Update `session.py` to add `namespace` as record data and modify the test accordingly
* Update `project.py` to add new option for the `-ns` or `--namespace` 

## How this is tested?
It is tested manually with a modified version of `spaceflight`, since the current starters do not use namespace, I followed [this modular pipeline documentation](https://kedro.readthedocs.io/en/stable/tutorial/add_another_pipeline.html#optional-modular-pipelines) for testing.

Steps
* `kedro new -s spaceflights`
* Follow along the page to modify it as a namesapce pipeline (https://kedro.readthedocs.io/en/stable/tutorial/add_another_pipeline.html#optional-modular-pipelines)

## Kedro CLI
`kedro run --namespace active_modelling` - 3 nodes should be run
`kedro run --namespace active_modelling --pipeline data_science` - 3 3 nodes should be run, since the namespace only available in one pipeline
`kedro run --namespace 1234` -- Error `ValueError: Pipeline does not contain nodes with namespace 
'1234'`

## KedroSession
Similar tests are run with the `kedro ipython`. run `%reload_kedro` to refresh a session
* `session.run(namespace="active_modelling")`
* `session.run(namespace="active_modelling", pipeline_name="data_science")`
* `session.run(namespace="1234")` -- error `ValueError: Pipeline does not contain nodes with namespace 
'1234'`


# Questions for Reviewers
1. Do we need a specific test for the `namespace` option? I don't think it's needed as it is more or less equal to other arguments and the filtering function is tested in the `test_pipeline.py` already.
2. The ordering for these cli options and `KedroSession.run` is kind of arbitrary, for `KedroSession` it is the code API so it makes sense to add the new option as the last argument to keep it backward compatible. Noted some arguments have inconsistent names.  For example `kedro run --pipeline` vs `KedroSession.run(pipline_name=xxx)`

## Checklist

- [ ] Read the [contributing](https://github.com/kedro-org/kedro/blob/main/CONTRIBUTING.md) guidelines
- [ ] Opened this PR as a 'Draft Pull Request' if it is work-in-progress
- [ ] Updated the documentation to reflect the code changes
- [ ] Added a description of this change in the [`RELEASE.md`](https://github.com/kedro-org/kedro/blob/main/RELEASE.md) file
- [ ] Added tests to cover my changes


<a href="https://gitpod.io/#https://github.com/kedro-org/kedro/pull/2306"><img src="https://gitpod.io/button/open-in-gitpod.svg"/></a>



Problem statement:
Add an option to run a specific namespace in `KedroSession` run and `kedro run`
## Description

Currently there's a way to select different sections from a pipeline to run, but selecting by namespace is missing. We should add a way for users to run a specific subpipeline (namespace) by executing this in the terminal:

```
kedro run --namespace data_engineering.preprocessing
```

```
kedro run -ns data_engineering
```


This will run all nodes whose namespace starts with `data_engineering.preprocessing`, regardless of depth. In order to implement this, we'd need to provide a similar option in `KedroSession`.
The `Pipeline` object already has the relevant filter `only_nodes_with_namespace` which can be used for implementing this functionality.

## Context

This change is important to providing a nice way to convert Kedro pipelines into orchestrator pipelines. Currently most conversions of Kedro pipelines happen on a node to node mapping,
which is far from ideal. Orchestrator run pipelines tend to be much less granular than Kedro pipelines and this 1-to-1 translation produces suboptimal executions. By enabling running pipelines
by namespace, we will make it easier to create much more clever deployment plugins, which will take advantage of the topology of the underlying pipelines better.

## Possible Implementation

To be decided by the team.

## Possible Alternatives

Hints from issue discussion:
This should be `multiple=False` just like pipeline

Commit message: Fix broken tests and update new test for namespace argument

Signed-off-by: Nok Chan <nok.lam.chan@quantumblack.com>

Patch to review:
diff --git a/RELEASE.md b/RELEASE.md
index c28405b148..3ba79aef7c 100644
--- a/RELEASE.md
+++ b/RELEASE.md
@@ -19,6 +19,7 @@
   * Save node outputs after every `yield` before proceeding with next chunk.
 * Fixed incorrect parsing of Azure Data Lake Storage Gen2 URIs used in datasets.
 * Added support for loading credentials from environment variables using `OmegaConfigLoader`.
+* Added new ability to filter by namespace for `KedroSession.run` and `kedro run` with `--namespace`.
 
 ## Bug fixes and other changes
 * Commas surrounded by square brackets (only possible for nodes with default names) will no longer split the arguments to `kedro run` options which take a list of nodes as inputs (`--from-nodes` and `--to-nodes`).
diff --git a/kedro/framework/cli/project.py b/kedro/framework/cli/project.py
index ad8e8e3e54..db0e26806b 100644
--- a/kedro/framework/cli/project.py
+++ b/kedro/framework/cli/project.py
@@ -53,6 +53,8 @@
 override the loaded ones."""
 PIPELINE_ARG_HELP = """Name of the registered pipeline to run.
 If not set, the '__default__' pipeline is run."""
+NAMESPACE_ARG_HELP = """Name of the namespace to run.
+If not set, no namespace is used."""
 PARAMS_ARG_HELP = """Specify extra parameters that you want to pass
 to the context initialiser. Items must be separated by comma, keys - by colon or equals sign,
 example: param1=value1,param2=value2. Each parameter is split by the first comma,
@@ -353,6 +355,7 @@ def activate_nbstripout(
     callback=_reformat_load_versions,
 )
 @click.option("--pipeline", "-p", type=str, default=None, help=PIPELINE_ARG_HELP)
+@click.option("--namespace", "-ns", type=str, default=None, help=NAMESPACE_ARG_HELP)
 @click.option(
     "--config",
     "-c",
@@ -372,7 +375,7 @@ def activate_nbstripout(
     help=PARAMS_ARG_HELP,
     callback=_split_params,
 )
-# pylint: disable=too-many-arguments,unused-argument
+# pylint: disable=too-many-arguments,unused-argument,too-many-locals
 def run(
     tag,
     env,
@@ -388,6 +391,7 @@ def run(
     config,
     conf_source,
     params,
+    namespace,
 ):
     """Run the pipeline."""
 
@@ -409,4 +413,5 @@ def run(
             to_outputs=to_outputs,
             load_versions=load_version,
             pipeline_name=pipeline,
+            namespace=namespace,
         )
diff --git a/kedro/framework/session/session.py b/kedro/framework/session/session.py
index a435ffd582..50fa264548 100644
--- a/kedro/framework/session/session.py
+++ b/kedro/framework/session/session.py
@@ -313,6 +313,7 @@ def run(  # pylint: disable=too-many-arguments,too-many-locals
         from_inputs: Iterable[str] = None,
         to_outputs: Iterable[str] = None,
         load_versions: Dict[str, str] = None,
+        namespace: str = None,
     ) -> Dict[str, Any]:
         """Runs the pipeline with a specified runner.
 
@@ -336,6 +337,7 @@ def run(  # pylint: disable=too-many-arguments,too-many-locals
                 used as an end point of the new ``Pipeline``.
             load_versions: An optional flag to specify a particular dataset
                 version timestamp to load.
+            namespace: The namespace of the nodes that is being run.
         Raises:
             ValueError: If the named or `__default__` pipeline is not
                 defined by `register_pipelines`.
@@ -382,6 +384,7 @@ def run(  # pylint: disable=too-many-arguments,too-many-locals
             node_names=node_names,
             from_inputs=from_inputs,
             to_outputs=to_outputs,
+            node_namespace=namespace,
         )
 
         record_data = {
@@ -398,6 +401,7 @@ def run(  # pylint: disable=too-many-arguments,too-many-locals
             "load_versions": load_versions,
             "extra_params": extra_params,
             "pipeline_name": pipeline_name,
+            "namespace": namespace,
             "runner": getattr(runner, "__name__", str(runner)),
         }
 
diff --git a/tests/framework/cli/test_cli.py b/tests/framework/cli/test_cli.py
index bcfaf215ac..99982b6dfd 100644
--- a/tests/framework/cli/test_cli.py
+++ b/tests/framework/cli/test_cli.py
@@ -472,6 +472,7 @@ def test_run_successfully(
             to_outputs=[],
             load_versions={},
             pipeline_name=None,
+            namespace=None,
         )
 
         runner = fake_session.run.call_args_list[0][1]["runner"]
@@ -484,8 +485,11 @@ def test_run_with_pipeline_filters(
         from_nodes = ["--from-nodes", "splitting_data"]
         to_nodes = ["--to-nodes", "training_model"]
         tags = ["--tag", "de"]
+        namespace = ["--namespace", "fake_namespace"]
         result = CliRunner().invoke(
-            fake_project_cli, ["run", *from_nodes, *to_nodes, *tags], obj=fake_metadata
+            fake_project_cli,
+            ["run", *from_nodes, *to_nodes, *tags, *namespace],
+            obj=fake_metadata,
         )
         assert not result.exit_code
 
@@ -499,6 +503,7 @@ def test_run_with_pipeline_filters(
             to_outputs=[],
             load_versions={},
             pipeline_name=None,
+            namespace="fake_namespace",
         )
 
         runner = fake_session.run.call_args_list[0][1]["runner"]
@@ -522,6 +527,7 @@ def test_run_successfully_parallel(
             to_outputs=[],
             load_versions={},
             pipeline_name=None,
+            namespace=None,
         )
 
         runner = fake_session.run.call_args_list[0][1]["runner"]
@@ -561,6 +567,7 @@ def test_run_with_config(
             to_outputs=[],
             load_versions={},
             pipeline_name="pipeline1",
+            namespace=None,
         )
 
     @mark.parametrize(
@@ -604,6 +611,7 @@ def test_run_with_params_in_config(
             to_outputs=[],
             load_versions={},
             pipeline_name="pipeline1",
+            namespace=None,
         )
         mock_session_create.assert_called_once_with(
             env=mocker.ANY, conf_source=None, extra_params=expected
@@ -703,6 +711,7 @@ def test_reformat_load_versions(
             to_outputs=[],
             load_versions={ds: t},
             pipeline_name=None,
+            namespace=None,
         )
 
     def test_fail_reformat_load_versions(self, fake_project_cli, fake_metadata):
@@ -762,6 +771,7 @@ def test_safe_split_option_arguments(
             to_outputs=[],
             load_versions={},
             pipeline_name=None,
+            namespace=None,
         )
 
     def test_run_with_alternative_conf_source(self, fake_project_cli, fake_metadata):
diff --git a/tests/framework/session/test_session.py b/tests/framework/session/test_session.py
index 9f6d623150..1cccce0426 100644
--- a/tests/framework/session/test_session.py
+++ b/tests/framework/session/test_session.py
@@ -623,6 +623,7 @@ def test_run(
             "load_versions": None,
             "extra_params": {},
             "pipeline_name": fake_pipeline_name,
+            "namespace": None,
             "runner": mock_runner.__name__,
         }
 
@@ -692,6 +693,7 @@ def test_run_multiple_times(  # pylint: disable=too-many-locals
             "load_versions": None,
             "extra_params": {},
             "pipeline_name": fake_pipeline_name,
+            "namespace": None,
             "runner": mock_runner.__name__,
         }
 
@@ -779,6 +781,7 @@ def test_run_exception(  # pylint: disable=too-many-locals
             "load_versions": None,
             "extra_params": {},
             "pipeline_name": fake_pipeline_name,
+            "namespace": None,
             "runner": mock_runner.__name__,
         }
 
@@ -846,6 +849,7 @@ def test_run_broken_pipeline_multiple_times(  # pylint: disable=too-many-locals
             "load_versions": None,
             "extra_params": {},
             "pipeline_name": fake_pipeline_name,
+            "namespace": None,
             "runner": broken_runner.__name__,
         }
 


Code review comments:
diff_hunk: @@ -19,6 +19,7 @@
   * Save node outputs after every `yield` before proceeding with next chunk.
 * Fixed incorrect parsing of Azure Data Lake Storage Gen2 URIs used in datasets.
 * Added support for loading credentials from environment variables using `OmegaConfigLoader`.
+* Added new ability to filter by namespace for `KedroSession.run` and `kedro run` with `--namespace`.
path: RELEASE.md
text: @user1:
```suggestion
* Added new `--namespace` flag to `kedro run` to enable filtering by node namespace.
```
I don't think `KedroSession.run` is worth mentioning here just to keep release notes concise, since CLI `kedro run` is how 99% of kedro runs are implemented rather than the Python API.

diff_hunk: @@ -53,6 +53,8 @@
 override the loaded ones."""
 PIPELINE_ARG_HELP = """Name of the registered pipeline to run.
 If not set, the '__default__' pipeline is run."""
+NAMESPACE_ARG_HELP = """Name of the namespace to run.
+If not set, no namespace is used."""
path: kedro/framework/cli/project.py
text: @user1:
```suggestion
NAMESPACE_ARG_HELP = """Name of the node namespace to run."""
```
IMO the "no namespace is used" just makes this less clear since it can't be really explained fully in just a line.