Repository: deepset-ai/haystack
Pull request: Generate code from pipeline (pipeline.to_code())
**Proposed changes**:
- add methods `to_code()` and `to_notebook_cell()` to `Pipeline`
- `to_code()` returns the code as string 
- `to_notebook_cell` creates a new cell containing the code
- param `pipeline_variable_name` controls the name of the pipeline variable to be generated
- param `generate_imports` controls whether respecting imports should be created along with the code
- param `add_comment` controls whether to show a comment depicting that it has been generated before the code. In `to_code()` defaults to `False`, in `to_notebook_cell()` defaults to `True`.

Main flow for notebook users:
```python
p = Pipeline.load_from_deepset_cloud(pipeline_config_name=MY_PIPELINE_CONFIG_NAME)
```

```python
p.to_notebook_cell()
```

```python
# This code has been generated.
from haystack.document_stores import DeepsetCloudDocumentStore
from haystack.nodes import ElasticsearchRetriever

deepset_cloud_document_store = DeepsetCloudDocumentStore(index="document_search_pipeline_2")
retriever = ElasticsearchRetriever(document_store=deepset_cloud_document_store, top_k=5)

pipeline = Pipeline()
pipeline.add_node(component=retriever, name="Retriever", inputs=["Query"])
```

Main flow for backend code users:
```python
p = Pipeline.load_from_yaml(path=MY_YAML_PATH)
code = p.to_code()
# do whatever you like with the code
```

**Status (please check what you already did)**:
- [x] First draft (up for discussions & feedback)
- [x] Final code
- [x] Added tests
- [x] Updated documentation

closes https://github.com/deepset-ai/haystack/issues/2195


Problem statement:
pipeline.to_code()
**What?**
As a user I want to be able to manipulate a pipeline besides isomorphic changes that are already possible through `pipeline.set_node()`.

**How?**
To add, change or delete one or multiple nodes of a pipeline other parts of the pipeline will be affected (e.g. different input, output). So the best and most transparent way is to generate the code from the existing pipeline, enabling the user to change anything she wishes to.
We implement `set_node()` which
- in Jupyter notebook creates a new cell with the whole pipeline code
- outside of Jupyter returns a string of the whole pipeline code

Hints from issue discussion:

Commit message: Update Documentation & Code Style

Patch to review:
diff --git a/docs/_src/api/api/pipelines.md b/docs/_src/api/api/pipelines.md
index b68750be81..48bbb7eda8 100644
--- a/docs/_src/api/api/pipelines.md
+++ b/docs/_src/api/api/pipelines.md
@@ -37,6 +37,44 @@ Returns a configuration for the Pipeline that can be used with `BasePipeline.loa
 
 - `return_defaults`: whether to output parameters that have the default values.
 
+<a id="base.BasePipeline.to_code"></a>
+
+#### to\_code
+
+```python
+def to_code(pipeline_variable_name: str = "pipeline", generate_imports: bool = True, add_comment: bool = False) -> str
+```
+
+Returns the code to create this pipeline as string.
+
+**Arguments**:
+
+- `pipeline_variable_name`: The variable name of the generated pipeline.
+Default value is 'pipeline'.
+- `generate_imports`: Whether to include the required import statements into the code.
+Default value is True.
+- `add_comment`: Whether to add a preceding comment that this code has been generated.
+Default value is False.
+
+<a id="base.BasePipeline.to_notebook_cell"></a>
+
+#### to\_notebook\_cell
+
+```python
+def to_notebook_cell(pipeline_variable_name: str = "pipeline", generate_imports: bool = True, add_comment: bool = True)
+```
+
+Creates a new notebook cell with the code to create this pipeline.
+
+**Arguments**:
+
+- `pipeline_variable_name`: The variable name of the generated pipeline.
+Default value is 'pipeline'.
+- `generate_imports`: Whether to include the required import statements into the code.
+Default value is True.
+- `add_comment`: Whether to add a preceding comment that this code has been generated.
+Default value is True.
+
 <a id="base.BasePipeline.load_from_config"></a>
 
 #### load\_from\_config
diff --git a/docs/_src/api/openapi/openapi.json b/docs/_src/api/openapi/openapi.json
index af6d9ce457..f5fcee5b74 100644
--- a/docs/_src/api/openapi/openapi.json
+++ b/docs/_src/api/openapi/openapi.json
@@ -278,7 +278,7 @@
                     "file-upload"
                 ],
                 "summary": "Upload File",
-                "description": "You can use this endpoint to upload a file for indexing\n(see [http://localhost:3000/guides/rest-api#indexing-documents-in-the-haystack-rest-api-document-store]).",
+                "description": "You can use this endpoint to upload a file for indexing\n(see https://haystack.deepset.ai/guides/rest-api#indexing-documents-in-the-haystack-rest-api-document-store).",
                 "operationId": "upload_file_file_upload_post",
                 "requestBody": {
                     "content": {
diff --git a/haystack/pipelines/base.py b/haystack/pipelines/base.py
index 760883d9b8..84c58ffa27 100644
--- a/haystack/pipelines/base.py
+++ b/haystack/pipelines/base.py
@@ -1,4 +1,5 @@
 from __future__ import annotations
+import sys
 from typing import Dict, List, Optional, Any
 
 import copy
@@ -6,6 +7,7 @@
 import inspect
 import logging
 import os
+import re
 import traceback
 import numpy as np
 import pandas as pd
@@ -38,6 +40,10 @@
 
 
 ROOT_NODE_TO_PIPELINE_NAME = {"query": "query", "file": "indexing"}
+CAMEL_CASE_TO_SNAKE_CASE_REGEX = re.compile(r"(?<!^)(?=[A-Z])")
+MODULE_NOT_FOUND = "MODULE_NOT_FOUND"
+CODE_GEN_ALLOWED_IMPORTS = ["haystack.document_stores", "haystack.nodes"]
+CODE_GEN_DEFAULT_COMMENT = "This code has been generated."
 
 
 class RootNode(BaseComponent):
@@ -68,6 +74,54 @@ def get_config(self, return_defaults: bool = False) -> dict:
         """
         raise NotImplementedError
 
+    def to_code(
+        self,
+        pipeline_variable_name: str = "pipeline",
+        generate_imports: bool = True,
+        add_comment: bool = False,
+    ) -> str:
+        """
+        Returns the code to create this pipeline as string.
+
+        :param pipeline_variable_name: The variable name of the generated pipeline.
+                                       Default value is 'pipeline'.
+        :param generate_imports: Whether to include the required import statements into the code.
+                                 Default value is True.
+        :param add_comment: Whether to add a preceding comment that this code has been generated.
+                            Default value is False.
+        """
+        code = _PipelineCodeGen.generate_code(
+            pipeline=self,
+            pipeline_variable_name=pipeline_variable_name,
+            generate_imports=generate_imports,
+            comment=CODE_GEN_DEFAULT_COMMENT if add_comment else None,
+        )
+        return code
+
+    def to_notebook_cell(
+        self,
+        pipeline_variable_name: str = "pipeline",
+        generate_imports: bool = True,
+        add_comment: bool = True,
+    ):
+        """
+        Creates a new notebook cell with the code to create this pipeline.
+
+        :param pipeline_variable_name: The variable name of the generated pipeline.
+                                       Default value is 'pipeline'.
+        :param generate_imports: Whether to include the required import statements into the code.
+                                 Default value is True.
+        :param add_comment: Whether to add a preceding comment that this code has been generated.
+                            Default value is True.
+        """
+        code = self.to_code(
+            pipeline_variable_name=pipeline_variable_name, generate_imports=generate_imports, add_comment=add_comment
+        )
+        try:
+            get_ipython().set_next_input(code)  # type: ignore
+        except NameError:
+            logger.error("Could not create notebook cell. Make sure you're running in a notebook environment.")
+
     @classmethod
     def load_from_config(
         cls, pipeline_config: Dict, pipeline_name: Optional[str] = None, overwrite_with_env_variables: bool = True
@@ -335,7 +389,7 @@ def save_to_deepset_cloud(
             logger.info(f"Pipeline config '{pipeline_config_name}' successfully created.")
 
     @classmethod
-    def _get_pipeline_definition(cls, pipeline_config: Dict, pipeline_name: Optional[str] = None):
+    def _get_pipeline_definition(cls, pipeline_config: Dict, pipeline_name: Optional[str] = None) -> dict:
         """
         Get the definition of Pipeline from a given pipeline config. If the config contains more than one Pipeline,
         then the pipeline_name must be supplied.
@@ -357,7 +411,7 @@ def _get_pipeline_definition(cls, pipeline_config: Dict, pipeline_name: Optional
         return pipeline_definition
 
     @classmethod
-    def _get_component_definitions(cls, pipeline_config: Dict, overwrite_with_env_variables: bool):
+    def _get_component_definitions(cls, pipeline_config: Dict, overwrite_with_env_variables: bool) -> dict:
         """
         Returns the definitions of all components from a given pipeline config.
 
@@ -1040,85 +1094,6 @@ def get_config(self, return_defaults: bool = False) -> dict:
         config = {"components": list(components.values()), "pipelines": list(pipelines.values()), "version": "0.8"}
         return config
 
-    def _format_document_answer(self, document_or_answer: dict):
-        return "\n \t".join([f"{name}: {value}" for name, value in document_or_answer.items()])
-
-    def _format_wrong_sample(self, query: dict):
-        metrics = "\n \t".join([f"{name}: {value}" for name, value in query["metrics"].items()])
-        documents = "\n\n \t".join([self._format_document_answer(doc) for doc in query.get("documents", [])])
-        documents = f"Documents: \n \t{documents}\n" if len(documents) > 0 else ""
-        answers = "\n\n \t".join([self._format_document_answer(answer) for answer in query.get("answers", [])])
-        answers = f"Answers: \n \t{answers}\n" if len(answers) > 0 else ""
-        gold_document_ids = "\n \t".join(query["gold_document_ids"])
-        gold_answers = "\n \t".join(query.get("gold_answers", []))
-        gold_answers = f"Gold Answers: \n \t{gold_answers}\n" if len(gold_answers) > 0 else ""
-        s = (
-            f"Query: \n \t{query['query']}\n"
-            f"{gold_answers}"
-            f"Gold Document Ids: \n \t{gold_document_ids}\n"
-            f"Metrics: \n \t{metrics}\n"
-            f"{answers}"
-            f"{documents}"
-            f"_______________________________________________________"
-        )
-        return s
-
-    def _format_wrong_samples_node(self, node_name: str, wrong_samples_formatted: str):
-        s = (
-            f"                Wrong {node_name} Examples\n"
-            f"=======================================================\n"
-            f"{wrong_samples_formatted}\n"
-            f"=======================================================\n"
-        )
-        return s
-
-    def _format_wrong_samples_report(self, eval_result: EvaluationResult, n_wrong_examples: int = 3):
-        examples = {
-            node: eval_result.wrong_examples(node, doc_relevance_col="gold_id_or_answer_match", n=n_wrong_examples)
-            for node in eval_result.node_results.keys()
-        }
-        examples_formatted = {
-            node: "\n".join([self._format_wrong_sample(example) for example in examples])
-            for node, examples in examples.items()
-        }
-
-        return "\n".join(
-            [self._format_wrong_samples_node(node, examples) for node, examples in examples_formatted.items()]
-        )
-
-    def _format_pipeline_node(self, node: str, calculated_metrics: dict):
-        node_metrics: dict = {}
-        for metric_mode in calculated_metrics:
-            for metric, value in calculated_metrics[metric_mode].get(node, {}).items():
-                node_metrics[f"{metric}{metric_mode}"] = value
-
-        node_metrics_formatted = "\n".join(
-            sorted([f"                        | {metric}: {value:5.3}" for metric, value in node_metrics.items()])
-        )
-        node_metrics_formatted = f"{node_metrics_formatted}\n" if len(node_metrics_formatted) > 0 else ""
-        s = (
-            f"                      {node}\n"
-            f"                        |\n"
-            f"{node_metrics_formatted}"
-            f"                        |"
-        )
-        return s
-
-    def _format_pipeline_overview(self, calculated_metrics: dict):
-        pipeline_overview = "\n".join(
-            [self._format_pipeline_node(node, calculated_metrics) for node in self.graph.nodes]
-        )
-        s = (
-            f"================== Evaluation Report ==================\n"
-            f"=======================================================\n"
-            f"                   Pipeline Overview\n"
-            f"=======================================================\n"
-            f"{pipeline_overview}\n"
-            f"                      Output\n"
-            f"=======================================================\n"
-        )
-        return s
-
     def print_eval_report(
         self,
         eval_result: EvaluationResult,
@@ -1132,36 +1107,10 @@ def print_eval_report(
         :param n_wrong_examples: The number of worst queries to show.
         :param metrics_filter: The metrics to show per node. If None all metrics will be shown.
         """
-        if any(degree > 1 for node, degree in self.graph.out_degree):
-            logger.warning("Pipelines with junctions are currently not supported.")
-            return
-
-        calculated_metrics = {
-            "": eval_result.calculate_metrics(doc_relevance_col="gold_id_or_answer_match"),
-            "_top_1": eval_result.calculate_metrics(
-                doc_relevance_col="gold_id_or_answer_match", simulated_top_k_reader=1
-            ),
-            " upper bound": eval_result.calculate_metrics(
-                doc_relevance_col="gold_id_or_answer_match", eval_mode="isolated"
-            ),
-        }
-
-        if metrics_filter is not None:
-            for metric_mode in calculated_metrics:
-                calculated_metrics[metric_mode] = {
-                    node: metrics
-                    if node not in metrics_filter
-                    else {metric: value for metric, value in metrics.items() if metric in metrics_filter[node]}
-                    for node, metrics in calculated_metrics[metric_mode].items()
-                }
-
-        pipeline_overview = self._format_pipeline_overview(calculated_metrics)
-        wrong_samples_report = self._format_wrong_samples_report(
-            eval_result=eval_result, n_wrong_examples=n_wrong_examples
+        _PipelineEvalReportGen.print_eval_report(
+            eval_result=eval_result, pipeline=self, n_wrong_examples=n_wrong_examples, metrics_filter=metrics_filter
         )
 
-        print(f"{pipeline_overview}\n" f"{wrong_samples_report}")
-
 
 class RayPipeline(Pipeline):
     """
@@ -1446,3 +1395,257 @@ def __call__(self, *args, **kwargs):
         Ray calls this method which is then re-directed to the corresponding component's run().
         """
         return self.node._dispatch_run(*args, **kwargs)
+
+
+class _PipelineCodeGen:
+    @classmethod
+    def _camel_to_snake_case(cls, input: str) -> str:
+        return CAMEL_CASE_TO_SNAKE_CASE_REGEX.sub("_", input).lower()
+
+    @classmethod
+    def generate_code(
+        cls,
+        pipeline: BasePipeline,
+        pipeline_variable_name: str = "pipeline",
+        generate_imports: bool = True,
+        comment: Optional[str] = None,
+    ) -> str:
+        pipeline_config = pipeline.get_config()
+        component_definitions = pipeline._get_component_definitions(
+            pipeline_config=pipeline_config, overwrite_with_env_variables=False
+        )
+        component_variable_names = {name: cls._camel_to_snake_case(name) for name in component_definitions.keys()}
+        pipeline_definition = pipeline._get_pipeline_definition(pipeline_config=pipeline_config)
+
+        code_parts = []
+        if generate_imports:
+            types_to_import = [component["type"] for component in component_definitions.values()]
+            imports_code = cls._generate_imports_code(types_to_import=types_to_import)
+            code_parts.append(imports_code)
+
+        components_code = cls._generate_components_code(
+            component_definitions=component_definitions, component_variable_names=component_variable_names
+        )
+        pipeline_code = cls._generate_pipeline_code(
+            pipeline_definition=pipeline_definition,
+            component_variable_names=component_variable_names,
+            pipeline_variable_name=pipeline_variable_name,
+        )
+
+        code_parts.append(components_code)
+        code_parts.append(pipeline_code)
+        code = "\n\n".join(code_parts)
+
+        if comment:
+            comment = re.sub(r"^(#\s)?", "# ", comment, flags=re.MULTILINE)
+            code = "\n".join([comment, code])
+
+        return code
+
+    @classmethod
+    def _generate_pipeline_code(
+        cls, pipeline_definition: dict, component_variable_names: Dict[str, str], pipeline_variable_name: str
+    ) -> str:
+        code_lines = [f"{pipeline_variable_name} = Pipeline()"]
+        for node in pipeline_definition["nodes"]:
+            node_name = node["name"]
+            component_variable_name = component_variable_names[node_name]
+            inputs = ", ".join(f'"{name}"' for name in node["inputs"])
+            code_lines.append(
+                f'{pipeline_variable_name}.add_node(component={component_variable_name}, name="{node_name}", inputs=[{inputs}])'
+            )
+
+        code = "\n".join(code_lines)
+        return code
+
+    @classmethod
+    def _generate_components_code(cls, component_definitions: dict, component_variable_names: Dict[str, str]) -> str:
+        code = ""
+        declarations = {}
+        dependency_map = {}
+        for name, definition in component_definitions.items():
+            variable_name = component_variable_names[name]
+            class_name = definition["type"]
+            param_value_dict = {
+                key: component_variable_names.get(value, f'"{value}"') if type(value) == str else value
+                for key, value in definition["params"].items()
+            }
+            init_args = ", ".join(f"{key}={value}" for key, value in param_value_dict.items())
+            declarations[name] = f"{variable_name} = {class_name}({init_args})"
+            dependency_map[name] = [
+                param_value for param_value in definition["params"].values() if param_value in component_variable_names
+            ]
+
+        ordered_components = cls._order_components(dependency_map=dependency_map)
+        ordered_declarations = [declarations[component] for component in ordered_components]
+        code = "\n".join(ordered_declarations)
+        return code
+
+    @classmethod
+    def _generate_imports_code(cls, types_to_import: List[str]) -> str:
+        code_lines = []
+        importable_classes = {
+            name: mod
+            for mod in CODE_GEN_ALLOWED_IMPORTS
+            for name, obj in inspect.getmembers(sys.modules[mod])
+            if inspect.isclass(obj)
+        }
+
+        imports_by_module: Dict[str, List[str]] = {}
+        for t in types_to_import:
+            mod = importable_classes.get(t, MODULE_NOT_FOUND)
+            if mod in imports_by_module:
+                imports_by_module[mod].append(t)
+            else:
+                imports_by_module[mod] = [t]
+
+        for mod in sorted(imports_by_module.keys()):
+            sorted_types = sorted(set(imports_by_module[mod]))
+            import_types = ", ".join(sorted_types)
+            line_prefix = "# " if mod == MODULE_NOT_FOUND else ""
+            code_lines.append(f"{line_prefix}from {mod} import {import_types}")
+
+        code = "\n".join(code_lines)
+        return code
+
+    @classmethod
+    def _order_components(
+        cls, dependency_map: Dict[str, List[str]], components_to_order: Optional[List[str]] = None
+    ) -> List[str]:
+        ordered_components = []
+        if components_to_order is None:
+            components_to_order = list(dependency_map.keys())
+        for component in components_to_order:
+            dependencies = dependency_map[component]
+            ordered_dependencies = cls._order_components(
+                dependency_map=dependency_map, components_to_order=dependencies
+            )
+            ordered_components += [d for d in ordered_dependencies if d not in ordered_components]
+            if component not in ordered_components:
+                ordered_components.append(component)
+        return ordered_components
+
+
+class _PipelineEvalReportGen:
+    @classmethod
+    def print_eval_report(
+        cls,
+        eval_result: EvaluationResult,
+        pipeline: Pipeline,
+        n_wrong_examples: int = 3,
+        metrics_filter: Optional[Dict[str, List[str]]] = None,
+    ):
+        if any(degree > 1 for node, degree in pipeline.graph.out_degree):
+            logger.warning("Pipelines with junctions are currently not supported.")
+            return
+
+        calculated_metrics = {
+            "": eval_result.calculate_metrics(doc_relevance_col="gold_id_or_answer_match"),
+            "_top_1": eval_result.calculate_metrics(
+                doc_relevance_col="gold_id_or_answer_match", simulated_top_k_reader=1
+            ),
+            " upper bound": eval_result.calculate_metrics(
+                doc_relevance_col="gold_id_or_answer_match", eval_mode="isolated"
+            ),
+        }
+
+        if metrics_filter is not None:
+            for metric_mode in calculated_metrics:
+                calculated_metrics[metric_mode] = {
+                    node: metrics
+                    if node not in metrics_filter
+                    else {metric: value for metric, value in metrics.items() if metric in metrics_filter[node]}
+                    for node, metrics in calculated_metrics[metric_mode].items()
+                }
+
+        pipeline_overview = cls._format_pipeline_overview(calculated_metrics=calculated_metrics, pipeline=pipeline)
+        wrong_samples_report = cls._format_wrong_samples_report(
+            eval_result=eval_result, n_wrong_examples=n_wrong_examples
+        )
+
+        print(f"{pipeline_overview}\n" f"{wrong_samples_report}")
+
+    @classmethod
+    def _format_document_answer(cls, document_or_answer: dict):
+        return "\n \t".join([f"{name}: {value}" for name, value in document_or_answer.items()])
+
+    @classmethod
+    def _format_wrong_sample(cls, query: dict):
+        metrics = "\n \t".join([f"{name}: {value}" for name, value in query["metrics"].items()])
+        documents = "\n\n \t".join([cls._format_document_answer(doc) for doc in query.get("documents", [])])
+        documents = f"Documents: \n \t{documents}\n" if len(documents) > 0 else ""
+        answers = "\n\n \t".join([cls._format_document_answer(answer) for answer in query.get("answers", [])])
+        answers = f"Answers: \n \t{answers}\n" if len(answers) > 0 else ""
+        gold_document_ids = "\n \t".join(query["gold_document_ids"])
+        gold_answers = "\n \t".join(query.get("gold_answers", []))
+        gold_answers = f"Gold Answers: \n \t{gold_answers}\n" if len(gold_answers) > 0 else ""
+        s = (
+            f"Query: \n \t{query['query']}\n"
+            f"{gold_answers}"
+            f"Gold Document Ids: \n \t{gold_document_ids}\n"
+            f"Metrics: \n \t{metrics}\n"
+            f"{answers}"
+            f"{documents}"
+            f"_______________________________________________________"
+        )
+        return s
+
+    @classmethod
+    def _format_wrong_samples_node(cls, node_name: str, wrong_samples_formatted: str):
+        s = (
+            f"                Wrong {node_name} Examples\n"
+            f"=======================================================\n"
+            f"{wrong_samples_formatted}\n"
+            f"=======================================================\n"
+        )
+        return s
+
+    @classmethod
+    def _format_wrong_samples_report(cls, eval_result: EvaluationResult, n_wrong_examples: int = 3):
+        examples = {
+            node: eval_result.wrong_examples(node, doc_relevance_col="gold_id_or_answer_match", n=n_wrong_examples)
+            for node in eval_result.node_results.keys()
+        }
+        examples_formatted = {
+            node: "\n".join([cls._format_wrong_sample(example) for example in examples])
+            for node, examples in examples.items()
+        }
+
+        return "\n".join(
+            [cls._format_wrong_samples_node(node, examples) for node, examples in examples_formatted.items()]
+        )
+
+    @classmethod
+    def _format_pipeline_node(cls, node: str, calculated_metrics: dict):
+        node_metrics: dict = {}
+        for metric_mode in calculated_metrics:
+            for metric, value in calculated_metrics[metric_mode].get(node, {}).items():
+                node_metrics[f"{metric}{metric_mode}"] = value
+
+        node_metrics_formatted = "\n".join(
+            sorted([f"                        | {metric}: {value:5.3}" for metric, value in node_metrics.items()])
+        )
+        node_metrics_formatted = f"{node_metrics_formatted}\n" if len(node_metrics_formatted) > 0 else ""
+        s = (
+            f"                      {node}\n"
+            f"                        |\n"
+            f"{node_metrics_formatted}"
+            f"                        |"
+        )
+        return s
+
+    @classmethod
+    def _format_pipeline_overview(cls, calculated_metrics: dict, pipeline: Pipeline):
+        pipeline_overview = "\n".join(
+            [cls._format_pipeline_node(node, calculated_metrics) for node in pipeline.graph.nodes]
+        )
+        s = (
+            f"================== Evaluation Report ==================\n"
+            f"=======================================================\n"
+            f"                   Pipeline Overview\n"
+            f"=======================================================\n"
+            f"{pipeline_overview}\n"
+            f"                      Output\n"
+            f"=======================================================\n"
+        )
+        return s
diff --git a/test/test_pipeline.py b/test/test_pipeline.py
index 75b94c2f86..07c103410f 100644
--- a/test/test_pipeline.py
+++ b/test/test_pipeline.py
@@ -8,13 +8,11 @@
 from haystack.document_stores.deepsetcloud import DeepsetCloudDocumentStore
 
 from haystack.document_stores.elasticsearch import ElasticsearchDocumentStore
+from haystack.document_stores.memory import InMemoryDocumentStore
+from haystack.nodes.other.join_docs import JoinDocuments
 from haystack.nodes.retriever.sparse import ElasticsearchRetriever
-from haystack.pipelines import (
-    Pipeline,
-    DocumentSearchPipeline,
-    RootNode,
-)
-from haystack.pipelines import ExtractiveQAPipeline
+from haystack.pipelines import Pipeline, DocumentSearchPipeline, RootNode, ExtractiveQAPipeline
+from haystack.pipelines.base import _PipelineCodeGen
 from haystack.nodes import DensePassageRetriever, EmbeddingRetriever
 
 from conftest import MOCK_DC, DC_API_ENDPOINT, DC_API_KEY, DC_TEST_INDEX, SAMPLES_PATH, deepset_cloud_fixture
@@ -162,6 +160,142 @@ def test_load_tfidfretriever_yaml(tmp_path):
     assert prediction["answers"][0].answer == "haystack"
 
 
+@pytest.mark.elasticsearch
+def test_to_code():
+    index_pipeline = Pipeline.load_from_yaml(
+        SAMPLES_PATH / "pipeline" / "test_pipeline.yaml", pipeline_name="indexing_pipeline"
+    )
+    query_pipeline = Pipeline.load_from_yaml(
+        SAMPLES_PATH / "pipeline" / "test_pipeline.yaml", pipeline_name="query_pipeline"
+    )
+    query_pipeline_code = query_pipeline.to_code(pipeline_variable_name="query_pipeline_from_code")
+    index_pipeline_code = index_pipeline.to_code(pipeline_variable_name="index_pipeline_from_code")
+    exec(query_pipeline_code)
+    exec(index_pipeline_code)
+    assert locals()["query_pipeline_from_code"] is not None
+    assert locals()["index_pipeline_from_code"] is not None
+    assert query_pipeline.get_config() == locals()["query_pipeline_from_code"].get_config()
+    assert index_pipeline.get_config() == locals()["index_pipeline_from_code"].get_config()
+
+
+@pytest.mark.elasticsearch
+def test_PipelineCodeGen_simple_dense_pipeline():
+    doc_store = InMemoryDocumentStore(index="my-index")
+    retriever = EmbeddingRetriever(document_store=doc_store, embedding_model="sentence-transformers/all-MiniLM-L6-v2")
+    pipeline = Pipeline()
+    pipeline.add_node(component=retriever, name="retri", inputs=["Query"])
+
+    code = _PipelineCodeGen.generate_code(pipeline=pipeline, pipeline_variable_name="p", generate_imports=False)
+    assert code == (
+        'in_memory_document_store = InMemoryDocumentStore(index="my-index")\n'
+        'retri = EmbeddingRetriever(document_store=in_memory_document_store, embedding_model="sentence-transformers/all-MiniLM-L6-v2")\n'
+        "\n"
+        "p = Pipeline()\n"
+        'p.add_node(component=retri, name="retri", inputs=["Query"])'
+    )
+
+
+@pytest.mark.elasticsearch
+def test_PipelineCodeGen_simple_sparse_pipeline():
+    doc_store = ElasticsearchDocumentStore(index="my-index")
+    retriever = ElasticsearchRetriever(document_store=doc_store, top_k=20)
+    pipeline = Pipeline()
+    pipeline.add_node(component=retriever, name="retri", inputs=["Query"])
+
+    code = _PipelineCodeGen.generate_code(pipeline=pipeline, pipeline_variable_name="p", generate_imports=False)
+    assert code == (
+        'elasticsearch_document_store = ElasticsearchDocumentStore(index="my-index")\n'
+        "retri = ElasticsearchRetriever(document_store=elasticsearch_document_store, top_k=20)\n"
+        "\n"
+        "p = Pipeline()\n"
+        'p.add_node(component=retri, name="retri", inputs=["Query"])'
+    )
+
+
+@pytest.mark.elasticsearch
+def test_PipelineCodeGen_dual_retriever_pipeline():
+    es_doc_store = ElasticsearchDocumentStore(index="my-index")
+    es_retriever = ElasticsearchRetriever(document_store=es_doc_store, top_k=20)
+    dense_doc_store = InMemoryDocumentStore(index="my-index")
+    emb_retriever = EmbeddingRetriever(
+        document_store=dense_doc_store, embedding_model="sentence-transformers/all-MiniLM-L6-v2"
+    )
+    p_ensemble = Pipeline()
+    p_ensemble.add_node(component=es_retriever, name="EsRetriever", inputs=["Query"])
+    p_ensemble.add_node(component=emb_retriever, name="EmbeddingRetriever", inputs=["Query"])
+    p_ensemble.add_node(
+        component=JoinDocuments(join_mode="merge"), name="JoinResults", inputs=["EsRetriever", "EmbeddingRetriever"]
+    )
+
+    code = _PipelineCodeGen.generate_code(pipeline=p_ensemble, pipeline_variable_name="p", generate_imports=False)
+    assert code == (
+        'elasticsearch_document_store = ElasticsearchDocumentStore(index="my-index")\n'
+        "es_retriever = ElasticsearchRetriever(document_store=elasticsearch_document_store, top_k=20)\n"
+        'in_memory_document_store = InMemoryDocumentStore(index="my-index")\n'
+        'embedding_retriever = EmbeddingRetriever(document_store=in_memory_document_store, embedding_model="sentence-transformers/all-MiniLM-L6-v2")\n'
+        'join_results = JoinDocuments(join_mode="merge")\n'
+        "\n"
+        "p = Pipeline()\n"
+        'p.add_node(component=es_retriever, name="EsRetriever", inputs=["Query"])\n'
+        'p.add_node(component=embedding_retriever, name="EmbeddingRetriever", inputs=["Query"])\n'
+        'p.add_node(component=join_results, name="JoinResults", inputs=["EsRetriever", "EmbeddingRetriever"])'
+    )
+
+
+@pytest.mark.elasticsearch
+def test_PipelineCodeGen_dual_retriever_pipeline_same_docstore():
+    es_doc_store = ElasticsearchDocumentStore(index="my-index")
+    es_retriever = ElasticsearchRetriever(document_store=es_doc_store, top_k=20)
+    emb_retriever = EmbeddingRetriever(
+        document_store=es_retriever, embedding_model="sentence-transformers/all-MiniLM-L6-v2"
+    )
+    p_ensemble = Pipeline()
+    p_ensemble.add_node(component=es_retriever, name="EsRetriever", inputs=["Query"])
+    p_ensemble.add_node(component=emb_retriever, name="EmbeddingRetriever", inputs=["Query"])
+    p_ensemble.add_node(
+        component=JoinDocuments(join_mode="merge"), name="JoinResults", inputs=["EsRetriever", "EmbeddingRetriever"]
+    )
+
+    code = _PipelineCodeGen.generate_code(pipeline=p_ensemble, pipeline_variable_name="p", generate_imports=False)
+    assert code == (
+        'elasticsearch_document_store = ElasticsearchDocumentStore(index="my-index")\n'
+        "es_retriever = ElasticsearchRetriever(document_store=elasticsearch_document_store, top_k=20)\n"
+        'embedding_retriever = EmbeddingRetriever(document_store=elasticsearch_document_store, embedding_model="sentence-transformers/all-MiniLM-L6-v2")\n'
+        'join_results = JoinDocuments(join_mode="merge")\n'
+        "\n"
+        "p = Pipeline()\n"
+        'p.add_node(component=es_retriever, name="EsRetriever", inputs=["Query"])\n'
+        'p.add_node(component=embedding_retriever, name="EmbeddingRetriever", inputs=["Query"])\n'
+        'p.add_node(component=join_results, name="JoinResults", inputs=["EsRetriever", "EmbeddingRetriever"])'
+    )
+
+
+@pytest.mark.elasticsearch
+def test_PipelineCodeGen_imports():
+    doc_store = ElasticsearchDocumentStore(index="my-index")
+    retriever = ElasticsearchRetriever(document_store=doc_store, top_k=20)
+    pipeline = Pipeline()
+    pipeline.add_node(component=retriever, name="retri", inputs=["Query"])
+
+    code = _PipelineCodeGen.generate_code(pipeline=pipeline, pipeline_variable_name="p", generate_imports=True)
+    assert code == (
+        "from haystack.document_stores import ElasticsearchDocumentStore\n"
+        "from haystack.nodes import ElasticsearchRetriever\n"
+        "\n"
+        'elasticsearch_document_store = ElasticsearchDocumentStore(index="my-index")\n'
+        "retri = ElasticsearchRetriever(document_store=elasticsearch_document_store, top_k=20)\n"
+        "\n"
+        "p = Pipeline()\n"
+        'p.add_node(component=retri, name="retri", inputs=["Query"])'
+    )
+
+
+def test_PipelineCodeGen_order_components():
+    dependency_map = {"a": ["aa", "ab"], "aa": [], "ab": ["aba"], "aba": [], "b": ["a", "c"], "c": ["a"]}
+    ordered = _PipelineCodeGen._order_components(dependency_map=dependency_map)
+    assert ordered == ["aa", "aba", "ab", "a", "c", "b"]
+
+
 @pytest.mark.usefixtures(deepset_cloud_fixture.__name__)
 @responses.activate
 def test_load_from_deepset_cloud_query():


Code review comments:
diff_hunk: @@ -335,7 +389,7 @@ def save_to_deepset_cloud(
             logger.info(f"Pipeline config '{pipeline_config_name}' successfully created.")
 
     @classmethod
-    def _get_pipeline_definition(cls, pipeline_config: Dict, pipeline_name: Optional[str] = None):
+    def _get_pipeline_definition(cls, pipeline_config: Dict, pipeline_name: Optional[str] = None) -> dict:
path: haystack/pipelines/base.py
text: @user1:
Two minor details:
1. `dict` is discouraged as a type, the proper one is `Dict`
2. Better to use `Dict[str, Any]` that a bare `Dict`. That's what we decided in the handbook: https://www.notion.so/deepsetai/Developer-s-Handbook-WIP-cfb348225e814d25afd0a0665edac165#108c9e637187418c9bcbe2831108326f  (1.1.1 Type annotations, "What" section, point 2)

diff_hunk: @@ -357,7 +411,7 @@ def _get_pipeline_definition(cls, pipeline_config: Dict, pipeline_name: Optional
         return pipeline_definition
 
     @classmethod
-    def _get_component_definitions(cls, pipeline_config: Dict, overwrite_with_env_variables: bool):
+    def _get_component_definitions(cls, pipeline_config: Dict, overwrite_with_env_variables: bool) -> dict:
path: haystack/pipelines/base.py
text: @user1:
Same minor type issues on `dict` and `Dict`

diff_hunk: @@ -1446,3 +1395,257 @@ def __call__(self, *args, **kwargs):
         Ray calls this method which is then re-directed to the corresponding component's run().
         """
         return self.node._dispatch_run(*args, **kwargs)
+
+
+class _PipelineCodeGen:
+    @classmethod
+    def _camel_to_snake_case(cls, input: str) -> str:
+        return CAMEL_CASE_TO_SNAKE_CASE_REGEX.sub("_", input).lower()
+
+    @classmethod
+    def generate_code(
+        cls,
+        pipeline: BasePipeline,
+        pipeline_variable_name: str = "pipeline",
+        generate_imports: bool = True,
+        comment: Optional[str] = None,
+    ) -> str:
+        pipeline_config = pipeline.get_config()
+        component_definitions = pipeline._get_component_definitions(
+            pipeline_config=pipeline_config, overwrite_with_env_variables=False
+        )
+        component_variable_names = {name: cls._camel_to_snake_case(name) for name in component_definitions.keys()}
+        pipeline_definition = pipeline._get_pipeline_definition(pipeline_config=pipeline_config)
+
+        code_parts = []
+        if generate_imports:
+            types_to_import = [component["type"] for component in component_definitions.values()]
+            imports_code = cls._generate_imports_code(types_to_import=types_to_import)
+            code_parts.append(imports_code)
+
+        components_code = cls._generate_components_code(
+            component_definitions=component_definitions, component_variable_names=component_variable_names
+        )
+        pipeline_code = cls._generate_pipeline_code(
+            pipeline_definition=pipeline_definition,
+            component_variable_names=component_variable_names,
+            pipeline_variable_name=pipeline_variable_name,
+        )
+
+        code_parts.append(components_code)
+        code_parts.append(pipeline_code)
+        code = "\n\n".join(code_parts)
+
+        if comment:
+            comment = re.sub(r"^(#\s)?", "# ", comment, flags=re.MULTILINE)
+            code = "\n".join([comment, code])
+
+        return code
+
+    @classmethod
+    def _generate_pipeline_code(
+        cls, pipeline_definition: dict, component_variable_names: Dict[str, str], pipeline_variable_name: str
+    ) -> str:
+        code_lines = [f"{pipeline_variable_name} = Pipeline()"]
+        for node in pipeline_definition["nodes"]:
+            node_name = node["name"]
+            component_variable_name = component_variable_names[node_name]
+            inputs = ", ".join(f'"{name}"' for name in node["inputs"])
+            code_lines.append(
+                f'{pipeline_variable_name}.add_node(component={component_variable_name}, name="{node_name}", inputs=[{inputs}])'
+            )
+
+        code = "\n".join(code_lines)
+        return code
+
+    @classmethod
+    def _generate_components_code(cls, component_definitions: dict, component_variable_names: Dict[str, str]) -> str:
+        code = ""
+        declarations = {}
+        dependency_map = {}
+        for name, definition in component_definitions.items():
+            variable_name = component_variable_names[name]
+            class_name = definition["type"]
+            param_value_dict = {
+                key: component_variable_names.get(value, f'"{value}"') if type(value) == str else value
+                for key, value in definition["params"].items()
+            }
+            init_args = ", ".join(f"{key}={value}" for key, value in param_value_dict.items())
+            declarations[name] = f"{variable_name} = {class_name}({init_args})"
+            dependency_map[name] = [
+                param_value for param_value in definition["params"].values() if param_value in component_variable_names
path: haystack/pipelines/base.py
text: @user1:
This check can also create some bugs.

I admit this is also quite evil, but try out this pipeline:
```yaml
version: '0.9'

components: 
  - name: DocumentStore
    type: ElasticsearchDocumentStore
    params:
      host: localhost

  - name: localhost
    type: ElasticsearchRetriever
    params:
      document_store: DocumentStore

pipelines:
  - name: query
    type: Query
    nodes:
      - name: localhost
        inputs: [Query]
```

In this case is done on purpose, but in complex pipelines a thing like this could slip in. Also, I noticed that in general you rely on the user to ensure their pipeline is a DAG (loopless graph) and don't validate it. I think a check is in order.

@user1:
I just realized that `networkx` can probably do this check for you. Have a look at their docs before implementing the validation yourself.

@author:
I think we don't even validate for these thinks during pipeline execution. So I wouldn't invest too much into this.

@user1:
I quickly checked, it probably can: https://networkx.org/nx-guides/content/algorithms/dag/index.html#directed-acyclic-graph Should look like `nx.is_directed_acyclic_graph(clothing_graph)`

diff_hunk: @@ -1446,3 +1395,257 @@ def __call__(self, *args, **kwargs):
         Ray calls this method which is then re-directed to the corresponding component's run().
         """
         return self.node._dispatch_run(*args, **kwargs)
+
+
+class _PipelineCodeGen:
+    @classmethod
+    def _camel_to_snake_case(cls, input: str) -> str:
+        return CAMEL_CASE_TO_SNAKE_CASE_REGEX.sub("_", input).lower()
+
+    @classmethod
+    def generate_code(
+        cls,
+        pipeline: BasePipeline,
+        pipeline_variable_name: str = "pipeline",
+        generate_imports: bool = True,
+        comment: Optional[str] = None,
+    ) -> str:
+        pipeline_config = pipeline.get_config()
+        component_definitions = pipeline._get_component_definitions(
+            pipeline_config=pipeline_config, overwrite_with_env_variables=False
+        )
+        component_variable_names = {name: cls._camel_to_snake_case(name) for name in component_definitions.keys()}
+        pipeline_definition = pipeline._get_pipeline_definition(pipeline_config=pipeline_config)
+
+        code_parts = []
+        if generate_imports:
+            types_to_import = [component["type"] for component in component_definitions.values()]
+            imports_code = cls._generate_imports_code(types_to_import=types_to_import)
+            code_parts.append(imports_code)
+
+        components_code = cls._generate_components_code(
+            component_definitions=component_definitions, component_variable_names=component_variable_names
+        )
+        pipeline_code = cls._generate_pipeline_code(
+            pipeline_definition=pipeline_definition,
+            component_variable_names=component_variable_names,
+            pipeline_variable_name=pipeline_variable_name,
+        )
+
+        code_parts.append(components_code)
+        code_parts.append(pipeline_code)
+        code = "\n\n".join(code_parts)
+
+        if comment:
+            comment = re.sub(r"^(#\s)?", "# ", comment, flags=re.MULTILINE)
+            code = "\n".join([comment, code])
+
+        return code
+
+    @classmethod
+    def _generate_pipeline_code(
+        cls, pipeline_definition: dict, component_variable_names: Dict[str, str], pipeline_variable_name: str
+    ) -> str:
+        code_lines = [f"{pipeline_variable_name} = Pipeline()"]
+        for node in pipeline_definition["nodes"]:
+            node_name = node["name"]
+            component_variable_name = component_variable_names[node_name]
+            inputs = ", ".join(f'"{name}"' for name in node["inputs"])
+            code_lines.append(
+                f'{pipeline_variable_name}.add_node(component={component_variable_name}, name="{node_name}", inputs=[{inputs}])'
+            )
+
+        code = "\n".join(code_lines)
+        return code
+
+    @classmethod
+    def _generate_components_code(cls, component_definitions: dict, component_variable_names: Dict[str, str]) -> str:
+        code = ""
+        declarations = {}
+        dependency_map = {}
+        for name, definition in component_definitions.items():
+            variable_name = component_variable_names[name]
+            class_name = definition["type"]
+            param_value_dict = {
+                key: component_variable_names.get(value, f'"{value}"') if type(value) == str else value
+                for key, value in definition["params"].items()
+            }
+            init_args = ", ".join(f"{key}={value}" for key, value in param_value_dict.items())
+            declarations[name] = f"{variable_name} = {class_name}({init_args})"
+            dependency_map[name] = [
+                param_value for param_value in definition["params"].values() if param_value in component_variable_names
+            ]
+
+        ordered_components = cls._order_components(dependency_map=dependency_map)
+        ordered_declarations = [declarations[component] for component in ordered_components]
+        code = "\n".join(ordered_declarations)
+        return code
+
+    @classmethod
+    def _generate_imports_code(cls, types_to_import: List[str]) -> str:
+        code_lines = []
+        importable_classes = {
+            name: mod
+            for mod in CODE_GEN_ALLOWED_IMPORTS
+            for name, obj in inspect.getmembers(sys.modules[mod])
+            if inspect.isclass(obj)
+        }
path: haystack/pipelines/base.py
text: @user1:
This is fully independent from the function's input. I propose to extract it as a global constant, which will be calculated at import time, to avoid re-calculating it many times.

@author:
Would on the other hand delay app startup. But I think performance really isn't an issue here unless you execute this function hundreds of times per second.

diff_hunk: @@ -1446,3 +1395,257 @@ def __call__(self, *args, **kwargs):
         Ray calls this method which is then re-directed to the corresponding component's run().
         """
         return self.node._dispatch_run(*args, **kwargs)
+
+
+class _PipelineCodeGen:
+    @classmethod
+    def _camel_to_snake_case(cls, input: str) -> str:
+        return CAMEL_CASE_TO_SNAKE_CASE_REGEX.sub("_", input).lower()
+
+    @classmethod
+    def generate_code(
+        cls,
+        pipeline: BasePipeline,
+        pipeline_variable_name: str = "pipeline",
+        generate_imports: bool = True,
+        comment: Optional[str] = None,
+    ) -> str:
+        pipeline_config = pipeline.get_config()
+        component_definitions = pipeline._get_component_definitions(
+            pipeline_config=pipeline_config, overwrite_with_env_variables=False
+        )
+        component_variable_names = {name: cls._camel_to_snake_case(name) for name in component_definitions.keys()}
+        pipeline_definition = pipeline._get_pipeline_definition(pipeline_config=pipeline_config)
+
+        code_parts = []
+        if generate_imports:
+            types_to_import = [component["type"] for component in component_definitions.values()]
+            imports_code = cls._generate_imports_code(types_to_import=types_to_import)
+            code_parts.append(imports_code)
+
+        components_code = cls._generate_components_code(
+            component_definitions=component_definitions, component_variable_names=component_variable_names
+        )
+        pipeline_code = cls._generate_pipeline_code(
+            pipeline_definition=pipeline_definition,
+            component_variable_names=component_variable_names,
+            pipeline_variable_name=pipeline_variable_name,
+        )
+
+        code_parts.append(components_code)
+        code_parts.append(pipeline_code)
+        code = "\n\n".join(code_parts)
+
+        if comment:
+            comment = re.sub(r"^(#\s)?", "# ", comment, flags=re.MULTILINE)
+            code = "\n".join([comment, code])
+
+        return code
+
+    @classmethod
+    def _generate_pipeline_code(
+        cls, pipeline_definition: dict, component_variable_names: Dict[str, str], pipeline_variable_name: str
+    ) -> str:
+        code_lines = [f"{pipeline_variable_name} = Pipeline()"]
+        for node in pipeline_definition["nodes"]:
+            node_name = node["name"]
+            component_variable_name = component_variable_names[node_name]
+            inputs = ", ".join(f'"{name}"' for name in node["inputs"])
+            code_lines.append(
+                f'{pipeline_variable_name}.add_node(component={component_variable_name}, name="{node_name}", inputs=[{inputs}])'
+            )
+
+        code = "\n".join(code_lines)
+        return code
+
+    @classmethod
+    def _generate_components_code(cls, component_definitions: dict, component_variable_names: Dict[str, str]) -> str:
+        code = ""
+        declarations = {}
+        dependency_map = {}
+        for name, definition in component_definitions.items():
+            variable_name = component_variable_names[name]
+            class_name = definition["type"]
+            param_value_dict = {
+                key: component_variable_names.get(value, f'"{value}"') if type(value) == str else value
+                for key, value in definition["params"].items()
+            }
+            init_args = ", ".join(f"{key}={value}" for key, value in param_value_dict.items())
+            declarations[name] = f"{variable_name} = {class_name}({init_args})"
+            dependency_map[name] = [
+                param_value for param_value in definition["params"].values() if param_value in component_variable_names
+            ]
+
+        ordered_components = cls._order_components(dependency_map=dependency_map)
+        ordered_declarations = [declarations[component] for component in ordered_components]
+        code = "\n".join(ordered_declarations)
+        return code
+
+    @classmethod
+    def _generate_imports_code(cls, types_to_import: List[str]) -> str:
+        code_lines = []
+        importable_classes = {
+            name: mod
+            for mod in CODE_GEN_ALLOWED_IMPORTS
+            for name, obj in inspect.getmembers(sys.modules[mod])
+            if inspect.isclass(obj)
+        }
+
+        imports_by_module: Dict[str, List[str]] = {}
+        for t in types_to_import:
+            mod = importable_classes.get(t, MODULE_NOT_FOUND)
path: haystack/pipelines/base.py
text: @user1:
I would offer the option for an exception here, a "strict" mode that fails if the module is not found. Can help later to make sure the generated code is always valid, if needed.

@author:
Providing custom nodes would make this fails too. So basically we would exclude custom node pipelines here. In the current implementation the user could at least take care of the imports herself.

@user1:
Sorry I expressed myself badly. I don't think we should always fail, but make this function take a parameter, like `only_valid_code`, that if set to `True` would trigger an exception here. Not a big deal though :+1:

@author:
Yes I like that. But maybe not really urgent though.

diff_hunk: @@ -162,6 +160,142 @@ def test_load_tfidfretriever_yaml(tmp_path):
     assert prediction["answers"][0].answer == "haystack"
 
 
+@pytest.mark.elasticsearch
+def test_to_code():
+    index_pipeline = Pipeline.load_from_yaml(
+        SAMPLES_PATH / "pipeline" / "test_pipeline.yaml", pipeline_name="indexing_pipeline"
+    )
+    query_pipeline = Pipeline.load_from_yaml(
+        SAMPLES_PATH / "pipeline" / "test_pipeline.yaml", pipeline_name="query_pipeline"
+    )
+    query_pipeline_code = query_pipeline.to_code(pipeline_variable_name="query_pipeline_from_code")
+    index_pipeline_code = index_pipeline.to_code(pipeline_variable_name="index_pipeline_from_code")
+    exec(query_pipeline_code)
+    exec(index_pipeline_code)
+    assert locals()["query_pipeline_from_code"] is not None
+    assert locals()["index_pipeline_from_code"] is not None
+    assert query_pipeline.get_config() == locals()["query_pipeline_from_code"].get_config()
+    assert index_pipeline.get_config() == locals()["index_pipeline_from_code"].get_config()
+
+
+@pytest.mark.elasticsearch
+def test_PipelineCodeGen_simple_dense_pipeline():
+    doc_store = InMemoryDocumentStore(index="my-index")
+    retriever = EmbeddingRetriever(document_store=doc_store, embedding_model="sentence-transformers/all-MiniLM-L6-v2")
path: test/test_pipeline.py
text: @user1:
We could totally use mocks here. `EmbeddingRetriever` is not a lightweight guy!

diff_hunk: @@ -162,6 +160,142 @@ def test_load_tfidfretriever_yaml(tmp_path):
     assert prediction["answers"][0].answer == "haystack"
 
 
+@pytest.mark.elasticsearch
+def test_to_code():
+    index_pipeline = Pipeline.load_from_yaml(
+        SAMPLES_PATH / "pipeline" / "test_pipeline.yaml", pipeline_name="indexing_pipeline"
+    )
+    query_pipeline = Pipeline.load_from_yaml(
+        SAMPLES_PATH / "pipeline" / "test_pipeline.yaml", pipeline_name="query_pipeline"
+    )
+    query_pipeline_code = query_pipeline.to_code(pipeline_variable_name="query_pipeline_from_code")
+    index_pipeline_code = index_pipeline.to_code(pipeline_variable_name="index_pipeline_from_code")
+    exec(query_pipeline_code)
+    exec(index_pipeline_code)
+    assert locals()["query_pipeline_from_code"] is not None
+    assert locals()["index_pipeline_from_code"] is not None
+    assert query_pipeline.get_config() == locals()["query_pipeline_from_code"].get_config()
+    assert index_pipeline.get_config() == locals()["index_pipeline_from_code"].get_config()
+
+
+@pytest.mark.elasticsearch
path: test/test_pipeline.py
text: @user1:
Probably a leftover

diff_hunk: @@ -162,6 +160,142 @@ def test_load_tfidfretriever_yaml(tmp_path):
     assert prediction["answers"][0].answer == "haystack"
 
 
+@pytest.mark.elasticsearch
+def test_to_code():
+    index_pipeline = Pipeline.load_from_yaml(
+        SAMPLES_PATH / "pipeline" / "test_pipeline.yaml", pipeline_name="indexing_pipeline"
+    )
+    query_pipeline = Pipeline.load_from_yaml(
+        SAMPLES_PATH / "pipeline" / "test_pipeline.yaml", pipeline_name="query_pipeline"
+    )
+    query_pipeline_code = query_pipeline.to_code(pipeline_variable_name="query_pipeline_from_code")
+    index_pipeline_code = index_pipeline.to_code(pipeline_variable_name="index_pipeline_from_code")
+    exec(query_pipeline_code)
+    exec(index_pipeline_code)
+    assert locals()["query_pipeline_from_code"] is not None
+    assert locals()["index_pipeline_from_code"] is not None
+    assert query_pipeline.get_config() == locals()["query_pipeline_from_code"].get_config()
+    assert index_pipeline.get_config() == locals()["index_pipeline_from_code"].get_config()
+
+
+@pytest.mark.elasticsearch
+def test_PipelineCodeGen_simple_dense_pipeline():
+    doc_store = InMemoryDocumentStore(index="my-index")
+    retriever = EmbeddingRetriever(document_store=doc_store, embedding_model="sentence-transformers/all-MiniLM-L6-v2")
+    pipeline = Pipeline()
+    pipeline.add_node(component=retriever, name="retri", inputs=["Query"])
+
+    code = _PipelineCodeGen.generate_code(pipeline=pipeline, pipeline_variable_name="p", generate_imports=False)
+    assert code == (
+        'in_memory_document_store = InMemoryDocumentStore(index="my-index")\n'
+        'retri = EmbeddingRetriever(document_store=in_memory_document_store, embedding_model="sentence-transformers/all-MiniLM-L6-v2")\n'
+        "\n"
+        "p = Pipeline()\n"
+        'p.add_node(component=retri, name="retri", inputs=["Query"])'
+    )
path: test/test_pipeline.py
text: @user1:
Minor detail: I'd break the output on newlines and test the lines separately, so if tomorrow we decide to change the spacing the tests won't break for it.