feat: add run_async to CacheChecker

haystack/components/caching/cache_checker.py

@@ -94,3 +94,28 @@ def run(self, items: list[Any]) -> dict[str, Any]:
             else:
                 misses.append(item)
         return {"hits": found_documents, "misses": misses}
+
+    @component.output_types(hits=list[Document], misses=list)
+    async def run_async(self, items: list[Any]) -> dict[str, Any]:
+        """
+        Asynchronously checks if any document associated with the specified cache field is already present in the store.
+
+        :param items:
+            Values to be checked against the cache field.
+        :return:
+            A dictionary with two keys:
+            - `hits` - Documents that matched with at least one of the items.
+            - `misses` - Items that were not present in any documents.
+        """
+        found_documents = []
+        misses = []
+
+        for item in items:
+            filters = {"field": self.cache_field, "operator": "==", "value": item}
+            # 'ignore' since filter_documents_async is not defined in the Protocol but exists in the implementations
+            found = await self.document_store.filter_documents_async(filters=filters)  # type: ignore[attr-defined]
+            if found:
+                found_documents.extend(found)
+            else:
+                misses.append(item)
+        return {"hits": found_documents, "misses": misses}

releasenotes/notes/add-run-async-for-CacheChecker-a42fa8062c33466b.yaml

@@ -0,0 +1,4 @@
+---
+features:
+  - |
+    Add ``run_async`` to ``CacheChecker``, enabling it to be used in ``AsyncPipeline`` without blocking the event loop.

test/components/caching/test_cache_checker_async.py

@@ -0,0 +1,53 @@
+# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from haystack import Document
+from haystack.components.caching.cache_checker import CacheChecker
+
+
+class TestCacheCheckerAsync:
+    @pytest.mark.asyncio
+    async def test_run_async(self, in_memory_doc_store):
+        documents = [
+            Document(content="doc1", meta={"url": "https://example.com/1"}),
+            Document(content="doc2", meta={"url": "https://example.com/2"}),
+            Document(content="doc3", meta={"url": "https://example.com/1"}),
+            Document(content="doc4", meta={"url": "https://example.com/2"}),
+        ]
+        in_memory_doc_store.write_documents(documents)
+        checker = CacheChecker(in_memory_doc_store, cache_field="url")
+        results = await checker.run_async(items=["https://example.com/1", "https://example.com/5"])
+        assert results == {"hits": [documents[0], documents[2]], "misses": ["https://example.com/5"]}
+
+    @pytest.mark.asyncio
+    async def test_run_async_all_hits(self, in_memory_doc_store):
+        documents = [
+            Document(content="doc1", meta={"url": "https://example.com/1"}),
+            Document(content="doc2", meta={"url": "https://example.com/2"}),
+        ]
+        in_memory_doc_store.write_documents(documents)
+        checker = CacheChecker(in_memory_doc_store, cache_field="url")
+        results = await checker.run_async(items=["https://example.com/1", "https://example.com/2"])
+        assert results["hits"] == documents
+        assert results["misses"] == []
+
+    @pytest.mark.asyncio
+    async def test_run_async_all_misses(self, in_memory_doc_store):
+        checker = CacheChecker(in_memory_doc_store, cache_field="url")
+        results = await checker.run_async(items=["https://example.com/1", "https://example.com/2"])
+        assert results["hits"] == []
+        assert results["misses"] == ["https://example.com/1", "https://example.com/2"]
+
+    @pytest.mark.asyncio
+    async def test_run_async_filters_syntax(self):
+        mock_store = MagicMock()
+        mock_store.filter_documents_async = AsyncMock(return_value=[])
+        checker = CacheChecker(document_store=mock_store, cache_field="url")
+        await checker.run_async(items=["https://example.com/1"])
+        expected_filters = {"field": "url", "operator": "==", "value": "https://example.com/1"}
+        mock_store.filter_documents_async.assert_awaited_once_with(filters=expected_filters)