Coverage for functions \ flipdare \ search \ db \ app_general_search.py: 55%

1#!/usr/bin/env python 

2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved. 

3# 

4# This file is part of Flipdare's proprietary software and contains 

5# confidential and copyrighted material. Unauthorised copying, 

6# modification, distribution, or use of this file is strictly 

7# prohibited without prior written permission from Flipdare Pty Ltd. 

8# 

9# This software includes third-party components licensed under MIT, 

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details. 

11# 

12 

13 

14from typing import Any, Self, override 

15import typesense 

16 

17from flipdare.app_env import get_app_environment 

18from flipdare.app_log import LOG 

19from flipdare.constants import IS_DEBUG, MAX_SEARCH_RESULTS_PER_PAGE 

20from flipdare.error.app_error import CodePathError 

21from flipdare.generated.schema.search.general_document_schema import GeneralDocumentKey 

22from flipdare.generated.shared.search.search_collections import SearchCollections 

23from flipdare.generated.shared.search.search_sort_type import SearchSortType 

24from flipdare.search.core.filter.general_filter import GeneralFilterDict, SimpleGeneralFilter 

25from flipdare.search.core.query.general_query import GeneralQuery 

26from flipdare.search.core.query_by import QueryBy 

27from flipdare.search.db._app_search import AppSearch 

28from flipdare.search.doc.general_document import GeneralDocument 

29 

30 

31class AppGeneralSearch(AppSearch[GeneralDocument, GeneralDocumentKey]): 

32 def __init__( 

33 self, 

34 client: typesense.Client, 

35 per_page: int = MAX_SEARCH_RESULTS_PER_PAGE, 

36 ) -> "None": 

37 super().__init__( 

38 SearchCollections.GENERAL, 

39 client, 

40 GeneralDocument, 

41 per_page=per_page, 

42 ) 

43 

44 @classmethod 

45 def custom( 

46 cls, 

47 client: typesense.Client, 

48 collection: SearchCollections, 

49 per_page: int = MAX_SEARCH_RESULTS_PER_PAGE, 

50 ) -> Self: 

51 """Create a custom instance with a different collection (test use only).""" 

52 if get_app_environment().in_cloud: 

53 msg = f"Custom collections are NOT allowed in the cloud.\n{get_app_environment().debug_str()}" 

54 raise RuntimeError(msg) 

55 

56 # Create instance without calling __init__ 

57 instance = object.__new__(cls) 

58 

59 # Call parent __init__ directly with custom collection 

60 AppSearch.__init__( 

61 instance, 

62 collection, 

63 client, 

64 GeneralDocument, 

65 per_page=per_page, 

66 ) 

67 

68 return instance 

69 

70 @override 

71 def get_user(self, uid: str) -> list[GeneralDocument] | None: 

72 return self._get_all(uid) 

73 

74 @override 

75 def delete_user(self, uid: str) -> None: 

76 docs = self._get_all(uid) 

77 if docs is None: 

78 msg = f"No documents found for delete_user with uid={uid} in {self.collection.value}" 

79 LOG().info(msg) 

80 return 

81 

82 doc_ids = [doc.doc_id for doc in docs if doc.doc_id is not None] 

83 self.batch_delete(doc_ids) 

84 

85 @override 

86 def get_user_type( 

87 self, 

88 uid: str, 

89 identifier: str, # obj_id 

90 ) -> GeneralDocument | None: 

91 results = self._get(uid, GeneralDocumentKey.OBJ_ID, identifier) 

92 if results is None or len(results) == 0: 

93 return None 

94 if len(results) > 1: 

95 debug_str = f"{self.collection.value}/{GeneralDocumentKey.OBJ_ID}={identifier}" 

96 msg = f"Multiple documents found for {debug_str}, returning first." 

97 LOG().warning(msg) 

98 

99 return results[0] 

100 

101 @override 

102 def delete_user_type( 

103 self, 

104 uid: str, 

105 identifier: str, # obj_id 

106 **kwargs: Any, # for future extensibility if we want to add additional filters for delete_user_type 

107 ) -> GeneralDocument | None: 

108 if IS_DEBUG: 

109 LOG().debug(f"Deleting document for uid={uid}, identifier={identifier}") 

110 

111 return self._delete(uid, GeneralDocumentKey.OBJ_ID, identifier) 

112 

113 # 

114 # internal helpers 

115 # 

116 

117 def _get( 

118 self, 

119 uid: str, 

120 field: GeneralDocumentKey, 

121 value: str, 

122 sort_by_newest: bool = False, 

123 ) -> list[GeneralDocument] | None: 

124 filters: GeneralFilterDict = {field: value, GeneralDocumentKey.UID: uid} 

125 filter_by = SimpleGeneralFilter(filters=filters) 

126 query = GeneralQuery.query( 

127 query_by=QueryBy[GeneralDocumentKey](field), 

128 query_str=value, 

129 filter_by=filter_by, 

130 sort_type=SearchSortType.NEWEST if sort_by_newest else SearchSortType.OLDEST, 

131 ) 

132 

133 op_str = f"get:uid={uid},{field}={value}" 

134 if IS_DEBUG: 

135 LOG().debug( 

136 f"Searching for general document in {self.collection.value}/{op_str} with {query}", 

137 ) 

138 

139 result = self.search(query.search_params) 

140 return self.process_result(op_str, result) 

141 

142 def _get_all( 

143 self, 

144 uid: str, 

145 sort_by_newest: bool = False, 

146 ) -> list[GeneralDocument] | None: 

147 sort_type = SearchSortType.NEWEST if sort_by_newest else SearchSortType.OLDEST 

148 filter_by: SimpleGeneralFilter = SimpleGeneralFilter( 

149 filters={ 

150 GeneralDocumentKey.UID: uid, 

151 }, 

152 ) 

153 

154 query_by = QueryBy[GeneralDocumentKey](GeneralDocumentKey.UID) 

155 query = GeneralQuery.query( 

156 query_by=query_by, 

157 query_str="*", 

158 filter_by=filter_by, 

159 sort_type=sort_type, 

160 ) 

161 

162 op_str = f"get:uid={uid}" 

163 if IS_DEBUG: 

164 LOG().debug( 

165 f"Searching for general document in {self.collection.value}/{op_str} with {query}", 

166 ) 

167 

168 result = self.search(query.search_params) 

169 return self.process_result(op_str, result) 

170 

171 def _delete( 

172 self, 

173 uid: str, 

174 field: GeneralDocumentKey, 

175 value: str, 

176 ) -> GeneralDocument | None: 

177 col = self.collection 

178 

179 results = self._get(uid, field, value) 

180 if results is None: 

181 if IS_DEBUG: 

182 LOG().debug("Cant delete doc, doesnt exist: uid={uid} {field}={value}") 

183 return None 

184 

185 if len(results) > 1: 

186 msg = f"Use batch_delete to delete multiple documents for {col.value}:{field}={value}" 

187 LOG().error(msg) 

188 raise CodePathError(message=msg) 

189 

190 doc_id = results[0].doc_id 

191 if doc_id is None: 

192 # should never get here since doc_id should always be present, but handle just in case 

193 LOG().error(f"Document ID is None for {col.value}:{field}={value}, cannot delete.") 

194 return None 

195 

196 return self.delete(doc_id)