Coverage for functions \ flipdare \ search \ core \ query \ search_query.py: 98%

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 dataclasses import dataclass 

15from enum import StrEnum 

16from typing import TYPE_CHECKING, Self, override 

17 

18from typesense.types.document import SearchParameters 

19 

20from flipdare.app_globals import is_text_present, sanitize_input 

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

22from flipdare.search.core.filter.filter_guards import FilterGuards, QueryFilterType 

23from flipdare.search.core.query_by import QueryBy 

24from flipdare.search.core.query_options import QueryOptions 

25 

26if TYPE_CHECKING: 

27 

28 from flipdare.search.db.app_friend_search import AppFriendSearch 

29 

30 

31@dataclass 

32class QueryData[T: StrEnum]: 

33 query_str: str 

34 query_by: QueryBy[T] 

35 filter_by: QueryFilterType[T] | None = None 

36 sort_type: SearchSortType = SearchSortType.STANDARD 

37 page: int = 1 

38 

39 

40class SearchQuery[T: StrEnum]: 

41 __slots__ = [ 

42 "_query_data", 

43 "_query_options", 

44 ] 

45 

46 DEFAULT_OPTIONS: QueryOptions = QueryOptions.standard() 

47 

48 def __init__( 

49 self, 

50 query_data: QueryData[T], 

51 query_options: QueryOptions | None = None, 

52 ) -> None: 

53 self._query_data = query_data 

54 self._query_options = query_options or self.DEFAULT_OPTIONS 

55 

56 @classmethod 

57 def query( 

58 cls, 

59 query_str: str, 

60 query_by: QueryBy[T], 

61 filter_by: QueryFilterType[T] | None = None, 

62 sort_type: SearchSortType = SearchSortType.STANDARD, 

63 query_options: QueryOptions | None = None, 

64 page: int = 1, 

65 ) -> Self: 

66 """ 

67 General query with flexible field matching. 

68 Use uid for known/authenticated context. 

69 

70 """ 

71 if not is_text_present(query_str): 

72 query_str = "*" 

73 

74 return cls( 

75 query_data=QueryData( 

76 query_str=query_str, 

77 query_by=query_by, 

78 filter_by=filter_by, 

79 sort_type=sort_type, 

80 page=page, 

81 ), 

82 query_options=query_options, 

83 ) 

84 

85 @classmethod 

86 def auto_complete( 

87 cls, 

88 query_str: str, 

89 query_by: QueryBy[T], 

90 filter_by: QueryFilterType[T] | None = None, 

91 sort_type: SearchSortType = SearchSortType.STANDARD, 

92 query_options: QueryOptions | None = None, 

93 page: int = 1, 

94 ) -> Self: 

95 return cls( 

96 query_data=QueryData( 

97 query_str=query_str, 

98 query_by=query_by, 

99 filter_by=filter_by, 

100 sort_type=sort_type, 

101 page=page, 

102 ), 

103 query_options=query_options or QueryOptions.auto(), 

104 ) 

105 

106 @classmethod 

107 def or_( 

108 cls, 

109 field: T, 

110 or_field: T, 

111 value: str, 

112 filter_by: QueryFilterType[T] | None = None, 

113 sort_type: SearchSortType = SearchSortType.STANDARD, 

114 query_options: QueryOptions | None = None, 

115 page: int = 1, 

116 ) -> Self: 

117 """ 

118 Creates a query that matches documents where the specified field 

119 equals the given value. 

120 

121 """ 

122 return cls( 

123 query_data=QueryData[T]( 

124 query_str=value, 

125 query_by=QueryBy[T]([field, or_field]), 

126 filter_by=filter_by, 

127 sort_type=sort_type, 

128 page=page, 

129 ), 

130 query_options=query_options, 

131 ) 

132 

133 @staticmethod 

134 def sanitize(value: str) -> str: 

135 """Sanitizes the input value for search queries.""" 

136 # strip, if spaces in between words wrap in quotes 

137 value = sanitize_input(value, should_strip=True) 

138 if " " in value: 

139 value = f'"{value}"' 

140 return value 

141 

142 @property 

143 def data(self) -> QueryData[T]: 

144 return self._query_data 

145 

146 @property 

147 def options(self) -> QueryOptions: 

148 return self._query_options 

149 

150 @property 

151 def limit(self) -> int: 

152 return self.options.limit 

153 

154 @property 

155 def query_str(self) -> str: 

156 return self.sanitize(self.data.query_str) 

157 

158 @property 

159 def query_by(self) -> str: 

160 return self.data.query_by.create() 

161 

162 @property 

163 def filter_by(self) -> str | None: 

164 if self.data.filter_by is None: 

165 return None 

166 return self.data.filter_by.filters 

167 

168 @property 

169 def sort_by(self) -> list[str]: 

170 return self.data.sort_type.sort_values 

171 

172 @property 

173 def num_typos(self) -> int: 

174 return self.options.num_typos 

175 

176 @property 

177 def is_auto_complete(self) -> bool: 

178 return self.options.auto_complete 

179 

180 @property 

181 def per_page(self) -> int: 

182 return self.options.per_page 

183 

184 @property 

185 def page(self) -> int: 

186 return self.data.page 

187 

188 def prepare(self, friend_search: "AppFriendSearch") -> None: 

189 """ 

190 Public entry point to trigger the filter's internal _prepare logic. 

191 Call this before accessing search_params. 

192 """ 

193 if not self.data.filter_by or not FilterGuards.is_complex_filter( 

194 self.data.filter_by, 

195 ): 

196 return 

197 

198 # This satisfies the BaseFilter requirement: self._prepare_called = True 

199 self.data.filter_by.prepare(friend_search) 

200 

201 @property 

202 def search_params(self) -> SearchParameters: 

203 """Constructs the search query dictionary.""" 

204 params: SearchParameters = { 

205 "q": self.query_str, 

206 "query_by": self.query_by, 

207 "per_page": self.per_page, 

208 "page": self.page, 

209 "exhaustive_search": self.options.exhaustive_search, 

210 "limit_hits": self.limit, 

211 "prefix": self.options.prefix, 

212 } 

213 

214 if (num_typos := self.num_typos) > 0: 

215 params["num_typos"] = num_typos 

216 

217 if self.is_auto_complete: 

218 params["prefix"] = True 

219 

220 if (fb := self.data.filter_by) is not None and (filter_str := fb.filters): 

221 # This call will raise RuntimeError if .prepare() was skipped 

222 params["filter_by"] = filter_str 

223 

224 if sort := self.sort_by: 

225 params["sort_by"] = ",".join(sort) 

226 

227 params["highlight_fields"] = self.query_by 

228 params["highlight_affix_num_tokens"] = self.options.highlight_affix_num_tokens 

229 params["drop_tokens_threshold"] = self.options.drop_tokens_threshold 

230 params["typo_tokens_threshold"] = self.options.typo_tokens_threshold 

231 

232 return params 

233 

234 @override 

235 def __str__(self) -> str: 

236 return ( 

237 f"AppBaseQuery(query_str='{self.query_str}', query_by='{self.query_by}', " 

238 f"filter_by={self.filter_by}, page={self.page}, sort_by={self.sort_by}, " 

239 f"options={self._query_options}search_params=\n{self.search_params})" 

240 ) 

241 

242 @override 

243 def __repr__(self) -> str: 

244 return self.__str__()