fix: UpdateByFilterAsyncTest, CountDocumentsByFilterAsyncTest, CountUniqueMetadataByFilterAsyncTest (#10953)

* fix: address #10920

* formatting

---------

Co-authored-by: David S. Batista <dsbatista@gmail.com>

Author: pandego

haystack/document_stores/in_memory/document_store.py

@@ -863,6 +863,51 @@ async def delete_documents_async(self, document_ids: list[str]) -> None:
             self.executor, lambda: self.delete_documents(document_ids=document_ids)
         )
 
+    async def update_by_filter_async(self, filters: dict[str, Any], meta: dict[str, Any]) -> int:
+        """
+        Updates the metadata of all documents that match the provided filters.
+
+        :param filters: The filters to apply to select documents for updating.
+            For filter syntax, see filter_documents.
+        :param meta: The metadata fields to update. These will be merged with existing metadata.
+        :returns: The number of documents updated.
+        """
+        return await asyncio.get_running_loop().run_in_executor(
+            self.executor, lambda: self.update_by_filter(filters=filters, meta=meta)
+        )
+
+    async def count_documents_by_filter_async(self, filters: dict[str, Any]) -> int:
+        """
+        Returns the number of documents that match the provided filters.
+
+        :param filters: The filters to apply.
+            For a detailed specification of the filters, refer to the
+            [documentation](https://docs.haystack.deepset.ai/docs/metadata-filtering).
+        :returns: The number of documents that match the filters.
+        """
+        return await asyncio.get_running_loop().run_in_executor(
+            self.executor, lambda: self.count_documents_by_filter(filters=filters)
+        )
+
+    async def count_unique_metadata_by_filter_async(
+        self, filters: dict[str, Any], metadata_fields: list[str]
+    ) -> dict[str, int]:
+        """
+        Returns the number of unique values for each specified metadata field from documents matching the filters.
+
+        :param filters: The filters to apply.
+            For a detailed specification of the filters, refer to the
+            [documentation](https://docs.haystack.deepset.ai/docs/metadata-filtering).
+        :param metadata_fields: List of field names to count unique values for.
+            Field names can include or omit the "meta." prefix.
+        :returns: A dictionary mapping each metadata field name (without "meta." prefix)
+            to the count of its unique values among the filtered documents.
+        """
+        return await asyncio.get_running_loop().run_in_executor(
+            self.executor,
+            lambda: self.count_unique_metadata_by_filter(filters=filters, metadata_fields=metadata_fields),
+        )
+
     async def delete_all_documents_async(self) -> None:
         """
         Deletes all documents in the document store.

haystack/testing/document_store.py

@@ -930,6 +930,43 @@ def test_update_by_filter_advanced_filters(document_store: DocumentStore):
         assert len(featured_docs) == 2
 
 
+class UpdateByFilterAsyncTest:
+    """
+    Tests for Document Store update_by_filter_async().
+
+    Only mix in for stores that implement update_by_filter_async.
+    """
+
+    @staticmethod
+    @pytest.mark.asyncio
+    async def test_update_by_filter_async(document_store: DocumentStore, filterable_docs: list[Document]):
+        """Update documents matching a filter asynchronously and verify count and meta changes."""
+        document_store.write_documents(filterable_docs)
+        expected_count = len([d for d in filterable_docs if d.meta.get("chapter") == "intro"])
+        assert document_store.count_documents() == len(filterable_docs)
+
+        sig = inspect.signature(document_store.update_by_filter_async)  # type:ignore[attr-defined]
+        params = {"refresh": True} if "refresh" in sig.parameters else {}
+        updated_count = await document_store.update_by_filter_async(  # type:ignore[attr-defined]
+            filters={"field": "meta.chapter", "operator": "==", "value": "intro"}, meta={"updated": True}, **params
+        )
+        assert updated_count == expected_count
+
+        updated_docs = document_store.filter_documents(
+            filters={"field": "meta.updated", "operator": "==", "value": True}
+        )
+        assert len(updated_docs) == expected_count
+        for doc in updated_docs:
+            assert doc.meta["chapter"] == "intro"
+            assert doc.meta["updated"] is True
+
+        not_updated_docs = document_store.filter_documents(
+            filters={"field": "meta.chapter", "operator": "==", "value": "abstract"}
+        )
+        for doc in not_updated_docs:
+            assert doc.meta.get("updated") is not True
+
+
 class CountDocumentsByFilterTest:
     """
     Tests for Document Store count_documents_by_filter().
@@ -1005,6 +1042,85 @@ def test_count_documents_by_filter_empty_collection(document_store: DocumentStor
         assert count == 0
 
 
+class CountDocumentsByFilterAsyncTest:
+    """
+    Tests for Document Store count_documents_by_filter_async().
+
+    Only mix in for stores that implement count_documents_by_filter_async.
+    """
+
+    @staticmethod
+    @pytest.mark.asyncio
+    async def test_count_documents_by_filter_async_simple(document_store: DocumentStore):
+        """Test count_documents_by_filter_async() with a simple equality filter."""
+        docs = [
+            Document(content="Doc 1", meta={"category": "A", "status": "active"}),
+            Document(content="Doc 2", meta={"category": "B", "status": "active"}),
+            Document(content="Doc 3", meta={"category": "A", "status": "inactive"}),
+            Document(content="Doc 4", meta={"category": "A", "status": "active"}),
+        ]
+        document_store.write_documents(docs)
+        assert document_store.count_documents() == 4
+
+        count = await document_store.count_documents_by_filter_async(  # type:ignore[attr-defined]
+            filters={"field": "meta.category", "operator": "==", "value": "A"}
+        )
+        assert count == 3
+
+        count = await document_store.count_documents_by_filter_async(  # type:ignore[attr-defined]
+            filters={"field": "meta.category", "operator": "==", "value": "B"}
+        )
+        assert count == 1
+
+    @staticmethod
+    @pytest.mark.asyncio
+    async def test_count_documents_by_filter_async_compound(document_store: DocumentStore):
+        """Test count_documents_by_filter_async() with AND filter."""
+        docs = [
+            Document(content="Doc 1", meta={"category": "A", "status": "active"}),
+            Document(content="Doc 2", meta={"category": "B", "status": "active"}),
+            Document(content="Doc 3", meta={"category": "A", "status": "inactive"}),
+            Document(content="Doc 4", meta={"category": "A", "status": "active"}),
+        ]
+        document_store.write_documents(docs)
+        assert document_store.count_documents() == 4
+
+        count = await document_store.count_documents_by_filter_async(  # type:ignore[attr-defined]
+            filters={
+                "operator": "AND",
+                "conditions": [
+                    {"field": "meta.category", "operator": "==", "value": "A"},
+                    {"field": "meta.status", "operator": "==", "value": "active"},
+                ],
+            }
+        )
+        assert count == 2
+
+    @staticmethod
+    @pytest.mark.asyncio
+    async def test_count_documents_by_filter_async_no_matches(document_store: DocumentStore):
+        """Test count_documents_by_filter_async() when filter matches no documents."""
+        docs = [Document(content="Doc 1", meta={"category": "A"}), Document(content="Doc 2", meta={"category": "B"})]
+        document_store.write_documents(docs)
+        assert document_store.count_documents() == 2
+
+        count = await document_store.count_documents_by_filter_async(  # type:ignore[attr-defined]
+            filters={"field": "meta.category", "operator": "==", "value": "Z"}
+        )
+        assert count == 0
+
+    @staticmethod
+    @pytest.mark.asyncio
+    async def test_count_documents_by_filter_async_empty_collection(document_store: DocumentStore):
+        """Test count_documents_by_filter_async() on an empty store."""
+        assert document_store.count_documents() == 0
+
+        count = await document_store.count_documents_by_filter_async(  # type:ignore[attr-defined]
+            filters={"field": "meta.category", "operator": "==", "value": "A"}
+        )
+        assert count == 0
+
+
 class CountUniqueMetadataByFilterTest:
     """
     Tests for Document Store count_unique_metadata_by_filter().
@@ -1072,6 +1188,78 @@ def test_count_unique_metadata_by_filter_with_multiple_filters(document_store: D
         assert count == 1
 
 
+class CountUniqueMetadataByFilterAsyncTest:
+    """
+    Tests for Document Store count_unique_metadata_by_filter_async().
+
+    Only mix in for stores that implement count_unique_metadata_by_filter_async.
+    """
+
+    @staticmethod
+    @pytest.mark.asyncio
+    async def test_count_unique_metadata_by_filter_async_all_documents(document_store: DocumentStore):
+        """Test count_unique_metadata_by_filter_async() with no filter returns distinct counts for all docs."""
+        docs = [
+            Document(content="Doc 1", meta={"category": "A", "status": "active", "priority": 1}),
+            Document(content="Doc 2", meta={"category": "B", "status": "active", "priority": 2}),
+            Document(content="Doc 3", meta={"category": "A", "status": "inactive", "priority": 1}),
+            Document(content="Doc 4", meta={"category": "A", "status": "active", "priority": 3}),
+            Document(content="Doc 5", meta={"category": "C", "status": "active", "priority": 2}),
+        ]
+        document_store.write_documents(docs)
+        assert document_store.count_documents() == 5
+
+        counts = await document_store.count_unique_metadata_by_filter_async(  # type:ignore[attr-defined]
+            filters={}, metadata_fields=["category", "status", "priority"]
+        )
+        assert counts["category"] == 3
+        assert counts["status"] == 2
+        assert counts["priority"] == 3
+
+    @staticmethod
+    @pytest.mark.asyncio
+    async def test_count_unique_metadata_by_filter_async_with_filter(document_store: DocumentStore):
+        """Test count_unique_metadata_by_filter_async() with a filter."""
+        docs = [
+            Document(content="Doc 1", meta={"category": "A", "status": "active", "priority": 1}),
+            Document(content="Doc 2", meta={"category": "B", "status": "active", "priority": 2}),
+            Document(content="Doc 3", meta={"category": "A", "status": "inactive", "priority": 1}),
+            Document(content="Doc 4", meta={"category": "A", "status": "active", "priority": 3}),
+        ]
+        document_store.write_documents(docs)
+        assert document_store.count_documents() == 4
+
+        counts = await document_store.count_unique_metadata_by_filter_async(  # type:ignore[attr-defined]
+            filters={"field": "meta.category", "operator": "==", "value": "A"}, metadata_fields=["status", "priority"]
+        )
+        assert counts["status"] == 2
+        assert counts["priority"] == 2
+
+    @staticmethod
+    @pytest.mark.asyncio
+    async def test_count_unique_metadata_by_filter_async_with_multiple_filters(document_store: DocumentStore):
+        """Test counting unique metadata asynchronously with multiple filters."""
+        docs = [
+            Document(content="Doc 1", meta={"category": "A", "year": 2023}),
+            Document(content="Doc 2", meta={"category": "A", "year": 2024}),
+            Document(content="Doc 3", meta={"category": "B", "year": 2023}),
+            Document(content="Doc 4", meta={"category": "B", "year": 2024}),
+        ]
+        document_store.write_documents(docs)
+
+        counts = await document_store.count_unique_metadata_by_filter_async(  # type:ignore[attr-defined]
+            filters={
+                "operator": "AND",
+                "conditions": [
+                    {"field": "meta.category", "operator": "==", "value": "B"},
+                    {"field": "meta.year", "operator": "==", "value": 2023},
+                ],
+            },
+            metadata_fields=["category", "year"],
+        )
+        assert counts == {"category": 1, "year": 1}
+
+
 class GetMetadataFieldsInfoTest:
     """
     Tests for Document Store get_metadata_fields_info().

releasenotes/notes/inmemory-filter-async-apis-10920.yaml

@@ -0,0 +1,5 @@
+---
+features:
+  - |
+    Added async filter helpers to the ``InMemoryDocumentStore``: ``update_by_filter_async()``,
+    ``count_documents_by_filter_async()``, and ``count_unique_metadata_by_filter_async()``.

test/document_stores/test_in_memory.py

@@ -14,14 +14,17 @@
 from haystack.document_stores.errors import DocumentStoreError, DuplicateDocumentError
 from haystack.document_stores.in_memory import InMemoryDocumentStore
 from haystack.testing.document_store import (
+    CountDocumentsByFilterAsyncTest,
     CountDocumentsByFilterTest,
+    CountUniqueMetadataByFilterAsyncTest,
     CountUniqueMetadataByFilterTest,
     DocumentStoreBaseExtendedTests,
     DocumentStoreBaseTests,
     FilterableDocsFixtureMixin,
     GetMetadataFieldMinMaxTest,
     GetMetadataFieldsInfoTest,
     GetMetadataFieldUniqueValuesTest,
+    UpdateByFilterAsyncTest,
 )
 from haystack.testing.document_store_async import (
     CountDocumentsAsyncTest,
@@ -32,10 +35,13 @@
 
 class TestMemoryDocumentStore(
     DocumentStoreBaseExtendedTests,
+    UpdateByFilterAsyncTest,
+    CountDocumentsByFilterAsyncTest,
     CountDocumentsAsyncTest,
     WriteDocumentsAsyncTest,
     DeleteDocumentsAsyncTest,
     CountDocumentsByFilterTest,
+    CountUniqueMetadataByFilterAsyncTest,
     CountUniqueMetadataByFilterTest,
     FilterableDocsFixtureMixin,
     GetMetadataFieldMinMaxTest,