Repository: deepset-ai/haystack
Pull request: feat: extend `pipeline.add_component` to support stores
### Related Issues

- fixes https://github.com/deepset-ai/haystack/issues/5238
- blocked by #5259 

### Proposed Changes:

- Changes the API of v2 Pipelines with respect to connecting docstores with components (see the issue for usage examples). The old API left all responsibility to the components, while the new API uses the pipeline as an intermediary for the connection.
- Creates two mixins to help the implementation of components requiring docstore(s): `StoreMixin` and `MultiStoreMixin`. 
- Adapt `MemoryRetriever` to this new connection style
- Fix all tests and add new ones

### How did you test it?
- local test run
- ci

### Notes for the reviewer

Let's focus on how the API feels like both on the pipeline side and on the component side.

### Checklist

- I have read the [contributors guidelines](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md) and the [code of conduct](https://github.com/deepset-ai/haystack/blob/main/code_of_conduct.txt)
- I have updated the related issue with new insights and changes
- I added unit tests and updated the docstrings
- I've used one of the [conventional commit types](https://www.conventionalcommits.org/en/v1.0.0/) for my PR title: `fix:`, `feat:`, `build:`, `chore:`, `ci:`, `docs:`, `style:`, `refactor:`, `perf:`, `test:`.
- I documented my code
- I ran [pre-commit hooks](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md#installation) and fixed any issue


Problem statement:
Change docstore connection in pipelines preview
Connecting components to their own document stores is right now responsibility of the component. Let's shift that on the pipeline instead.

To achieve it, let's change:

```python
search_pipe.add_component("retriever", MemoryRetriever(document_store_name="docstore"))
```

into:

```python
search_pipe.add_component("retriever", MemoryRetriever(), stores=["docstore"])
```

Hints from issue discussion:

Commit message: pylint

Patch to review:
diff --git a/haystack/preview/components/retrievers/memory.py b/haystack/preview/components/retrievers/memory.py
index 54d0ce8b5d..43c52d169c 100644
--- a/haystack/preview/components/retrievers/memory.py
+++ b/haystack/preview/components/retrievers/memory.py
@@ -9,6 +9,8 @@
 class MemoryRetriever:
     """
     A component for retrieving documents from a MemoryDocumentStore using the BM25 algorithm.
+
+    Needs to be connected to a MemoryDocumentStore to run.
     """
 
     @dataclass
@@ -27,7 +29,6 @@ class Input(ComponentInput):
         filters: Dict[str, Any]
         top_k: int
         scale_score: bool
-        stores: Dict[str, Any]
 
     @dataclass
     class Output(ComponentOutput):
@@ -39,28 +40,54 @@ class Output(ComponentOutput):
 
         documents: List[Document]
 
-    def __init__(
-        self,
-        document_store_name: str,
-        filters: Optional[Dict[str, Any]] = None,
-        top_k: int = 10,
-        scale_score: bool = True,
-    ):
+    def __init__(self, filters: Optional[Dict[str, Any]] = None, top_k: int = 10, scale_score: bool = True):
         """
         Create a MemoryRetriever component.
 
-        :param document_store_name: The name of the MemoryDocumentStore to retrieve documents from.
         :param filters: A dictionary with filters to narrow down the search space (default is None).
         :param top_k: The maximum number of documents to retrieve (default is 10).
         :param scale_score: Whether to scale the BM25 score or not (default is True).
 
         :raises ValueError: If the specified top_k is not > 0.
         """
-        self.document_store_name = document_store_name
         if top_k <= 0:
             raise ValueError(f"top_k must be > 0, but got {top_k}")
         self.defaults = {"top_k": top_k, "scale_score": scale_score, "filters": filters or {}}
 
+    @property
+    def store(self) -> Optional[MemoryDocumentStore]:
+        """
+        This property allows Pipelines to connect the component with the stores it requires.
+
+        Stores have to be instances of MemoryDocumentStore, or the assignment will fail.
+        """
+        return getattr(self, "_store", None)
+
+    @store.setter
+    def store(self, store: MemoryDocumentStore):
+        """
+        This property allows Pipelines to connect the component with the stores it requires.
+
+        Stores have to be instances of MemoryDocumentStore, or the assignment will fail.
+
+        :param store: the MemoryDocumentStore instance to retrieve from.
+        :raises ValueError if the store is not an instance of MemoryDocumentStore.
+        """
+        if not store:
+            raise ValueError("Can't set the value of the store to None.")
+        if not isinstance(store, MemoryDocumentStore):
+            raise ValueError("MemoryRetriever can only be used with a MemoryDocumentStore instance.")
+        self._store = store
+
+    def warmup(self):
+        """
+        Double-checks that a store is set before running this component in a pipeline.
+        """
+        if not self.store:
+            raise ValueError(
+                "MemoryRetriever needs a store to run: use the 'store' parameter of 'add_component' to connect them."
+            )
+
     def run(self, data: Input) -> Output:
         """
         Run the MemoryRetriever on the given input data.
@@ -70,15 +97,9 @@ def run(self, data: Input) -> Output:
 
         :raises ValueError: If the specified document store is not found or is not a MemoryDocumentStore instance.
         """
-        if self.document_store_name not in data.stores:
-            raise ValueError(
-                f"MemoryRetriever's document store '{self.document_store_name}' not found "
-                f"in input stores {list(data.stores.keys())}"
-            )
-        document_store = data.stores[self.document_store_name]
-        if not isinstance(document_store, MemoryDocumentStore):
-            raise ValueError("MemoryRetriever can only be used with a MemoryDocumentStore instance.")
-        docs = document_store.bm25_retrieval(
+        if not self.store:
+            raise ValueError("MemoryRetriever needs a store to run: set the store instance to the self.store attribute")
+        docs = self.store.bm25_retrieval(
             query=data.query, filters=data.filters, top_k=data.top_k, scale_score=data.scale_score
         )
         return MemoryRetriever.Output(documents=docs)
diff --git a/haystack/preview/document_stores/__init__.py b/haystack/preview/document_stores/__init__.py
index 6d70bca627..ba4980f0ad 100644
--- a/haystack/preview/document_stores/__init__.py
+++ b/haystack/preview/document_stores/__init__.py
@@ -1,2 +1,4 @@
+from haystack.preview.document_stores.protocols import Store, DuplicatePolicy
+from haystack.preview.document_stores.mixins import StoreMixin, MultiStoreMixin, StoreComponent, MultiStoreComponent
 from haystack.preview.document_stores.memory.document_store import MemoryDocumentStore
 from haystack.preview.document_stores.errors import StoreError, DuplicateDocumentError, MissingDocumentError
diff --git a/haystack/preview/document_stores/memory/document_store.py b/haystack/preview/document_stores/memory/document_store.py
index 4e6f47e90b..3d64e0d3bd 100644
--- a/haystack/preview/document_stores/memory/document_store.py
+++ b/haystack/preview/document_stores/memory/document_store.py
@@ -8,12 +8,12 @@
 from tqdm.auto import tqdm
 
 from haystack.preview.dataclasses import Document
+from haystack.preview.document_stores.protocols import DuplicatePolicy
 from haystack.preview.document_stores.memory._filters import match
 from haystack.preview.document_stores.errors import DuplicateDocumentError, MissingDocumentError
 from haystack.utils.scipy_utils import expit
 
 logger = logging.getLogger(__name__)
-DuplicatePolicy = Literal["skip", "overwrite", "fail"]
 
 # document scores are essentially unbounded and will be scaled to values between 0 and 1 if scale_score is set to
 # True (default). Scaling uses the expit function (inverse of the logit function) after applying a SCALING_FACTOR. A
@@ -126,17 +126,17 @@ def filter_documents(self, filters: Optional[Dict[str, Any]] = None) -> List[Doc
             return [doc for doc in self.storage.values() if match(conditions=filters, document=doc)]
         return list(self.storage.values())
 
-    def write_documents(self, documents: List[Document], duplicates: DuplicatePolicy = "fail") -> None:
+    def write_documents(self, documents: List[Document], policy: DuplicatePolicy = DuplicatePolicy.FAIL) -> None:
         """
         Writes (or overwrites) documents into the store.
 
         :param documents: a list of documents.
-        :param duplicates: documents with the same ID count as duplicates. When duplicates are met,
+        :param policy: documents with the same ID count as duplicates. When duplicates are met,
             the store can:
              - skip: keep the existing document and ignore the new one.
              - overwrite: remove the old document and write the new one.
              - fail: an error is raised
-        :raises DuplicateError: Exception trigger on duplicate document if `duplicates="fail"`
+        :raises DuplicateError: Exception trigger on duplicate document if `policy=DuplicatePolicy.FAIL`
         :return: None
         """
         if (
@@ -147,10 +147,10 @@ def write_documents(self, documents: List[Document], duplicates: DuplicatePolicy
             raise ValueError("Please provide a list of Documents.")
 
         for document in documents:
-            if document.id in self.storage.keys():
-                if duplicates == "fail":
+            if policy != DuplicatePolicy.OVERWRITE and document.id in self.storage.keys():
+                if policy == DuplicatePolicy.FAIL:
                     raise DuplicateDocumentError(f"ID '{document.id}' already exists.")
-                if duplicates == "skip":
+                if policy == DuplicatePolicy.SKIP:
                     logger.warning("ID '%s' already exists", document.id)
             self.storage[document.id] = document
 
diff --git a/haystack/preview/document_stores/mixins.py b/haystack/preview/document_stores/mixins.py
new file mode 100644
index 0000000000..4f2b0cc608
--- /dev/null
+++ b/haystack/preview/document_stores/mixins.py
@@ -0,0 +1,67 @@
+from typing import Protocol, Dict, Optional
+
+from haystack.preview.document_stores.protocols import Store
+
+
+class StoreComponent(Protocol):
+    """
+    Represents a component that needs a single store to run.
+    """
+
+    _store: Store
+
+    @property
+    def store(self) -> Optional[Store]:
+        ...
+
+    @store.setter
+    def store(self, store: Store):
+        ...
+
+
+class StoreMixin(StoreComponent):
+    """
+    Adds the capability of a component to use a single document store from the `self.store` property.
+    """
+
+    @property
+    def store(self) -> Optional[Store]:
+        return getattr(self, "_store", None)
+
+    @store.setter
+    def store(self, store: Store):
+        if not store:
+            raise ValueError("Can't set the value of the store to None.")
+        self._store = store
+
+
+class MultiStoreComponent:
+    """
+    Represents a component that needs more than a single store to run.
+    """
+
+    @property
+    def stores(self) -> Optional[Dict[str, Store]]:
+        ...
+
+    @stores.setter
+    def stores(self, stores: Dict[str, Store]):
+        ...
+
+
+class MultiStoreMixin(MultiStoreComponent):
+    """
+    Adds the capability of a component to use several document stores from the `self.stores` property.
+    """
+
+    _stores: Dict[str, Store]
+
+    @property
+    def stores(self) -> Optional[Dict[str, Store]]:
+        return getattr(self, "_stores", None)
+
+    @stores.setter
+    def stores(self, stores: Dict[str, Store]):
+        if stores is None:
+            raise ValueError("The stores dictionary can't be None.")
+        self._stores = stores
diff --git a/haystack/preview/document_stores/protocols.py b/haystack/preview/document_stores/protocols.py
new file mode 100644
index 0000000000..1c269351fb
--- /dev/null
+++ b/haystack/preview/document_stores/protocols.py
@@ -0,0 +1,126 @@
+from typing import Protocol, Optional, Dict, Any, List
+
+import logging
+from enum import Enum
+
+from haystack.preview.dataclasses import Document
+
+
+logger = logging.getLogger(__name__)
+
+
+class DuplicatePolicy(Enum):
+    SKIP = "skip"
+    OVERWRITE = "overwrite"
+    FAIL = "fail"
+
+
+class Store(Protocol):
+    """
+    Stores Documents to be used by the components of a Pipeline.
+
+    Classes implementing this protocol often store the documents permanently and allow specialized components to
+    perform retrieval on them, either by embedding, by keyword, hybrid, and so on, depending on the backend used.
+
+    In order to retrieve documents, consider using a Retriever that supports the document store implementation that
+    you're using.
+    """
+
+    def count_documents(self) -> int:
+        """
+        Returns the number of documents stored.
+        """
+
+    def filter_documents(self, filters: Optional[Dict[str, Any]] = None) -> List[Document]:
+        """
+        Returns the documents that match the filters provided.
+
+        Filters are defined as nested dictionaries. The keys of the dictionaries can be a logical operator (`"$and"`,
+        `"$or"`, `"$not"`), a comparison operator (`"$eq"`, `$ne`, `"$in"`, `$nin`, `"$gt"`, `"$gte"`, `"$lt"`,
+        `"$lte"`) or a metadata field name.
+
+        Logical operator keys take a dictionary of metadata field names and/or logical operators as value. Metadata
+        field names take a dictionary of comparison operators as value. Comparison operator keys take a single value or
+        (in case of `"$in"`) a list of values as value. If no logical operator is provided, `"$and"` is used as default
+        operation. If no comparison operator is provided, `"$eq"` (or `"$in"` if the comparison value is a list) is used
+        as default operation.
+
+        Example:
+
+        ```python
+        filters = {
+            "$and": {
+                "type": {"$eq": "article"},
+                "date": {"$gte": "2015-01-01", "$lt": "2021-01-01"},
+                "rating": {"$gte": 3},
+                "$or": {
+                    "genre": {"$in": ["economy", "politics"]},
+                    "publisher": {"$eq": "nytimes"}
+                }
+            }
+        }
+        # or simpler using default operators
+        filters = {
+            "type": "article",
+            "date": {"$gte": "2015-01-01", "$lt": "2021-01-01"},
+            "rating": {"$gte": 3},
+            "$or": {
+                "genre": ["economy", "politics"],
+                "publisher": "nytimes"
+            }
+        }
+        ```
+
+        To use the same logical operator multiple times on the same level, logical operators can take a list of
+        dictionaries as value.
+
+        Example:
+
+        ```python
+        filters = {
+            "$or": [
+                {
+                    "$and": {
+                        "Type": "News Paper",
+                        "Date": {
+                            "$lt": "2019-01-01"
+                        }
+                    }
+                },
+                {
+                    "$and": {
+                        "Type": "Blog Post",
+                        "Date": {
+                            "$gte": "2019-01-01"
+                        }
+                    }
+                }
+            ]
+        }
+        ```
+
+        :param filters: the filters to apply to the document list.
+        :return: a list of Documents that match the given filters.
+        """
+
+    def write_documents(self, documents: List[Document], policy: DuplicatePolicy = DuplicatePolicy.FAIL) -> None:
+        """
+        Writes (or overwrites) documents into the store.
+
+        :param documents: a list of documents.
+        :param policy: documents with the same ID count as duplicates. When duplicates are met,
+            the store can:
+             - skip: keep the existing document and ignore the new one.
+             - overwrite: remove the old document and write the new one.
+             - fail: an error is raised
+        :raises DuplicateError: Exception trigger on duplicate document if `policy=DuplicatePolicy.FAIL`
+        :return: None
+        """
+
+    def delete_documents(self, document_ids: List[str]) -> None:
+        """
+        Deletes all documents with a matching document_ids from the document store.
+        Fails with `MissingDocumentError` if no document with this id is present in the store.
+
+        :param object_ids: the object_ids to delete
+        """
diff --git a/haystack/preview/pipeline.py b/haystack/preview/pipeline.py
index 6a087cb7b3..746cd83742 100644
--- a/haystack/preview/pipeline.py
+++ b/haystack/preview/pipeline.py
@@ -2,14 +2,14 @@
 
 from pathlib import Path
 
-from canals.component import ComponentInput, ComponentOutput
 from canals.pipeline import (
     Pipeline as CanalsPipeline,
     PipelineError,
     load_pipelines as load_canals_pipelines,
     save_pipelines as save_canals_pipelines,
 )
-from canals.pipeline.sockets import find_input_sockets
+
+from haystack.preview.document_stores.protocols import Store
 
 
 class NoSuchStoreError(PipelineError):
@@ -23,9 +23,9 @@ class Pipeline(CanalsPipeline):
 
     def __init__(self):
         super().__init__()
-        self.stores = {}
+        self.stores: Dict[str, Store] = {}
 
-    def add_store(self, name: str, store: object) -> None:
+    def add_store(self, name: str, store: Store) -> None:
         """
         Make a store available to all nodes of this pipeline.
 
@@ -43,7 +43,7 @@ def list_stores(self) -> List[str]:
         """
         return list(self.stores.keys())
 
-    def get_store(self, name: str) -> object:
+    def get_store(self, name: str) -> Store:
         """
         Returns the store associated with the given name.
 
@@ -55,29 +55,41 @@ def get_store(self, name: str) -> object:
         except KeyError as e:
             raise NoSuchStoreError(f"No store named '{name}' is connected to this pipeline.") from e
 
-    def run(self, data: Dict[str, ComponentInput], debug: bool = False) -> Dict[str, ComponentOutput]:
+    def add_component(self, name: str, instance: Any, stores: Optional[List[str]] = None) -> None:
         """
-        Wrapper on top of Canals Pipeline.run(). Adds the `stores` parameter to all nodes.
+        Make this component available to the pipeline. Components are not connected to anything by default:
+        use `Pipeline.connect()` to connect components together.
+
+        Component names must be unique, but component instances can be reused if needed.
+
+        If `stores` has a value, the pipeline will also connect this component to the requested document store(s).
+        Note that only components that respect the StoreMixin or StoresMixin protocols can be connected to stores.
 
-        :params data: the inputs to give to the input components of the Pipeline.
-        :params parameters: a dictionary with all the parameters of all the components, namespaced by component.
-        :params debug: whether to collect and return debug information.
-        :returns A dictionary with the outputs of the output components of the Pipeline.
+        :param name: the name of the component.
+        :param instance: the component instance.
+        :param stores: the stores this component needs access to, if any.
+        :raises ValueError: if a component with the same name already exists
+        :raises PipelineValidationError: if the given instance is not a component
         """
-        # Get all nodes in this pipelines instance
-        for node_name in self.graph.nodes:
-            # Get node inputs
-            node = self.graph.nodes[node_name]["instance"]
-            input_params = find_input_sockets(node)
-
-            # If the node needs a store, adds the list of stores to its default inputs
-            if "stores" in input_params:
-                if not hasattr(node, "defaults"):
-                    setattr(node, "defaults", {})
-                node.defaults["stores"] = self.stores
-
-        # Run the pipeline
-        return super().run(data=data, debug=debug)
+        super().add_component(name, instance)
+
+        stores = stores or []
+        for store in stores:
+            if store not in self.stores:
+                raise NoSuchStoreError(
+                    f"Store named '{store}' not found. "
+                    f"Add it with 'pipeline.add_store('{store}', <the docstore instance>)'."
+                )
+
+        # This component implements the MultiStoreComponent protocol
+        if hasattr(instance, "stores"):
+            instance.stores = {store: self.stores[store] for store in stores}
+
+        # This component implements the StoreComponent protocol
+        elif hasattr(instance, "store"):
+            if len(stores) != 1:
+                raise ValueError(f"Component '{name}' needs exactly one store.")
+            instance.store = self.stores[stores[0]]
 
 
 def load_pipelines(path: Path, _reader: Optional[Callable[..., Any]] = None):
diff --git a/test/preview/components/retrievers/test_memory_retriever.py b/test/preview/components/retrievers/test_memory_retriever.py
index 33963a6e58..5ab4b388b8 100644
--- a/test/preview/components/retrievers/test_memory_retriever.py
+++ b/test/preview/components/retrievers/test_memory_retriever.py
@@ -1,14 +1,16 @@
-from typing import Dict, Any, List
+from typing import Dict, Any, List, Optional
 
 import pytest
 
 from haystack.preview import Pipeline
 from haystack.preview.components.retrievers.memory import MemoryRetriever
 from haystack.preview.dataclasses import Document
-from haystack.preview.document_stores import MemoryDocumentStore
+from haystack.preview.document_stores import Store, MemoryDocumentStore
 
 from test.preview.components.base import BaseTestComponent
 
+from haystack.preview.document_stores.protocols import DuplicatePolicy
+
 
 @pytest.fixture()
 def mock_docs():
@@ -24,64 +26,59 @@ def mock_docs():
 class Test_MemoryRetriever(BaseTestComponent):
     @pytest.mark.unit
     def test_save_load(self, tmp_path):
-        self.assert_can_be_saved_and_loaded_in_pipeline(MemoryRetriever(document_store_name="memory"), tmp_path)
+        self.assert_can_be_saved_and_loaded_in_pipeline(MemoryRetriever(), tmp_path)
 
     @pytest.mark.unit
     def test_save_load_with_parameters(self, tmp_path):
-        self.assert_can_be_saved_and_loaded_in_pipeline(
-            MemoryRetriever(document_store_name="memory", top_k=5, scale_score=False), tmp_path
-        )
+        self.assert_can_be_saved_and_loaded_in_pipeline(MemoryRetriever(top_k=5, scale_score=False), tmp_path)
 
     @pytest.mark.unit
     def test_init_default(self):
-        retriever = MemoryRetriever(document_store_name="memory")
-        assert retriever.document_store_name == "memory"
+        retriever = MemoryRetriever()
         assert retriever.defaults == {"filters": {}, "top_k": 10, "scale_score": True}
 
     @pytest.mark.unit
     def test_init_with_parameters(self):
-        retriever = MemoryRetriever(document_store_name="memory-test", top_k=5, scale_score=False)
-        assert retriever.document_store_name == "memory-test"
+        retriever = MemoryRetriever(top_k=5, scale_score=False)
         assert retriever.defaults == {"filters": {}, "top_k": 5, "scale_score": False}
 
     @pytest.mark.unit
     def test_init_with_invalid_top_k_parameter(self):
         with pytest.raises(ValueError, match="top_k must be > 0, but got -2"):
-            MemoryRetriever(document_store_name="memory-test", top_k=-2, scale_score=False)
+            MemoryRetriever(top_k=-2, scale_score=False)
 
     @pytest.mark.unit
     def test_valid_run(self, mock_docs):
         top_k = 5
         ds = MemoryDocumentStore()
         ds.write_documents(mock_docs)
-        mr = MemoryRetriever(document_store_name="memory", top_k=top_k)
-        result: MemoryRetriever.Output = mr.run(data=MemoryRetriever.Input(query="PHP", stores={"memory": ds}))
+
+        mr = MemoryRetriever(top_k=top_k)
+        mr.store = ds
+        result = mr.run(data=MemoryRetriever.Input(query="PHP"))
 
         assert getattr(result, "documents")
         assert len(result.documents) == top_k
         assert result.documents[0].content == "PHP is a popular programming language"
 
-    @pytest.mark.unit
-    def test_invalid_run_wrong_store_name(self):
-        # Test invalid run with wrong store name
-        ds = MemoryDocumentStore()
-        mr = MemoryRetriever(document_store_name="memory")
-        with pytest.raises(ValueError, match=r"MemoryRetriever's document store 'memory' not found"):
-            invalid_input_data = MemoryRetriever.Input(
-                query="test", top_k=10, scale_score=True, stores={"invalid_store": ds}
-            )
-            mr.run(invalid_input_data)
-
     @pytest.mark.unit
     def test_invalid_run_wrong_store_type(self):
-        # Test invalid run with wrong store type
-        ds = MemoryDocumentStore()
-        mr = MemoryRetriever(document_store_name="memory")
-        with pytest.raises(ValueError, match=r"MemoryRetriever can only be used with a MemoryDocumentStore instance."):
-            invalid_input_data = MemoryRetriever.Input(
-                query="test", top_k=10, scale_score=True, stores={"memory": "not a MemoryDocumentStore"}
-            )
-            mr.run(invalid_input_data)
+        class MockStore(Store):
+            def count_documents(self) -> int:
+                return 0
+
+            def filter_documents(self, filters: Optional[Dict[str, Any]] = None) -> List[Document]:
+                return []
+
+            def write_documents(self, documents: List[Document], duplicates: DuplicatePolicy = "fail") -> None:
+                return
+
+            def delete_documents(self, document_ids: List[str]) -> None:
+                return
+
+        mr = MemoryRetriever()
+        with pytest.raises(ValueError, match="MemoryRetriever can only be used with a MemoryDocumentStore instance"):
+            mr.store = MockStore()
 
     @pytest.mark.integration
     @pytest.mark.parametrize(
@@ -94,11 +91,11 @@ def test_invalid_run_wrong_store_type(self):
     def test_run_with_pipeline(self, mock_docs, query: str, query_result: str):
         ds = MemoryDocumentStore()
         ds.write_documents(mock_docs)
-        mr = MemoryRetriever(document_store_name="memory")
+        mr = MemoryRetriever()
 
         pipeline = Pipeline()
-        pipeline.add_component("retriever", mr)
         pipeline.add_store("memory", ds)
+        pipeline.add_component("retriever", mr, stores=["memory"])
         result: Dict[str, Any] = pipeline.run(data={"retriever": MemoryRetriever.Input(query=query)})
 
         assert result
@@ -119,11 +116,11 @@ def test_run_with_pipeline(self, mock_docs, query: str, query_result: str):
     def test_run_with_pipeline_and_top_k(self, mock_docs, query: str, query_result: str, top_k: int):
         ds = MemoryDocumentStore()
         ds.write_documents(mock_docs)
-        mr = MemoryRetriever(document_store_name="memory")
+        mr = MemoryRetriever()
 
         pipeline = Pipeline()
-        pipeline.add_component("retriever", mr)
         pipeline.add_store("memory", ds)
+        pipeline.add_component("retriever", mr, stores=["memory"])
         result: Dict[str, Any] = pipeline.run(data={"retriever": MemoryRetriever.Input(query=query, top_k=top_k)})
 
         assert result
diff --git a/test/preview/document_stores/_base.py b/test/preview/document_stores/_base.py
index b60daaa391..3f9497d65b 100644
--- a/test/preview/document_stores/_base.py
+++ b/test/preview/document_stores/_base.py
@@ -5,17 +5,17 @@
 import pandas as pd
 
 from haystack.preview.dataclasses import Document
-from haystack.preview.document_stores import StoreError
+from haystack.preview.document_stores import Store, StoreError, DuplicatePolicy
 from haystack.preview.document_stores import MissingDocumentError, DuplicateDocumentError
 
 
 class DocumentStoreBaseTests:
     @pytest.fixture
-    def docstore(self):
+    def docstore(self) -> Store:
         raise NotImplementedError()
 
     @pytest.fixture
-    def filterable_docs(self):
+    def filterable_docs(self) -> List[Document]:
         embedding_zero = np.zeros([768, 1]).astype(np.float32)
         embedding_one = np.ones([768, 1]).astype(np.float32)
 
@@ -70,42 +70,42 @@ def contains_same_docs(self, first_list: List[Document], second_list: List[Docum
         )
 
     @pytest.mark.unit
-    def test_count_empty(self, docstore):
+    def test_count_empty(self, docstore: Store):
         assert docstore.count_documents() == 0
 
     @pytest.mark.unit
-    def test_count_not_empty(self, docstore):
+    def test_count_not_empty(self, docstore: Store):
         docstore.write_documents(
             [Document(content="test doc 1"), Document(content="test doc 2"), Document(content="test doc 3")]
         )
         assert docstore.count_documents() == 3
 
     @pytest.mark.unit
-    def test_no_filter_empty(self, docstore):
+    def test_no_filter_empty(self, docstore: Store):
         assert docstore.filter_documents() == []
         assert docstore.filter_documents(filters={}) == []
 
     @pytest.mark.unit
-    def test_no_filter_not_empty(self, docstore):
+    def test_no_filter_not_empty(self, docstore: Store):
         docs = [Document(content="test doc")]
         docstore.write_documents(docs)
         assert docstore.filter_documents() == docs
         assert docstore.filter_documents(filters={}) == docs
 
     @pytest.mark.unit
-    def test_filter_simple_metadata_value(self, docstore, filterable_docs):
+    def test_filter_simple_metadata_value(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"page": "100"})
         assert self.contains_same_docs(result, [doc for doc in filterable_docs if doc.metadata.get("page") == "100"])
 
     @pytest.mark.unit
-    def test_filter_simple_list_single_element(self, docstore, filterable_docs):
+    def test_filter_simple_list_single_element(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"page": ["100"]})
         assert self.contains_same_docs(result, [doc for doc in filterable_docs if doc.metadata.get("page") == "100"])
 
     @pytest.mark.unit
-    def test_filter_document_content(self, docstore, filterable_docs):
+    def test_filter_document_content(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"content": "A Foo Document 1"})
         assert self.contains_same_docs(
@@ -113,19 +113,19 @@ def test_filter_document_content(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_filter_document_type(self, docstore, filterable_docs):
+    def test_filter_document_type(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"content_type": "table"})
         assert self.contains_same_docs(result, [doc for doc in filterable_docs if doc.content_type == "table"])
 
     @pytest.mark.unit
-    def test_filter_simple_list_one_value(self, docstore, filterable_docs):
+    def test_filter_simple_list_one_value(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"page": ["100"]})
         assert self.contains_same_docs(result, [doc for doc in filterable_docs if doc.metadata.get("page") in ["100"]])
 
     @pytest.mark.unit
-    def test_filter_simple_list(self, docstore, filterable_docs):
+    def test_filter_simple_list(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"page": ["100", "123"]})
         assert self.contains_same_docs(
@@ -133,49 +133,49 @@ def test_filter_simple_list(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_incorrect_filter_name(self, docstore, filterable_docs):
+    def test_incorrect_filter_name(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"non_existing_meta_field": ["whatever"]})
         assert len(result) == 0
 
     @pytest.mark.unit
-    def test_incorrect_filter_type(self, docstore, filterable_docs):
+    def test_incorrect_filter_type(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         with pytest.raises(ValueError, match="dictionaries or lists"):
             docstore.filter_documents(filters="something odd")
 
     @pytest.mark.unit
-    def test_incorrect_filter_value(self, docstore, filterable_docs):
+    def test_incorrect_filter_value(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"page": ["nope"]})
         assert len(result) == 0
 
     @pytest.mark.unit
-    def test_incorrect_filter_nesting(self, docstore, filterable_docs):
+    def test_incorrect_filter_nesting(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         with pytest.raises(ValueError, match="malformed"):
             docstore.filter_documents(filters={"number": {"page": "100"}})
 
     @pytest.mark.unit
-    def test_deeper_incorrect_filter_nesting(self, docstore, filterable_docs):
+    def test_deeper_incorrect_filter_nesting(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         with pytest.raises(ValueError, match="malformed"):
             docstore.filter_documents(filters={"number": {"page": {"chapter": "intro"}}})
 
     @pytest.mark.unit
-    def test_eq_filter_explicit(self, docstore, filterable_docs):
+    def test_eq_filter_explicit(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"page": {"$eq": "100"}})
         assert self.contains_same_docs(result, [doc for doc in filterable_docs if doc.metadata.get("page") == "100"])
 
     @pytest.mark.unit
-    def test_eq_filter_implicit(self, docstore, filterable_docs):
+    def test_eq_filter_implicit(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"page": "100"})
         assert self.contains_same_docs(result, [doc for doc in filterable_docs if doc.metadata.get("page") == "100"])
 
     @pytest.mark.unit
-    def test_eq_filter_table(self, docstore, filterable_docs):
+    def test_eq_filter_table(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"content": pd.DataFrame([1])})
         assert self.contains_same_docs(
@@ -188,7 +188,7 @@ def test_eq_filter_table(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_eq_filter_tensor(self, docstore, filterable_docs):
+    def test_eq_filter_tensor(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         embedding = np.zeros([768, 1]).astype(np.float32)
         result = docstore.filter_documents(filters={"embedding": embedding})
@@ -197,25 +197,25 @@ def test_eq_filter_tensor(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_deeper_incorrect_filter_nesting(self, docstore, filterable_docs):
+    def test_deeper_incorrect_filter_nesting(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         with pytest.raises(ValueError, match="malformed"):
             docstore.filter_documents(filters={"number": {"page": {"chapter": "intro"}}})
 
     @pytest.mark.unit
-    def test_eq_filter_explicit(self, docstore, filterable_docs):
+    def test_eq_filter_explicit(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"page": {"$eq": "100"}})
         assert self.contains_same_docs(result, [doc for doc in filterable_docs if doc.metadata.get("page") == "100"])
 
     @pytest.mark.unit
-    def test_eq_filter_implicit(self, docstore, filterable_docs):
+    def test_eq_filter_implicit(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"page": "100"})
         assert self.contains_same_docs(result, [doc for doc in filterable_docs if doc.metadata.get("page") == "100"])
 
     @pytest.mark.unit
-    def test_in_filter_explicit(self, docstore, filterable_docs):
+    def test_in_filter_explicit(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"page": {"$in": ["100", "123", "n.a."]}})
         assert self.contains_same_docs(
@@ -223,7 +223,7 @@ def test_in_filter_explicit(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_in_filter_implicit(self, docstore, filterable_docs):
+    def test_in_filter_implicit(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"page": ["100", "123", "n.a."]})
         assert self.contains_same_docs(
@@ -231,7 +231,7 @@ def test_in_filter_implicit(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_in_filter_table(self, docstore, filterable_docs):
+    def test_in_filter_table(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"content": {"$in": [pd.DataFrame([1]), pd.DataFrame([2])]}})
         assert self.contains_same_docs(
@@ -245,7 +245,7 @@ def test_in_filter_table(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_in_filter_tensor(self, docstore, filterable_docs):
+    def test_in_filter_tensor(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         embedding_zero = np.zeros([768, 1]).astype(np.float32)
         embedding_one = np.ones([768, 1]).astype(np.float32)
@@ -261,13 +261,13 @@ def test_in_filter_tensor(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_ne_filter(self, docstore, filterable_docs):
+    def test_ne_filter(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"page": {"$ne": "100"}})
         assert self.contains_same_docs(result, [doc for doc in filterable_docs if doc.metadata.get("page") != "100"])
 
     @pytest.mark.unit
-    def test_ne_filter_table(self, docstore, filterable_docs):
+    def test_ne_filter_table(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"content": {"$ne": pd.DataFrame([1])}})
         assert self.contains_same_docs(
@@ -280,7 +280,7 @@ def test_ne_filter_table(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_ne_filter_tensor(self, docstore, filterable_docs):
+    def test_ne_filter_tensor(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         embedding = np.zeros([768, 1]).astype(np.float32)
         result = docstore.filter_documents(filters={"embedding": {"$ne": embedding}})
@@ -294,7 +294,7 @@ def test_ne_filter_tensor(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_nin_filter(self, docstore, filterable_docs):
+    def test_nin_filter(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"page": {"$nin": ["100", "123", "n.a."]}})
         assert self.contains_same_docs(
@@ -302,7 +302,7 @@ def test_nin_filter(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_nin_filter_table(self, docstore, filterable_docs):
+    def test_nin_filter_table(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"content": {"$nin": [pd.DataFrame([1]), pd.DataFrame([0])]}})
         assert self.contains_same_docs(
@@ -316,7 +316,7 @@ def test_nin_filter_table(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_nin_filter_tensor(self, docstore, filterable_docs):
+    def test_nin_filter_tensor(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         embedding_zeros = np.zeros([768, 1]).astype(np.float32)
         embedding_ones = np.zeros([768, 1]).astype(np.float32)
@@ -335,7 +335,7 @@ def test_nin_filter_tensor(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_nin_filter(self, docstore, filterable_docs):
+    def test_nin_filter(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"page": {"$nin": ["100", "123", "n.a."]}})
         assert self.contains_same_docs(
@@ -343,7 +343,7 @@ def test_nin_filter(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_gt_filter(self, docstore, filterable_docs):
+    def test_gt_filter(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"number": {"$gt": 0.0}})
         assert self.contains_same_docs(
@@ -351,26 +351,26 @@ def test_gt_filter(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_gt_filter_non_numeric(self, docstore, filterable_docs):
+    def test_gt_filter_non_numeric(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         with pytest.raises(StoreError, match="Can't evaluate"):
             docstore.filter_documents(filters={"page": {"$gt": "100"}})
 
     @pytest.mark.unit
-    def test_gt_filter_table(self, docstore, filterable_docs):
+    def test_gt_filter_table(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         with pytest.raises(StoreError, match="Can't evaluate"):
             docstore.filter_documents(filters={"content": {"$gt": pd.DataFrame([[1, 2, 3], [-1, -2, -3]])}})
 
     @pytest.mark.unit
-    def test_gt_filter_tensor(self, docstore, filterable_docs):
+    def test_gt_filter_tensor(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         embedding_zeros = np.zeros([768, 1]).astype(np.float32)
         with pytest.raises(StoreError, match="Can't evaluate"):
             docstore.filter_documents(filters={"embedding": {"$gt": embedding_zeros}})
 
     @pytest.mark.unit
-    def test_gte_filter(self, docstore, filterable_docs):
+    def test_gte_filter(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"number": {"$gte": -2.0}})
         assert self.contains_same_docs(
@@ -378,26 +378,26 @@ def test_gte_filter(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_gte_filter_non_numeric(self, docstore, filterable_docs):
+    def test_gte_filter_non_numeric(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         with pytest.raises(StoreError, match="Can't evaluate"):
             docstore.filter_documents(filters={"page": {"$gte": "100"}})
 
     @pytest.mark.unit
-    def test_gte_filter_table(self, docstore, filterable_docs):
+    def test_gte_filter_table(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         with pytest.raises(StoreError, match="Can't evaluate"):
             docstore.filter_documents(filters={"content": {"$gte": pd.DataFrame([[1, 2, 3], [-1, -2, -3]])}})
 
     @pytest.mark.unit
-    def test_gte_filter_tensor(self, docstore, filterable_docs):
+    def test_gte_filter_tensor(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         embedding_zeros = np.zeros([768, 1]).astype(np.float32)
         with pytest.raises(StoreError, match="Can't evaluate"):
             docstore.filter_documents(filters={"embedding": {"$gte": embedding_zeros}})
 
     @pytest.mark.unit
-    def test_lt_filter(self, docstore, filterable_docs):
+    def test_lt_filter(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"number": {"$lt": 0.0}})
         assert self.contains_same_docs(
@@ -405,26 +405,26 @@ def test_lt_filter(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_lt_filter_non_numeric(self, docstore, filterable_docs):
+    def test_lt_filter_non_numeric(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         with pytest.raises(StoreError, match="Can't evaluate"):
             docstore.filter_documents(filters={"page": {"$lt": "100"}})
 
     @pytest.mark.unit
-    def test_lt_filter_table(self, docstore, filterable_docs):
+    def test_lt_filter_table(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         with pytest.raises(StoreError, match="Can't evaluate"):
             docstore.filter_documents(filters={"content": {"$lt": pd.DataFrame([[1, 2, 3], [-1, -2, -3]])}})
 
     @pytest.mark.unit
-    def test_lt_filter_tensor(self, docstore, filterable_docs):
+    def test_lt_filter_tensor(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         embedding_ones = np.ones([768, 1]).astype(np.float32)
         with pytest.raises(StoreError, match="Can't evaluate"):
             docstore.filter_documents(filters={"embedding": {"$lt": embedding_ones}})
 
     @pytest.mark.unit
-    def test_lte_filter(self, docstore, filterable_docs):
+    def test_lte_filter(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"number": {"$lte": 2.0}})
         assert self.contains_same_docs(
@@ -432,26 +432,26 @@ def test_lte_filter(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_lte_filter_non_numeric(self, docstore, filterable_docs):
+    def test_lte_filter_non_numeric(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         with pytest.raises(StoreError, match="Can't evaluate"):
             docstore.filter_documents(filters={"page": {"$lte": "100"}})
 
     @pytest.mark.unit
-    def test_lte_filter_table(self, docstore, filterable_docs):
+    def test_lte_filter_table(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         with pytest.raises(StoreError, match="Can't evaluate"):
             docstore.filter_documents(filters={"content": {"$lte": pd.DataFrame([[1, 2, 3], [-1, -2, -3]])}})
 
     @pytest.mark.unit
-    def test_lte_filter_tensor(self, docstore, filterable_docs):
+    def test_lte_filter_tensor(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         embedding_ones = np.ones([768, 1]).astype(np.float32)
         with pytest.raises(StoreError, match="Can't evaluate"):
             docstore.filter_documents(filters={"embedding": {"$lte": embedding_ones}})
 
     @pytest.mark.unit
-    def test_filter_simple_implicit_and_with_multi_key_dict(self, docstore, filterable_docs):
+    def test_filter_simple_implicit_and_with_multi_key_dict(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"number": {"$lte": 2.0, "$gte": 0.0}})
         assert self.contains_same_docs(
@@ -464,7 +464,7 @@ def test_filter_simple_implicit_and_with_multi_key_dict(self, docstore, filterab
         )
 
     @pytest.mark.unit
-    def test_filter_simple_explicit_and_with_multikey_dict(self, docstore, filterable_docs):
+    def test_filter_simple_explicit_and_with_multikey_dict(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"number": {"$and": {"$lte": 0, "$gte": -2}}})
         assert self.contains_same_docs(
@@ -477,7 +477,7 @@ def test_filter_simple_explicit_and_with_multikey_dict(self, docstore, filterabl
         )
 
     @pytest.mark.unit
-    def test_filter_simple_explicit_and_with_list(self, docstore, filterable_docs):
+    def test_filter_simple_explicit_and_with_list(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"number": {"$and": [{"$lte": 2}, {"$gte": 0}]}})
         assert self.contains_same_docs(
@@ -490,7 +490,7 @@ def test_filter_simple_explicit_and_with_list(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_filter_simple_implicit_and(self, docstore, filterable_docs):
+    def test_filter_simple_implicit_and(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         result = docstore.filter_documents(filters={"number": {"$lte": 2.0, "$gte": 0}})
         assert self.contains_same_docs(
@@ -503,7 +503,7 @@ def test_filter_simple_implicit_and(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_filter_nested_explicit_and(self, docstore, filterable_docs):
+    def test_filter_nested_explicit_and(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         filters = {"$and": {"number": {"$and": {"$lte": 2, "$gte": 0}}, "name": {"$in": ["name_0", "name_1"]}}}
         result = docstore.filter_documents(filters=filters)
@@ -522,7 +522,7 @@ def test_filter_nested_explicit_and(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_filter_nested_implicit_and(self, docstore, filterable_docs):
+    def test_filter_nested_implicit_and(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         filters_simplified = {"number": {"$lte": 2, "$gte": 0}, "name": ["name_0", "name_1"]}
         result = docstore.filter_documents(filters=filters_simplified)
@@ -541,7 +541,7 @@ def test_filter_nested_implicit_and(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_filter_simple_or(self, docstore, filterable_docs):
+    def test_filter_simple_or(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         filters = {"$or": {"name": {"$in": ["name_0", "name_1"]}, "number": {"$lt": 1.0}}}
         result = docstore.filter_documents(filters=filters)
@@ -558,7 +558,7 @@ def test_filter_simple_or(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_filter_nested_or(self, docstore, filterable_docs):
+    def test_filter_nested_or(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         filters = {"$or": {"name": {"$or": [{"$eq": "name_0"}, {"$eq": "name_1"}]}, "number": {"$lt": 1.0}}}
         result = docstore.filter_documents(filters=filters)
@@ -575,7 +575,7 @@ def test_filter_nested_or(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_filter_nested_and_or_explicit(self, docstore, filterable_docs):
+    def test_filter_nested_and_or_explicit(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         filters_simplified = {
             "$and": {"page": {"$eq": "123"}, "$or": {"name": {"$in": ["name_0", "name_1"]}, "number": {"$lt": 1.0}}}
@@ -597,7 +597,7 @@ def test_filter_nested_and_or_explicit(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_filter_nested_and_or_implicit(self, docstore, filterable_docs):
+    def test_filter_nested_and_or_implicit(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         filters_simplified = {
             "page": {"$eq": "123"},
@@ -620,7 +620,7 @@ def test_filter_nested_and_or_implicit(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_filter_nested_or_and(self, docstore, filterable_docs):
+    def test_filter_nested_or_and(self, docstore: Store, filterable_docs: List[Document]):
         docstore.write_documents(filterable_docs)
         filters_simplified = {
             "$or": {
@@ -645,7 +645,9 @@ def test_filter_nested_or_and(self, docstore, filterable_docs):
         )
 
     @pytest.mark.unit
-    def test_filter_nested_multiple_identical_operators_same_level(self, docstore, filterable_docs):
+    def test_filter_nested_multiple_identical_operators_same_level(
+        self, docstore: Store, filterable_docs: List[Document]
+    ):
         docstore.write_documents(filterable_docs)
         filters = {
             "$or": [
@@ -667,54 +669,54 @@ def test_filter_nested_multiple_identical_operators_same_level(self, docstore, f
         )
 
     @pytest.mark.unit
-    def test_write(self, docstore):
+    def test_write(self, docstore: Store):
         doc = Document(content="test doc")
         docstore.write_documents([doc])
         assert docstore.filter_documents(filters={"id": doc.id}) == [doc]
 
     @pytest.mark.unit
-    def test_write_duplicate_fail(self, docstore):
+    def test_write_duplicate_fail(self, docstore: Store):
         doc = Document(content="test doc")
         docstore.write_documents([doc])
         with pytest.raises(DuplicateDocumentError, match=f"ID '{doc.id}' already exists."):
-            docstore.write_documents(documents=[doc])
+            docstore.write_documents(documents=[doc], policy=DuplicatePolicy.FAIL)
         assert docstore.filter_documents(filters={"id": doc.id}) == [doc]
 
     @pytest.mark.unit
-    def test_write_duplicate_skip(self, docstore):
+    def test_write_duplicate_skip(self, docstore: Store):
         doc = Document(content="test doc")
         docstore.write_documents([doc])
-        docstore.write_documents(documents=[doc], duplicates="skip")
+        docstore.write_documents(documents=[doc], policy=DuplicatePolicy.SKIP)
         assert docstore.filter_documents(filters={"id": doc.id}) == [doc]
 
     @pytest.mark.unit
-    def test_write_duplicate_overwrite(self, docstore):
+    def test_write_duplicate_overwrite(self, docstore: Store):
         doc1 = Document(content="test doc 1")
         doc2 = Document(content="test doc 2")
         object.__setattr__(doc2, "id", doc1.id)  # Make two docs with different content but same ID
 
         docstore.write_documents([doc2])
         docstore.filter_documents(filters={"id": doc1.id}) == [doc2]
-        docstore.write_documents(documents=[doc1], duplicates="overwrite")
+        docstore.write_documents(documents=[doc1], policy=DuplicatePolicy.OVERWRITE)
         assert docstore.filter_documents(filters={"id": doc1.id}) == [doc1]
 
     @pytest.mark.unit
-    def test_write_not_docs(self, docstore):
+    def test_write_not_docs(self, docstore: Store):
         with pytest.raises(ValueError, match="Please provide a list of Documents"):
             docstore.write_documents(["not a document for sure"])
 
     @pytest.mark.unit
-    def test_write_not_list(self, docstore):
+    def test_write_not_list(self, docstore: Store):
         with pytest.raises(ValueError, match="Please provide a list of Documents"):
             docstore.write_documents("not a list actually")
 
     @pytest.mark.unit
-    def test_delete_empty(self, docstore):
+    def test_delete_empty(self, docstore: Store):
         with pytest.raises(MissingDocumentError):
             docstore.delete_documents(["test"])
 
     @pytest.mark.unit
-    def test_delete_not_empty(self, docstore):
+    def test_delete_not_empty(self, docstore: Store):
         doc = Document(content="test doc")
         docstore.write_documents([doc])
 
@@ -724,7 +726,7 @@ def test_delete_not_empty(self, docstore):
             assert docstore.filter_documents(filters={"id": doc.id})
 
     @pytest.mark.unit
-    def test_delete_not_empty_nonexisting(self, docstore):
+    def test_delete_not_empty_nonexisting(self, docstore: Store):
         doc = Document(content="test doc")
         docstore.write_documents([doc])
 
diff --git a/test/preview/document_stores/test_memory.py b/test/preview/document_stores/test_memory.py
index 1344234bd4..69206d44de 100644
--- a/test/preview/document_stores/test_memory.py
+++ b/test/preview/document_stores/test_memory.py
@@ -4,7 +4,7 @@
 import pytest
 
 from haystack.preview import Document
-from haystack.preview.document_stores import MemoryDocumentStore
+from haystack.preview.document_stores import Store, MemoryDocumentStore
 
 from test.preview.document_stores._base import DocumentStoreBaseTests
 
@@ -19,7 +19,7 @@ def docstore(self) -> MemoryDocumentStore:
         return MemoryDocumentStore()
 
     @pytest.mark.unit
-    def test_bm25_retrieval(self, docstore):
+    def test_bm25_retrieval(self, docstore: Store):
         docstore = MemoryDocumentStore()
         # Tests if the bm25_retrieval method returns the correct document based on the input query.
         docs = [Document(content="Hello world"), Document(content="Haystack supports multiple languages")]
@@ -29,7 +29,7 @@ def test_bm25_retrieval(self, docstore):
         assert results[0].content == "Haystack supports multiple languages"
 
     @pytest.mark.unit
-    def test_bm25_retrieval_with_empty_document_store(self, docstore, caplog):
+    def test_bm25_retrieval_with_empty_document_store(self, docstore: Store, caplog):
         caplog.set_level(logging.INFO)
         # Tests if the bm25_retrieval method correctly returns an empty list when there are no documents in the store.
         results = docstore.bm25_retrieval(query="How to test this?", top_k=2)
@@ -37,7 +37,7 @@ def test_bm25_retrieval_with_empty_document_store(self, docstore, caplog):
         assert "No documents found for BM25 retrieval. Returning empty list." in caplog.text
 
     @pytest.mark.unit
-    def test_bm25_retrieval_empty_query(self, docstore):
+    def test_bm25_retrieval_empty_query(self, docstore: Store):
         # Tests if the bm25_retrieval method returns a document when the query is an empty string.
         docs = [Document(content="Hello world"), Document(content="Haystack supports multiple languages")]
         docstore.write_documents(docs)
@@ -45,7 +45,7 @@ def test_bm25_retrieval_empty_query(self, docstore):
             docstore.bm25_retrieval(query="", top_k=1)
 
     @pytest.mark.unit
-    def test_bm25_retrieval_filter_only_one_allowed_doc_type_as_string(self, docstore):
+    def test_bm25_retrieval_filter_only_one_allowed_doc_type_as_string(self, docstore: Store):
         # Tests if the bm25_retrieval method returns a document when the query is an empty string.
         docs = [
             Document.from_dict({"content": pd.DataFrame({"language": ["Python", "Java"]}), "content_type": "table"}),
@@ -57,7 +57,7 @@ def test_bm25_retrieval_filter_only_one_allowed_doc_type_as_string(self, docstor
         assert results[0].content == "Haystack supports multiple languages"
 
     @pytest.mark.unit
-    def test_bm25_retrieval_filter_only_one_allowed_doc_type_as_list(self, docstore):
+    def test_bm25_retrieval_filter_only_one_allowed_doc_type_as_list(self, docstore: Store):
         # Tests if the bm25_retrieval method returns a document when the query is an empty string.
         docs = [
             Document.from_dict({"content": pd.DataFrame({"language": ["Python", "Java"]}), "content_type": "table"}),
@@ -69,7 +69,7 @@ def test_bm25_retrieval_filter_only_one_allowed_doc_type_as_list(self, docstore)
         assert results[0].content == "Haystack supports multiple languages"
 
     @pytest.mark.unit
-    def test_bm25_retrieval_filter_two_allowed_doc_type_as_list(self, docstore):
+    def test_bm25_retrieval_filter_two_allowed_doc_type_as_list(self, docstore: Store):
         # Tests if the bm25_retrieval method returns a document when the query is an empty string.
         docs = [
             Document.from_dict({"content": pd.DataFrame({"language": ["Python", "Java"]}), "content_type": "table"}),
@@ -80,7 +80,7 @@ def test_bm25_retrieval_filter_two_allowed_doc_type_as_list(self, docstore):
         assert len(results) == 2
 
     @pytest.mark.unit
-    def test_bm25_retrieval_filter_only_one_not_allowed_doc_type_as_string(self, docstore):
+    def test_bm25_retrieval_filter_only_one_not_allowed_doc_type_as_string(self, docstore: Store):
         # Tests if the bm25_retrieval method returns a document when the query is an empty string.
         docs = [
             Document(content=pd.DataFrame({"language": ["Python", "Java"]}), content_type="table"),
@@ -93,7 +93,7 @@ def test_bm25_retrieval_filter_only_one_not_allowed_doc_type_as_string(self, doc
             docstore.bm25_retrieval(query="Python", top_k=3, filters={"content_type": "audio"})
 
     @pytest.mark.unit
-    def test_bm25_retrieval_filter_only_one_not_allowed_doc_type_as_list(self, docstore):
+    def test_bm25_retrieval_filter_only_one_not_allowed_doc_type_as_list(self, docstore: Store):
         # Tests if the bm25_retrieval method returns a document when the query is an empty string.
         docs = [
             Document(content=pd.DataFrame({"language": ["Python", "Java"]}), content_type="table"),
@@ -106,7 +106,7 @@ def test_bm25_retrieval_filter_only_one_not_allowed_doc_type_as_list(self, docst
             docstore.bm25_retrieval(query="Python", top_k=3, filters={"content_type": ["audio"]})
 
     @pytest.mark.unit
-    def test_bm25_retrieval_filter_two_not_all_allowed_doc_type_as_list(self, docstore):
+    def test_bm25_retrieval_filter_two_not_all_allowed_doc_type_as_list(self, docstore: Store):
         # Tests if the bm25_retrieval method returns a document when the query is an empty string.
         docs = [
             Document.from_dict({"content": pd.DataFrame({"language": ["Python", "Java"]}), "content_type": "table"}),
@@ -119,7 +119,7 @@ def test_bm25_retrieval_filter_two_not_all_allowed_doc_type_as_list(self, docsto
             docstore.bm25_retrieval(query="Python", top_k=3, filters={"content_type": ["text", "audio"]})
 
     @pytest.mark.unit
-    def test_bm25_retrieval_with_different_top_k(self, docstore):
+    def test_bm25_retrieval_with_different_top_k(self, docstore: Store):
         # Tests if the bm25_retrieval method correctly changes the number of returned documents
         # based on the top_k parameter.
         docs = [
@@ -139,7 +139,7 @@ def test_bm25_retrieval_with_different_top_k(self, docstore):
 
     # Test two queries and make sure the results are different
     @pytest.mark.unit
-    def test_bm25_retrieval_with_two_queries(self, docstore):
+    def test_bm25_retrieval_with_two_queries(self, docstore: Store):
         # Tests if the bm25_retrieval method returns different documents for different queries.
         docs = [
             Document(content="Javascript is a popular programming language"),
@@ -158,7 +158,7 @@ def test_bm25_retrieval_with_two_queries(self, docstore):
 
     # Test a query, add a new document and make sure results are appropriately updated
     @pytest.mark.unit
-    def test_bm25_retrieval_with_updated_docs(self, docstore):
+    def test_bm25_retrieval_with_updated_docs(self, docstore: Store):
         # Tests if the bm25_retrieval method correctly updates the retrieved documents when new
         # documents are added to the store.
         docs = [Document(content="Hello world")]
@@ -178,7 +178,7 @@ def test_bm25_retrieval_with_updated_docs(self, docstore):
         assert results[0].content == "Python is a popular programming language"
 
     @pytest.mark.unit
-    def test_bm25_retrieval_with_scale_score(self, docstore):
+    def test_bm25_retrieval_with_scale_score(self, docstore: Store):
         docs = [Document(content="Python programming"), Document(content="Java programming")]
         docstore.write_documents(docs)
 
@@ -191,7 +191,7 @@ def test_bm25_retrieval_with_scale_score(self, docstore):
         assert results[0].score != results1[0].score
 
     @pytest.mark.unit
-    def test_bm25_retrieval_with_table_content(self, docstore):
+    def test_bm25_retrieval_with_table_content(self, docstore: Store):
         # Tests if the bm25_retrieval method correctly returns a dataframe when the content_type is table.
         table_content = pd.DataFrame({"language": ["Python", "Java"], "use": ["Data Science", "Web Development"]})
         docs = [
diff --git a/test/preview/pipeline/test_pipeline.py b/test/preview/pipeline/test_pipeline.py
index dfe437bd89..31e50ac4fb 100644
--- a/test/preview/pipeline/test_pipeline.py
+++ b/test/preview/pipeline/test_pipeline.py
@@ -4,6 +4,7 @@
 import pytest
 
 from haystack.preview import Pipeline, component, NoSuchStoreError, ComponentInput, ComponentOutput
+from haystack.preview.document_stores import StoreMixin, MultiStoreMixin
 
 
 class MockStore:
@@ -28,29 +29,236 @@ def test_pipeline_store_api():
 
 
 @pytest.mark.unit
-def test_pipeline_stores_in_params():
+def test_pipeline_storemixin_one_existing_docstore():
     store_1 = MockStore()
     store_2 = MockStore()
 
     @component
-    class MockComponent:
+    class MockComponent(StoreMixin):
         @dataclass
         class Input(ComponentInput):
             value: int
-            stores: Dict[str, Any]
 
         @dataclass
         class Output(ComponentOutput):
             value: int
 
         def run(self, data: Input) -> Output:
-            assert data.stores == {"first_store": store_1, "second_store": store_2}
+            assert self.store == store_1
             return MockComponent.Output(value=data.value)
 
     pipe = Pipeline()
-    pipe.add_component("component", MockComponent())
+    pipe.add_store(name="first_store", store=store_1)
+    pipe.add_store(name="second_store", store=store_2)
+    pipe.add_component("component", MockComponent(), stores=["first_store"])
+    assert pipe.run(data={"component": MockComponent.Input(value=1)}) == {"component": MockComponent.Output(value=1)}
+
+
+@pytest.mark.unit
+def test_pipeline_storemixin_no_docstore():
+    store_1 = MockStore()
+    store_2 = MockStore()
+
+    @component
+    class MockComponent(StoreMixin):
+        @dataclass
+        class Input(ComponentInput):
+            value: int
+
+        @dataclass
+        class Output(ComponentOutput):
+            value: int
+
+        def run(self, data: Input) -> Output:
+            assert data.store == store_1
+            return MockComponent.Output(value=data.value)
+
+    pipe = Pipeline()
+    pipe.add_store(name="first_store", store=store_1)
+    pipe.add_store(name="second_store", store=store_2)
+
+    with pytest.raises(ValueError, match="Component 'component' needs exactly one store."):
+        pipe.add_component("component", MockComponent())
+
+
+@pytest.mark.unit
+def test_pipeline_storemixin_many_existing_docstores():
+    store_1 = MockStore()
+    store_2 = MockStore()
+
+    @component
+    class MockComponent(StoreMixin):
+        @dataclass
+        class Input(ComponentInput):
+            value: int
+
+        @dataclass
+        class Output(ComponentOutput):
+            value: int
+
+        def run(self, data: Input) -> Output:
+            assert data.store == store_1
+            return MockComponent.Output(value=data.value)
+
+    pipe = Pipeline()
+    pipe.add_store(name="first_store", store=store_1)
+    pipe.add_store(name="second_store", store=store_2)
+
+    with pytest.raises(ValueError, match="Component 'component' needs exactly one store."):
+        pipe.add_component("component", MockComponent(), stores=["first_store", "second_store"])
+
+
+@pytest.mark.unit
+def test_pipeline_storemixin_one_non_existing_docstore():
+    store_1 = MockStore()
+    store_2 = MockStore()
+
+    @component
+    class MockComponent(StoreMixin):
+        @dataclass
+        class Input(ComponentInput):
+            value: int
 
+        @dataclass
+        class Output(ComponentOutput):
+            value: int
+
+        def run(self, data: Input) -> Output:
+            assert data.store == store_1
+            return MockComponent.Output(value=data.value)
+
+    pipe = Pipeline()
     pipe.add_store(name="first_store", store=store_1)
     pipe.add_store(name="second_store", store=store_2)
 
+    with pytest.raises(NoSuchStoreError, match="Store named 'wrong_store' not found."):
+        pipe.add_component("component", MockComponent(), stores=["wrong_store"])
+
+
+@pytest.mark.unit
+def test_pipeline_storesmixin_one_existing_docstore():
+    store_1 = MockStore()
+    store_2 = MockStore()
+
+    @component
+    class MockComponent(MultiStoreMixin):
+        @dataclass
+        class Input(ComponentInput):
+            value: int
+
+        @dataclass
+        class Output(ComponentOutput):
+            value: int
+
+        def run(self, data: Input) -> Output:
+            assert self.stores == {"first_store": store_1}
+            return MockComponent.Output(value=data.value)
+
+    pipe = Pipeline()
+    pipe.add_store(name="first_store", store=store_1)
+    pipe.add_store(name="second_store", store=store_2)
+    pipe.add_component("component", MockComponent(), stores=["first_store"])
     assert pipe.run(data={"component": MockComponent.Input(value=1)}) == {"component": MockComponent.Output(value=1)}
+
+
+@pytest.mark.unit
+def test_pipeline_storesmixin_no_docstore():
+    store_1 = MockStore()
+    store_2 = MockStore()
+
+    @component
+    class MockComponent(MultiStoreMixin):
+        @dataclass
+        class Input(ComponentInput):
+            value: int
+
+        @dataclass
+        class Output(ComponentOutput):
+            value: int
+
+        def run(self, data: Input) -> Output:
+            assert self.stores == {}
+            return MockComponent.Output(value=data.value)
+
+    pipe = Pipeline()
+    pipe.add_store(name="first_store", store=store_1)
+    pipe.add_store(name="second_store", store=store_2)
+    pipe.add_component("component", MockComponent())
+    assert pipe.run(data={"component": MockComponent.Input(value=1)}) == {"component": MockComponent.Output(value=1)}
+
+
+@pytest.mark.unit
+def test_pipeline_storesmixin_many_existing_docstores():
+    store_1 = MockStore()
+    store_2 = MockStore()
+
+    @component
+    class MockComponent(MultiStoreMixin):
+        @dataclass
+        class Input(ComponentInput):
+            value: int
+
+        @dataclass
+        class Output(ComponentOutput):
+            value: int
+
+        def run(self, data: Input) -> Output:
+            assert self.stores == {"first_store": store_1, "second_store": store_2}
+            return MockComponent.Output(value=data.value)
+
+    pipe = Pipeline()
+    pipe.add_store(name="first_store", store=store_1)
+    pipe.add_store(name="second_store", store=store_2)
+    pipe.add_component("component", MockComponent(), stores=["first_store", "second_store"])
+
+
+@pytest.mark.unit
+def test_pipeline_storemixin_one_non_existing_docstore_alone():
+    store_1 = MockStore()
+    store_2 = MockStore()
+
+    @component
+    class MockComponent(MultiStoreMixin):
+        @dataclass
+        class Input(ComponentInput):
+            value: int
+
+        @dataclass
+        class Output(ComponentOutput):
+            value: int
+
+        def run(self, data: Input) -> Output:
+            return MockComponent.Output(value=data.value)
+
+    pipe = Pipeline()
+    pipe.add_store(name="first_store", store=store_1)
+    pipe.add_store(name="second_store", store=store_2)
+
+    with pytest.raises(NoSuchStoreError, match="Store named 'wrong_store' not found."):
+        pipe.add_component("component", MockComponent(), stores=["wrong_store"])
+
+
+@pytest.mark.unit
+def test_pipeline_storemixin_one_non_existing_docstore_among_existing_ones():
+    store_1 = MockStore()
+    store_2 = MockStore()
+
+    @component
+    class MockComponent(MultiStoreMixin):
+        @dataclass
+        class Input(ComponentInput):
+            value: int
+
+        @dataclass
+        class Output(ComponentOutput):
+            value: int
+
+        def run(self, data: Input) -> Output:
+            return MockComponent.Output(value=data.value)
+
+    pipe = Pipeline()
+    pipe.add_store(name="first_store", store=store_1)
+    pipe.add_store(name="second_store", store=store_2)
+
+    with pytest.raises(NoSuchStoreError, match="Store named 'wrong_store' not found."):
+        pipe.add_component("component", MockComponent(), stores=["first_store", "wrong_store"])


Code review comments:
path: haystack/preview/document_stores/mixins.py
text: @user1:
We don't need the `Protocol`s in this case.
The interface is already implemented by the `Mixin`s.

@author:
In some cases the signature (the typing) will differ. I was not sure it makes sense to have a protocol at all but I wanted to get your opinion first. I'm going to remove it.

diff_hunk: @@ -39,28 +40,54 @@ class Output(ComponentOutput):
 
         documents: List[Document]
 
-    def __init__(
-        self,
-        document_store_name: str,
-        filters: Optional[Dict[str, Any]] = None,
-        top_k: int = 10,
-        scale_score: bool = True,
-    ):
+    def __init__(self, filters: Optional[Dict[str, Any]] = None, top_k: int = 10, scale_score: bool = True):
         """
         Create a MemoryRetriever component.
 
-        :param document_store_name: The name of the MemoryDocumentStore to retrieve documents from.
         :param filters: A dictionary with filters to narrow down the search space (default is None).
         :param top_k: The maximum number of documents to retrieve (default is 10).
         :param scale_score: Whether to scale the BM25 score or not (default is True).
 
         :raises ValueError: If the specified top_k is not > 0.
         """
-        self.document_store_name = document_store_name
         if top_k <= 0:
             raise ValueError(f"top_k must be > 0, but got {top_k}")
         self.defaults = {"top_k": top_k, "scale_score": scale_score, "filters": filters or {}}
 
+    @property
+    def store(self) -> Optional[MemoryDocumentStore]:
+        """
+        This property allows Pipelines to connect the component with the stores it requires.
+
+        Stores have to be instances of MemoryDocumentStore, or the assignment will fail.
+        """
+        return getattr(self, "_store", None)
+
+    @store.setter
+    def store(self, store: MemoryDocumentStore):
+        """
+        This property allows Pipelines to connect the component with the stores it requires.
+
+        Stores have to be instances of MemoryDocumentStore, or the assignment will fail.
+
+        :param store: the MemoryDocumentStore instance to retrieve from.
+        :raises ValueError if the store is not an instance of MemoryDocumentStore.
+        """
+        if not store:
+            raise ValueError("Can't set the value of the store to None.")
+        if not isinstance(store, MemoryDocumentStore):
+            raise ValueError("MemoryRetriever can only be used with a MemoryDocumentStore instance.")
+        self._store = store
path: haystack/preview/components/retrievers/memory.py
text: @user1:
I think we must replace these with the `StoreMixin` methods.

@author:
They have a different signature...

diff_hunk: @@ -24,64 +26,59 @@ def mock_docs():
 class Test_MemoryRetriever(BaseTestComponent):
     @pytest.mark.unit
     def test_save_load(self, tmp_path):
-        self.assert_can_be_saved_and_loaded_in_pipeline(MemoryRetriever(document_store_name="memory"), tmp_path)
+        self.assert_can_be_saved_and_loaded_in_pipeline(MemoryRetriever(), tmp_path)
 
     @pytest.mark.unit
     def test_save_load_with_parameters(self, tmp_path):
-        self.assert_can_be_saved_and_loaded_in_pipeline(
-            MemoryRetriever(document_store_name="memory", top_k=5, scale_score=False), tmp_path
-        )
+        self.assert_can_be_saved_and_loaded_in_pipeline(MemoryRetriever(top_k=5, scale_score=False), tmp_path)
 
     @pytest.mark.unit
     def test_init_default(self):
-        retriever = MemoryRetriever(document_store_name="memory")
-        assert retriever.document_store_name == "memory"
+        retriever = MemoryRetriever()
         assert retriever.defaults == {"filters": {}, "top_k": 10, "scale_score": True}
 
     @pytest.mark.unit
     def test_init_with_parameters(self):
-        retriever = MemoryRetriever(document_store_name="memory-test", top_k=5, scale_score=False)
-        assert retriever.document_store_name == "memory-test"
+        retriever = MemoryRetriever(top_k=5, scale_score=False)
         assert retriever.defaults == {"filters": {}, "top_k": 5, "scale_score": False}
 
     @pytest.mark.unit
     def test_init_with_invalid_top_k_parameter(self):
         with pytest.raises(ValueError, match="top_k must be > 0, but got -2"):
-            MemoryRetriever(document_store_name="memory-test", top_k=-2, scale_score=False)
+            MemoryRetriever(top_k=-2, scale_score=False)
 
     @pytest.mark.unit
     def test_valid_run(self, mock_docs):
         top_k = 5
         ds = MemoryDocumentStore()
         ds.write_documents(mock_docs)
-        mr = MemoryRetriever(document_store_name="memory", top_k=top_k)
-        result: MemoryRetriever.Output = mr.run(data=MemoryRetriever.Input(query="PHP", stores={"memory": ds}))
+
+        mr = MemoryRetriever(top_k=top_k)
+        mr.store = ds
+        result = mr.run(data=MemoryRetriever.Input(query="PHP"))
path: test/preview/components/retrievers/test_memory_retriever.py
text: @user1:
```suggestion
        retriever = MemoryRetriever(top_k=top_k)
        retriever.store = ds
        result = retriever.run(data=MemoryRetriever.Input(query="PHP"))
```

diff_hunk: @@ -24,64 +26,59 @@ def mock_docs():
 class Test_MemoryRetriever(BaseTestComponent):
     @pytest.mark.unit
     def test_save_load(self, tmp_path):
-        self.assert_can_be_saved_and_loaded_in_pipeline(MemoryRetriever(document_store_name="memory"), tmp_path)
+        self.assert_can_be_saved_and_loaded_in_pipeline(MemoryRetriever(), tmp_path)
 
     @pytest.mark.unit
     def test_save_load_with_parameters(self, tmp_path):
-        self.assert_can_be_saved_and_loaded_in_pipeline(
-            MemoryRetriever(document_store_name="memory", top_k=5, scale_score=False), tmp_path
-        )
+        self.assert_can_be_saved_and_loaded_in_pipeline(MemoryRetriever(top_k=5, scale_score=False), tmp_path)
 
     @pytest.mark.unit
     def test_init_default(self):
-        retriever = MemoryRetriever(document_store_name="memory")
-        assert retriever.document_store_name == "memory"
+        retriever = MemoryRetriever()
         assert retriever.defaults == {"filters": {}, "top_k": 10, "scale_score": True}
 
     @pytest.mark.unit
     def test_init_with_parameters(self):
-        retriever = MemoryRetriever(document_store_name="memory-test", top_k=5, scale_score=False)
-        assert retriever.document_store_name == "memory-test"
+        retriever = MemoryRetriever(top_k=5, scale_score=False)
         assert retriever.defaults == {"filters": {}, "top_k": 5, "scale_score": False}
 
     @pytest.mark.unit
     def test_init_with_invalid_top_k_parameter(self):
         with pytest.raises(ValueError, match="top_k must be > 0, but got -2"):
-            MemoryRetriever(document_store_name="memory-test", top_k=-2, scale_score=False)
+            MemoryRetriever(top_k=-2, scale_score=False)
 
     @pytest.mark.unit
     def test_valid_run(self, mock_docs):
         top_k = 5
         ds = MemoryDocumentStore()
         ds.write_documents(mock_docs)
-        mr = MemoryRetriever(document_store_name="memory", top_k=top_k)
-        result: MemoryRetriever.Output = mr.run(data=MemoryRetriever.Input(query="PHP", stores={"memory": ds}))
+
+        mr = MemoryRetriever(top_k=top_k)
+        mr.store = ds
+        result = mr.run(data=MemoryRetriever.Input(query="PHP"))
 
         assert getattr(result, "documents")
         assert len(result.documents) == top_k
         assert result.documents[0].content == "PHP is a popular programming language"
 
-    @pytest.mark.unit
-    def test_invalid_run_wrong_store_name(self):
-        # Test invalid run with wrong store name
-        ds = MemoryDocumentStore()
-        mr = MemoryRetriever(document_store_name="memory")
-        with pytest.raises(ValueError, match=r"MemoryRetriever's document store 'memory' not found"):
-            invalid_input_data = MemoryRetriever.Input(
-                query="test", top_k=10, scale_score=True, stores={"invalid_store": ds}
-            )
-            mr.run(invalid_input_data)
-
     @pytest.mark.unit
     def test_invalid_run_wrong_store_type(self):
-        # Test invalid run with wrong store type
-        ds = MemoryDocumentStore()
-        mr = MemoryRetriever(document_store_name="memory")
-        with pytest.raises(ValueError, match=r"MemoryRetriever can only be used with a MemoryDocumentStore instance."):
-            invalid_input_data = MemoryRetriever.Input(
-                query="test", top_k=10, scale_score=True, stores={"memory": "not a MemoryDocumentStore"}
-            )
-            mr.run(invalid_input_data)
+        class MockStore(Store):
+            def count_documents(self) -> int:
+                return 0
+
+            def filter_documents(self, filters: Optional[Dict[str, Any]] = None) -> List[Document]:
+                return []
+
+            def write_documents(self, documents: List[Document], duplicates: DuplicatePolicy = "fail") -> None:
+                return
+
+            def delete_documents(self, document_ids: List[str]) -> None:
+                return
+
+        mr = MemoryRetriever()
+        with pytest.raises(ValueError, match="MemoryRetriever can only be used with a MemoryDocumentStore instance"):
+            mr.store = MockStore()
path: test/preview/components/retrievers/test_memory_retriever.py
text: @user1:
```suggestion
        retriever = MemoryRetriever()
        with pytest.raises(ValueError, match="MemoryRetriever can only be used with a MemoryDocumentStore instance"):
            retriever.store = MockStore()
```

diff_hunk: @@ -94,11 +91,11 @@ def test_invalid_run_wrong_store_type(self):
     def test_run_with_pipeline(self, mock_docs, query: str, query_result: str):
         ds = MemoryDocumentStore()
         ds.write_documents(mock_docs)
-        mr = MemoryRetriever(document_store_name="memory")
+        mr = MemoryRetriever()
path: test/preview/components/retrievers/test_memory_retriever.py
text: @user1:
```suggestion
        retriever = MemoryRetriever()
```

diff_hunk: @@ -119,11 +116,11 @@ def test_run_with_pipeline(self, mock_docs, query: str, query_result: str):
     def test_run_with_pipeline_and_top_k(self, mock_docs, query: str, query_result: str, top_k: int):
         ds = MemoryDocumentStore()
         ds.write_documents(mock_docs)
-        mr = MemoryRetriever(document_store_name="memory")
+        mr = MemoryRetriever()
path: test/preview/components/retrievers/test_memory_retriever.py
text: @user1:
```suggestion
        retriever = MemoryRetriever()
```

diff_hunk: @@ -0,0 +1,67 @@
+from typing import Protocol, Dict, Optional
+
+from haystack.preview.document_stores.protocols import Store
+
+
+class StoreComponent(Protocol):
+    """
+    Represents a component that needs a single store to run.
+    """
+
+    _store: Store
+
+    @property
+    def store(self) -> Optional[Store]:
+        ...
+
+    @store.setter
+    def store(self, store: Store):
+        ...
+
+
+class StoreMixin(StoreComponent):
+    """
+    Adds the capability of a component to use a single document store from the `self.store` property.
+    """
+
+    @property
+    def store(self) -> Optional[Store]:
+        return getattr(self, "_store", None)
path: haystack/preview/document_stores/mixins.py
text: @user2:
```suggestion
    self._store: Store = None

    @user1
    def store(self) -> Optional[Store]:
        return self._store

@author:
Wait, you can't do that. `self` is not defined in that scope.

@user2:
Ah yeah, bad copy and paste. I meant to have this:

```suggestion
    _store: Store = None

    @user1
    def store(self) -> Optional[Store]:
        return self._store

@author:
Aren't you defining a class attribute that way?


@user2:
If you access it from the class it is, but all instances of classes that inherit this mixin will store their own value.

```
In [1]: class Foo:
   ...:     value: int = 10
   ...:

In [2]: Foo.value
Out[2]: 10

In [3]: f = Foo()

In [4]: b = Foo()

In [5]: f.value, b.value
Out[5]: (10, 10)

In [6]: f.value = 100

In [7]: f.value, b.value
Out[7]: (100, 10)

In [8]: Foo.value
Out[8]: 10
```

diff_hunk: @@ -119,11 +116,11 @@ def test_run_with_pipeline(self, mock_docs, query: str, query_result: str):
     def test_run_with_pipeline_and_top_k(self, mock_docs, query: str, query_result: str, top_k: int):
         ds = MemoryDocumentStore()
         ds.write_documents(mock_docs)
-        mr = MemoryRetriever(document_store_name="memory")
+        mr = MemoryRetriever()
 
         pipeline = Pipeline()
-        pipeline.add_component("retriever", mr)
         pipeline.add_store("memory", ds)
+        pipeline.add_component("retriever", mr, stores=["memory"])
path: test/preview/components/retrievers/test_memory_retriever.py
text: @user1:
```suggestion
        pipeline.add_component("retriever", retriever, stores=["memory"])
```

diff_hunk: @@ -28,29 +29,236 @@ def test_pipeline_store_api():
 
 
 @pytest.mark.unit
-def test_pipeline_stores_in_params():
+def test_pipeline_storemixin_one_existing_docstore():
     store_1 = MockStore()
     store_2 = MockStore()
 
     @component
-    class MockComponent:
+    class MockComponent(StoreMixin):
         @dataclass
         class Input(ComponentInput):
             value: int
-            stores: Dict[str, Any]
 
         @dataclass
         class Output(ComponentOutput):
             value: int
 
         def run(self, data: Input) -> Output:
-            assert data.stores == {"first_store": store_1, "second_store": store_2}
+            assert self.store == store_1
path: test/preview/pipeline/test_pipeline.py
text: @user1:
I wouldn't `assert` this here, let's keep it in the body of the test.

@user1:
Same thing for all the other `assert`s in other tests.

@author:
Is there any reason why not? I find testing in run() quite elegant - it tests that the method that needs the store always has access to them. Given the amount of magic we often do in Canals, I like being able to assert in run(), but if there are downsides I'll take them out.

@user1:
It makes it less maintainable in my opinion, and also a bit hidden. I'd expect that in the future we'll have a sort of factory to build these mocks and it would be quite easy to lose that `assert` when making that change.

@author:
Done :heavy_check_mark:

diff_hunk: @@ -23,9 +23,9 @@ class Pipeline(CanalsPipeline):
 
     def __init__(self):
         super().__init__()
-        self.stores = {}
+        self.stores: Dict[str, Store] = {}
path: haystack/preview/pipeline.py
text: @user1:
I think the stores should be part of the private interface of `Pipeline`, I see no reason to have accesor methods otherwise.

@author:
I don't see much harm at leaving them accessible either, but no strong opinions. I'll turn it private.

diff_hunk: @@ -55,29 +55,41 @@ def get_store(self, name: str) -> object:
         except KeyError as e:
             raise NoSuchStoreError(f"No store named '{name}' is connected to this pipeline.") from e
 
-    def run(self, data: Dict[str, ComponentInput], debug: bool = False) -> Dict[str, ComponentOutput]:
+    def add_component(self, name: str, instance: Any, stores: Optional[List[str]] = None) -> None:
         """
-        Wrapper on top of Canals Pipeline.run(). Adds the `stores` parameter to all nodes.
+        Make this component available to the pipeline. Components are not connected to anything by default:
+        use `Pipeline.connect()` to connect components together.
+
+        Component names must be unique, but component instances can be reused if needed.
+
+        If `stores` has a value, the pipeline will also connect this component to the requested document store(s).
+        Note that only components that respect the StoreMixin or StoresMixin protocols can be connected to stores.
 
-        :params data: the inputs to give to the input components of the Pipeline.
-        :params parameters: a dictionary with all the parameters of all the components, namespaced by component.
-        :params debug: whether to collect and return debug information.
-        :returns A dictionary with the outputs of the output components of the Pipeline.
+        :param name: the name of the component.
+        :param instance: the component instance.
+        :param stores: the stores this component needs access to, if any.
+        :raises ValueError: if a component with the same name already exists
+        :raises PipelineValidationError: if the given instance is not a component
         """
-        # Get all nodes in this pipelines instance
-        for node_name in self.graph.nodes:
-            # Get node inputs
-            node = self.graph.nodes[node_name]["instance"]
-            input_params = find_input_sockets(node)
-
-            # If the node needs a store, adds the list of stores to its default inputs
-            if "stores" in input_params:
-                if not hasattr(node, "defaults"):
-                    setattr(node, "defaults", {})
-                node.defaults["stores"] = self.stores
-
-        # Run the pipeline
-        return super().run(data=data, debug=debug)
+        super().add_component(name, instance)
+
+        stores = stores or []
+        for store in stores:
+            if store not in self.stores:
+                raise NoSuchStoreError(
+                    f"Store named '{store}' not found. "
+                    f"Add it with 'pipeline.add_store('{store}', <the docstore instance>)'."
+                )
+
+        # This component implements the MultiStoreComponent protocol
+        if hasattr(instance, "stores"):
+            instance.stores = {store: self.stores[store] for store in stores}
+
+        # This component implements the StoreComponent protocol
+        elif hasattr(instance, "store"):
path: haystack/preview/pipeline.py
text: @user1:
You'll need to import both `StoreMixin` and `MultiStoreMixin`.

```suggestion
        if issubclass(type(instance), MultiStoreMixin):
            instance.stores = {store: self.stores[store] for store in stores}
        elif issubclass(type(instance), StoreMixin):
```

@author:
That won't work for `MemoryRetriever`, it's not using `StoreMixin` and it can't do so.

diff_hunk: @@ -39,28 +40,54 @@ class Output(ComponentOutput):
 
         documents: List[Document]
 
-    def __init__(
-        self,
-        document_store_name: str,
-        filters: Optional[Dict[str, Any]] = None,
-        top_k: int = 10,
-        scale_score: bool = True,
-    ):
+    def __init__(self, filters: Optional[Dict[str, Any]] = None, top_k: int = 10, scale_score: bool = True):
         """
         Create a MemoryRetriever component.
 
-        :param document_store_name: The name of the MemoryDocumentStore to retrieve documents from.
         :param filters: A dictionary with filters to narrow down the search space (default is None).
         :param top_k: The maximum number of documents to retrieve (default is 10).
         :param scale_score: Whether to scale the BM25 score or not (default is True).
 
         :raises ValueError: If the specified top_k is not > 0.
         """
-        self.document_store_name = document_store_name
         if top_k <= 0:
             raise ValueError(f"top_k must be > 0, but got {top_k}")
         self.defaults = {"top_k": top_k, "scale_score": scale_score, "filters": filters or {}}
 
+    @property
+    def store(self) -> Optional[MemoryDocumentStore]:
+        """
+        This property allows Pipelines to connect the component with the stores it requires.
+
+        Stores have to be instances of MemoryDocumentStore, or the assignment will fail.
+        """
+        return getattr(self, "_store", None)
+
+    @store.setter
+    def store(self, store: MemoryDocumentStore):
+        """
+        This property allows Pipelines to connect the component with the stores it requires.
+
+        Stores have to be instances of MemoryDocumentStore, or the assignment will fail.
+
+        :param store: the MemoryDocumentStore instance to retrieve from.
+        :raises ValueError if the store is not an instance of MemoryDocumentStore.
+        """
+        if not store:
+            raise ValueError("Can't set the value of the store to None.")
+        if not isinstance(store, MemoryDocumentStore):
+            raise ValueError("MemoryRetriever can only be used with a MemoryDocumentStore instance.")
+        self._store = store
+
+    def warmup(self):
+        """
+        Double-checks that a store is set before running this component in a pipeline.
+        """
+        if not self.store:
+            raise ValueError(
+                "MemoryRetriever needs a store to run: use the 'store' parameter of 'add_component' to connect them."
+            )
path: haystack/preview/components/retrievers/memory.py
text: @user1:
Isn't this sort of validation? 

@author:
Yes and I hate that I had to put it in here... where shall I put it? In `run()` is far too late imho, but in `__init__` it makes impossible to initialize it.

@author:
I mean, I also have it in run(), but that's only for direct calls to the method, outside of Pipelines. Pipelines should have a way to do this kind of checks earlier and warm_up seems like the only valid place. Open to suggestions