Repository: deepset-ai/haystack

Issue:
Update `SuperComponent` to specify if Pipeline or AsyncPipeline should be used in `from_dict`
Currently in SuperComponent, the `from_dict` method does not specify whether the component should be loaded using Pipeline or AsyncPipeline. It just always uses `Pipeline.from_dict`. We should update the internal logic so `to_dict` specifies whether the component should be loaded using Pipeline or AsyncPipeline, and then `from_dict` should use that information.

This is especially problematic if loading a SuperComponent from yaml within an async pipeline. Since SuperComponent has a `run_async` method defined if it's placed in an `AsyncPipeline` at runtime `AsyncPipeline` will try and run the `run_async` method which will throw an error. This error is caused by `SuperComponent` complaining that it was given a normal `Pipeline` in it's init and not a `AsyncPipeline`. This means that it's not possible currently to use `SuperComponent` within an `AsyncPipeline` when loading from yaml. 

Related to this issue https://github.com/deepset-ai/haystack/issues/9435

Discussion hints:
coveralls: ## Pull Request Test Coverage Report for [Build 15709197968](https://coveralls.io/builds/74194905)


### Details

* **0** of **0**   changed or added relevant lines in **0** files are covered.
* **1** unchanged line in **1** file lost coverage.
* Overall coverage increased (+**0.002%**) to **90.145%**

---


|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [core/super_component/super_component.py](https://coveralls.io/builds/74194905/source?filename=core%2Fsuper_component%2Fsuper_component.py#L570) | 1 | 95.92% |
<!-- | **Total:** | **1** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/74194905/badge)](https://coveralls.io/builds/74194905) |
| :-- | --: |
| Change from base [Build 15706928119](https://coveralls.io/builds/74192897): |  0.002% |
| Covered Lines: | 11553 |
| Relevant Lines: | 12816 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Solution patch:
diff --git a/haystack/core/super_component/super_component.py b/haystack/core/super_component/super_component.py
index b13662e4..1e56bb3d 100644
--- a/haystack/core/super_component/super_component.py
+++ b/haystack/core/super_component/super_component.py
@@ -367,11 +367,13 @@ class _SuperComponent:
         :return: Dictionary containing serialized SuperComponent data
         """
         serialized_pipeline = self.pipeline.to_dict()
+        is_pipeline_async = isinstance(self.pipeline, AsyncPipeline)
         serialized = default_to_dict(
             self,
             pipeline=serialized_pipeline,
             input_mapping=self._original_input_mapping,
             output_mapping=self._original_output_mapping,
+            is_pipeline_async=is_pipeline_async,
         )
         serialized["type"] = generate_qualified_class_name(SuperComponent)
         return serialized
@@ -462,7 +464,9 @@ class SuperComponent(_SuperComponent):
         :returns:
             The deserialized SuperComponent.
         """
-        pipeline = Pipeline.from_dict(data["init_parameters"]["pipeline"])
+        is_pipeline_async = data["init_parameters"].pop("is_pipeline_async", False)
+        pipeline_class = AsyncPipeline if is_pipeline_async else Pipeline
+        pipeline = pipeline_class.from_dict(data["init_parameters"]["pipeline"])
         data["init_parameters"]["pipeline"] = pipeline
         return default_from_dict(cls, data)
 
diff --git a/releasenotes/notes/supercomponent-class-async-serde-4bfc357570b252db.yaml b/releasenotes/notes/supercomponent-class-async-serde-4bfc357570b252db.yaml
new file mode 100644
index 00000000..8fcfa0db
--- /dev/null
+++ b/releasenotes/notes/supercomponent-class-async-serde-4bfc357570b252db.yaml
@@ -0,0 +1,5 @@
+---
+fixes:
+  - |
+     The SuperComponent class can now correctly serialize and deserialize a SuperComponent based on an async pipeline.
+     Previously, the SuperComponent class always assumed the underlying pipeline was synchronous.


Test patch:
diff --git a/test/core/super_component/test_super_component.py b/test/core/super_component/test_super_component.py
index 5adba29d..60a91a5c 100644
--- a/test/core/super_component/test_super_component.py
+++ b/test/core/super_component/test_super_component.py
@@ -266,6 +266,7 @@ class TestSuperComponent:
         assert "type" in serialized
         assert "init_parameters" in serialized
         assert "pipeline" in serialized["init_parameters"]
+        assert serialized["init_parameters"]["is_pipeline_async"] is False
 
         # Test deserialization
         deserialized = SuperComponent.from_dict(serialized)
@@ -434,3 +435,33 @@ class TestSuperComponent:
         input_sockets = wrapper.__haystack_input__._sockets_dict
         assert "text" in input_sockets
         assert input_sockets["text"].type == str
+
+    @pytest.mark.asyncio
+    async def test_super_component_async_serialization_deserialization(self):
+        """
+        Test that when using the SuperComponent class, a SuperComponent based on an async pipeline can be serialized and
+        deserialized correctly.
+        """
+
+        @component
+        class AsyncComponent:
+            @component.output_types(output=str)
+            def run(self):
+                return {"output": "irrelevant"}
+
+            @component.output_types(output=str)
+            async def run_async(self):
+                return {"output": "Hello world"}
+
+        pipeline = AsyncPipeline()
+        pipeline.add_component("hello", AsyncComponent())
+
+        async_super_component = SuperComponent(pipeline=pipeline)
+        serialized_super_component = async_super_component.to_dict()
+        assert serialized_super_component["init_parameters"]["is_pipeline_async"] is True
+
+        deserialized_super_component = SuperComponent.from_dict(serialized_super_component)
+        assert isinstance(deserialized_super_component.pipeline, AsyncPipeline)
+
+        result = await deserialized_super_component.run_async()
+        assert result == {"output": "Hello world"}