Coverage for haystack/components/caching/cache_checker.py: 100%

1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai> 

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5from typing import Any 

6 

7from haystack import Document, component, default_from_dict, default_to_dict 

8from haystack.document_stores.types import DocumentStore 

9 

10 

11@component 

12class CacheChecker: 

13 """ 

14 Checks for the presence of documents in a Document Store based on a specified field in each document's metadata. 

15 

16 If matching documents are found, they are returned as "hits". If not found in the cache, the items 

17 are returned as "misses". 

18 

19 ### Usage example 

20 

21 ```python 

22 from haystack import Document 

23 from haystack.document_stores.in_memory import InMemoryDocumentStore 

24 from haystack.components.caching.cache_checker import CacheChecker 

25 

26 docstore = InMemoryDocumentStore() 

27 documents = [ 

28 Document(content="doc1", meta={"url": "https://example.com/1"}), 

29 Document(content="doc2", meta={"url": "https://example.com/2"}), 

30 Document(content="doc3", meta={"url": "https://example.com/1"}), 

31 Document(content="doc4", meta={"url": "https://example.com/2"}), 

32 ] 

33 docstore.write_documents(documents) 

34 checker = CacheChecker(docstore, cache_field="url") 

35 results = checker.run(items=["https://example.com/1", "https://example.com/5"]) 

36 assert results == {"hits": [documents[0], documents[2]], "misses": ["https://example.com/5"]} 

37 ``` 

38 """ 

39 

40 def __init__(self, document_store: DocumentStore, cache_field: str) -> None: 

41 """ 

42 Creates a CacheChecker component. 

43 

44 :param document_store: 

45 Document Store to check for the presence of specific documents. 

46 :param cache_field: 

47 Name of the document's metadata field 

48 to check for cache hits. 

49 """ 

50 self.document_store = document_store 

51 self.cache_field = cache_field 

52 

53 def to_dict(self) -> dict[str, Any]: 

54 """ 

55 Serializes the component to a dictionary. 

56 

57 :returns: 

58 Dictionary with serialized data. 

59 """ 

60 return default_to_dict(self, document_store=self.document_store, cache_field=self.cache_field) 

61 

62 @classmethod 

63 def from_dict(cls, data: dict[str, Any]) -> "CacheChecker": 

64 """ 

65 Deserializes the component from a dictionary. 

66 

67 :param data: 

68 Dictionary to deserialize from. 

69 :returns: 

70 Deserialized component. 

71 """ 

72 return default_from_dict(cls, data) 

73 

74 @component.output_types(hits=list[Document], misses=list) 

75 def run(self, items: list[Any]) -> dict[str, Any]: 

76 """ 

77 Checks if any document associated with the specified cache field is already present in the store. 

78 

79 :param items: 

80 Values to be checked against the cache field. 

81 :return: 

82 A dictionary with two keys: 

83 - `hits` - Documents that matched with at least one of the items. 

84 - `misses` - Items that were not present in any documents. 

85 """ 

86 found_documents = [] 

87 misses = [] 

88 

89 for item in items: 

90 filters = {"field": self.cache_field, "operator": "==", "value": item} 

91 found = self.document_store.filter_documents(filters=filters) 

92 if found: 

93 found_documents.extend(found) 

94 else: 

95 misses.append(item) 

96 return {"hits": found_documents, "misses": misses} 

97 

98 @component.output_types(hits=list[Document], misses=list) 

99 async def run_async(self, items: list[Any]) -> dict[str, Any]: 

100 """ 

101 Asynchronously checks if any document associated with the specified cache field is already present in the store. 

102 

103 :param items: 

104 Values to be checked against the cache field. 

105 :return: 

106 A dictionary with two keys: 

107 - `hits` - Documents that matched with at least one of the items. 

108 - `misses` - Items that were not present in any documents. 

109 """ 

110 found_documents = [] 

111 misses = [] 

112 

113 if not hasattr(self.document_store, "filter_documents_async"): 

114 raise TypeError(f"Document store {type(self.document_store).__name__} does not provide async support.") 

115 

116 for item in items: 

117 filters = {"field": self.cache_field, "operator": "==", "value": item} 

118 found = await self.document_store.filter_documents_async(filters=filters) 

119 if found: 

120 found_documents.extend(found) 

121 else: 

122 misses.append(item) 

123 return {"hits": found_documents, "misses": misses} 

124 

125 def close(self) -> None: 

126 """ 

127 Release the synchronous resources of the underlying Document Store. 

128 """ 

129 if hasattr(self.document_store, "close"): 

130 self.document_store.close() 

131 

132 async def close_async(self) -> None: 

133 """ 

134 Release the asynchronous resources of the underlying Document Store. 

135 """ 

136 if hasattr(self.document_store, "close_async"): 

137 await self.document_store.close_async()