add support for running multiple pipelines via --pipelines (#5296)

Mohmn · ankatiyar · web-flow · commit c6afa499eebb · 2026-01-26T13:44:39.000Z
* add support for running multiple pipelines via --pipelines

Signed-off-by: Mohmn &lt;naqashmohmin1@gmail.com&gt;

Signed-off-by: Mohmn &lt;naqashmohmin1@gmail.com&gt;

* Fix tests and release notes

Signed-off-by: Ankita Katiyar &lt;ankitakatiyar2401@gmail.com&gt;

---------

Signed-off-by: Mohmn &lt;naqashmohmin1@gmail.com&gt;
Signed-off-by: Ankita Katiyar &lt;110245118+ankatiyar@users.noreply.github.com&gt;
Signed-off-by: Ankita Katiyar &lt;ankitakatiyar2401@gmail.com&gt;
Co-authored-by: Ankita Katiyar &lt;110245118+ankatiyar@users.noreply.github.com&gt;
Co-authored-by: Ankita Katiyar &lt;ankitakatiyar2401@gmail.com&gt;
diff --git a/RELEASE.md b/RELEASE.md
@@ -2,6 +2,7 @@
 ## Major features and improvements
 * Added `@experimental` decorator to mark unstable or early-stage public APIs.
 * Added the new `support-agent-langgraph` starter. This starter contains pipelines that leverage LangGraph for agentic workflows and Langfuse or Opik for prompt management and tracing.
+* Added support for running multiple pipelines in a single Kedro session run via the `--pipelines` CLI option and `pipeline_names` argument in `KedroSession.run()` method.
 
 ## Experimental features
 * Added experimental `llm_context_node` and `LLMContextNode` for assembling LLMs, prompts, and tools into a runtime `LLMContext` within Kedro pipelines.
@@ -14,8 +15,9 @@
 ## Documentation changes
 * Added beginner-friendly notes on `uvx` installation.
 
-
 ## Community contributions
+Many thanks to the following Kedroids for contributing PRs to this release:
+* [Mohmn](https://github.com/Mohmn)
 
 # Release 1.1.1
 ## Bug fixes and other changes
diff --git a/kedro/framework/cli/project.py b/kedro/framework/cli/project.py
@@ -10,6 +10,7 @@
 import click
 
 from kedro.framework.cli.utils import (
+    KedroCliError,
     _check_module_importable,
     _config_file_callback,
     _split_load_versions,
@@ -53,6 +54,9 @@
 override the loaded ones."""
 PIPELINE_ARG_HELP = """Name of the registered pipeline to run.
 If not set, the '__default__' pipeline is run."""
+PIPELINES_ARG_HELP = """Comma-separated names of registered pipelines to run.
+Example: --pipelines data_engineering,feature_engineering
+If not set, the '__default__' pipeline is run."""
 NAMESPACES_ARG_HELP = """Run only node namespaces with specified names."""
 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,
@@ -200,6 +204,9 @@ def package(metadata: ProjectMetadata) -> None:
     callback=_split_load_versions,
 )
 @click.option("--pipeline", "-p", type=str, default=None, help=PIPELINE_ARG_HELP)
+@click.option(
+    "--pipelines", type=str, default="", help=PIPELINES_ARG_HELP, callback=split_string
+)
 @click.option(
     "--namespaces",
     "-ns",
@@ -244,6 +251,7 @@ def run(  # noqa: PLR0913
     to_outputs: str,
     load_versions: dict[str, str] | None,
     pipeline: str,
+    pipelines: list[str],
     config: str,
     conf_source: str,
     params: dict[str, Any],
@@ -252,6 +260,13 @@ def run(  # noqa: PLR0913
 ) -> dict[str, Any]:
     """Run the pipeline."""
 
+    if pipeline and pipelines:
+        raise KedroCliError(
+            "Options '--pipeline' and '--pipelines' cannot be used together"
+        )
+
+    pipelines_to_run = list(set(pipelines)) if pipelines else None
+
     runner_obj = load_obj(runner or "SequentialRunner", "kedro.runner")
     tuple_tags = tuple(tags)
     tuple_node_names = tuple(node_names)
@@ -269,6 +284,7 @@ def run(  # noqa: PLR0913
             to_outputs=to_outputs,
             load_versions=load_versions,
             pipeline_name=pipeline,
+            pipeline_names=pipelines_to_run,
             namespaces=namespaces,
             only_missing_outputs=only_missing_outputs,
         )
diff --git a/kedro/framework/session/session.py b/kedro/framework/session/session.py
@@ -26,6 +26,7 @@
 )
 from kedro.io.core import generate_timestamp
 from kedro.io.data_catalog import SharedMemoryDataCatalog
+from kedro.pipeline.pipeline import Pipeline
 from kedro.runner import AbstractRunner, ParallelRunner, SequentialRunner
 from kedro.utils import find_kedro_project
 
@@ -281,6 +282,7 @@ def __exit__(self, exc_type: Any, exc_value: Any, tb_: Any) -> None:
     def run(  # noqa: PLR0913
         self,
         pipeline_name: str | None = None,
+        pipeline_names: list[str] | None = None,
         tags: Iterable[str] | None = None,
         runner: AbstractRunner | None = None,
         node_names: Iterable[str] | None = None,
@@ -296,6 +298,7 @@ def run(  # noqa: PLR0913
 
         Args:
             pipeline_name: Name of the pipeline that is being run.
+            pipeline_names: Name of the pipelines that is being run.
             tags: An optional list of node tags which should be used to
                 filter the nodes of the ``Pipeline``. If specified, only the nodes
                 containing *any* of these tags will be run.
@@ -330,6 +333,12 @@ def run(  # noqa: PLR0913
         # Report project name
         project_name = self._package_name or self._project_path.name
         self._logger.info("Kedro project %s", project_name)
+        if pipeline_name:
+            self._logger.warning(
+                "`pipeline_name` is deprecated and will be removed in a future release. "
+                "Please use `pipeline_names` instead."
+            )
+            pipeline_names = [pipeline_name]
 
         if self._run_called:
             raise KedroSessionError(
@@ -343,18 +352,19 @@ def run(  # noqa: PLR0913
         runtime_params = self.store.get("runtime_params") or {}
         context = self.load_context()
 
-        name = pipeline_name or "__default__"
-
-        try:
-            pipeline = pipelines[name]
-        except KeyError as exc:
-            raise ValueError(
-                f"Failed to find the pipeline named '{name}'. "
-                f"It needs to be generated and returned "
-                f"by the 'register_pipelines' function."
-            ) from exc
-
-        filtered_pipeline = pipeline.filter(
+        names = pipeline_names or ["__default__"]
+        combined_pipelines = Pipeline([])
+        for name in names:
+            try:
+                combined_pipelines += pipelines[name]
+            except KeyError as exc:
+                raise ValueError(
+                    f"Failed to find the pipeline named '{name}'. "
+                    f"It needs to be generated and returned "
+                    f"by the 'register_pipelines' function."
+                ) from exc
+
+        filtered_pipeline = combined_pipelines.filter(
             tags=tags,
             from_nodes=from_nodes,
             to_nodes=to_nodes,
@@ -377,7 +387,7 @@ def run(  # noqa: PLR0913
             "to_outputs": to_outputs,
             "load_versions": load_versions,
             "runtime_params": runtime_params,
-            "pipeline_name": pipeline_name,
+            "pipeline_names": pipeline_names,
             "namespaces": namespaces,
             "runner": getattr(runner, "__name__", str(runner)),
             "only_missing_outputs": only_missing_outputs,
diff --git a/tests/framework/cli/test_cli.py b/tests/framework/cli/test_cli.py
@@ -577,6 +577,7 @@ def test_run_successfully(
             to_outputs=[],
             load_versions={},
             pipeline_name=None,
+            pipeline_names=None,
             namespaces=[],
             only_missing_outputs=False,
         )
@@ -618,6 +619,7 @@ def test_run_specific_nodes(
             to_outputs=[],
             load_versions={},
             pipeline_name=None,
+            pipeline_names=None,
             namespaces=[],
             only_missing_outputs=False,
         )
@@ -659,6 +661,7 @@ def test_run_with_tags(
             to_outputs=[],
             load_versions={},
             pipeline_name=None,
+            pipeline_names=None,
             namespaces=[],
             only_missing_outputs=False,
         )
@@ -691,6 +694,7 @@ def test_run_with_pipeline_filters(
             to_outputs=[],
             load_versions={},
             pipeline_name=None,
+            pipeline_names=None,
             namespaces=["fake_namespace"],
             only_missing_outputs=False,
         )
@@ -716,6 +720,7 @@ def test_run_successfully_parallel(
             to_outputs=[],
             load_versions={},
             pipeline_name=None,
+            pipeline_names=None,
             namespaces=[],
             only_missing_outputs=False,
         )
@@ -757,10 +762,65 @@ def test_run_with_config(
             to_outputs=[],
             load_versions={},
             pipeline_name="pipeline1",
+            pipeline_names=None,
             namespaces=[],
             only_missing_outputs=False,
         )
 
+    def test_run_multiple_pipelines(
+        self, fake_project_cli, fake_metadata, fake_session
+    ):
+        result = CliRunner().invoke(
+            fake_project_cli,
+            ["run", "--pipelines", "pipe1,pipe2"],
+            obj=fake_metadata,
+        )
+
+        assert not result.exit_code
+        assert fake_session.run.call_count == 1
+        assert fake_session.run.call_args.kwargs["pipeline_name"] is None
+
+        pipelines = fake_session.run.call_args.kwargs["pipeline_names"]
+        assert "pipe1" in pipelines
+        assert "pipe2" in pipelines
+
+    def test_run_multiple_pipelines_with_duplicate_name(
+        self, fake_project_cli, fake_metadata, fake_session
+    ):
+        result = CliRunner().invoke(
+            fake_project_cli,
+            ["run", "--pipelines", "pipe1,pipe1"],
+            obj=fake_metadata,
+        )
+
+        assert not result.exit_code
+        assert fake_session.run.call_count == 1
+        assert fake_session.run.call_args.kwargs["pipeline_name"] is None
+        assert fake_session.run.call_args.kwargs["pipeline_names"] == ["pipe1"]
+
+    def test_pipeline_and_pipelines_mutually_exclusive(
+        self, fake_project_cli, fake_metadata
+    ):
+        result = CliRunner().invoke(
+            fake_project_cli,
+            ["run", "--pipeline", "pipe1", "--pipelines", "pipe2"],
+            obj=fake_metadata,
+        )
+
+        assert result.exit_code != 0
+        assert "cannot be used together" in result.output.lower()
+
+    def test_pipeline_name_deprecation_warning(
+        self, fake_project_cli, fake_metadata, caplog
+    ):
+        CliRunner().invoke(
+            fake_project_cli,
+            ["run", "--pipeline", "pipe1"],
+            obj=fake_metadata,
+        )
+
+        assert "deprecated" in caplog.text.lower()
+
     @mark.parametrize("config_flag", ["--config", "-c"])
     def test_run_with_invalid_config(
         self,
@@ -823,6 +883,7 @@ def test_run_with_params_in_config(
             to_outputs=[],
             load_versions={},
             pipeline_name="pipeline1",
+            pipeline_names=None,
             namespaces=[],
             only_missing_outputs=False,
         )
@@ -938,6 +999,7 @@ def test_split_load_versions(
             to_outputs=[],
             load_versions=lv_dict,
             pipeline_name=None,
+            pipeline_names=None,
             namespaces=[],
             only_missing_outputs=False,
         )
@@ -1001,6 +1063,7 @@ def test_safe_split_option_arguments(
             to_outputs=[],
             load_versions={},
             pipeline_name=None,
+            pipeline_names=None,
             namespaces=[],
             only_missing_outputs=False,
         )
diff --git a/tests/framework/session/test_session.py b/tests/framework/session/test_session.py
