feat: Add apply_filter_policy function (#7902)

* Add apply_filter_policy

* Add release note

haystack/document_stores/types/filter_policy.py

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 from enum import Enum
+from typing import Any, Dict, Optional
 
 
 class FilterPolicy(Enum):
@@ -33,3 +34,28 @@ class FilterPolicy(Enum):
             msg = f"Unknown FilterPolicy type '{filter_policy}'. Supported types are: {list(enum_map.keys())}"
             raise ValueError(msg)
         return policy
+
+
+def apply_filter_policy(
+    filter_policy: FilterPolicy,
+    init_filters: Optional[Dict[str, Any]] = None,
+    runtime_filters: Optional[Dict[str, Any]] = None,
+) -> Optional[Dict[str, Any]]:
+    """
+    Apply the filter policy to the given initial and runtime filters to determine the final set of filters used.
+
+    The function combines or replaces the initial and runtime filters based on the specified filter policy.
+
+    :param filter_policy: The policy to apply when handling the filters. It can be one of the following:
+        - `FilterPolicy.REPLACE`: Runtime filters will replace the initial filters.
+        - `FilterPolicy.MERGE`: Runtime filters will be merged with the initial filters. If there are overlapping keys,
+          values from the runtime filters will overwrite those from the initial filters.
+    :param init_filters: The initial filters set during the initialization of the relevant retriever.
+    :param runtime_filters: The filters provided at runtime, usually during a query operation execution. These filters
+                            can change for each query/retreiver run invocation.
+    :returns: A dictionary containing the resulting filters based on the provided policy.
+    """
+    if filter_policy == FilterPolicy.MERGE and runtime_filters:
+        return {**(init_filters or {}), **runtime_filters}
+    else:
+        return runtime_filters or init_filters

releasenotes/notes/add-apply_filter_policy-function-ae3152e6afe0ca57.yaml

@@ -0,0 +1,4 @@
+---
+enhancements:
+  - |
+    Added the apply_filter_policy function to standardize the application of filter policies across all document store-specific retrievers, allowing for consistent handling of initial and runtime filters based on the chosen policy (replace or merge).

test/document_stores/test_filter_policy.py

@@ -0,0 +1,45 @@
+# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+import pytest
+from typing import Any, Dict, Optional
+from enum import Enum
+
+from haystack.document_stores.types import FilterPolicy
+from haystack.document_stores.types.filter_policy import apply_filter_policy
+
+
+def test_replace_policy_with_both_filters():
+    init_filters = {"status": "active", "category": "news"}
+    runtime_filters = {"author": "John Doe"}
+    result = apply_filter_policy(FilterPolicy.REPLACE, init_filters, runtime_filters)
+    assert result == runtime_filters
+
+
+def test_merge_policy_with_both_filters():
+    init_filters = {"status": "active", "category": "news"}
+    runtime_filters = {"author": "John Doe"}
+    result = apply_filter_policy(FilterPolicy.MERGE, init_filters, runtime_filters)
+    assert result == {"status": "active", "category": "news", "author": "John Doe"}
+
+
+def test_replace_policy_with_only_init_filters():
+    init_filters = {"status": "active", "category": "news"}
+    runtime_filters = None
+    result = apply_filter_policy(FilterPolicy.REPLACE, init_filters, runtime_filters)
+    assert result == init_filters
+
+
+def test_merge_policy_with_only_init_filters():
+    init_filters = {"status": "active", "category": "news"}
+    runtime_filters = None
+    result = apply_filter_policy(FilterPolicy.MERGE, init_filters, runtime_filters)
+    assert result == init_filters
+
+
+def test_merge_policy_with_overlapping_keys():
+    init_filters = {"status": "active", "category": "news"}
+    runtime_filters = {"category": "science", "author": "John Doe"}
+    result = apply_filter_policy(FilterPolicy.MERGE, init_filters, runtime_filters)
+    assert result == {"status": "active", "category": "science", "author": "John Doe"}