Coverage for functions \ flipdare \ search \ core \ query_builder.py: 97%

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 

13from __future__ import annotations 

14 

15from abc import ABC, abstractmethod 

16from typing import Any, Self, override 

17 

18from flipdare.app_log import LOG 

19from flipdare.constants import IS_DEBUG 

20from flipdare.generated.schema.search.friend_document_schema import FriendDocumentKey 

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

22from flipdare.generated.schema.search.search_request_schema import SearchRequestSchema 

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

24from flipdare.generated.shared.search.search_obj_type import SearchObjType 

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

26from flipdare.search.core.filter._complex_filter import Relationship 

27from flipdare.search.core.filter.friend_filter import ComplexFriendFilter 

28from flipdare.search.core.filter.general_filter import ComplexGeneralFilter 

29from flipdare.search.core.query.friend_query import FriendQuery 

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

31from flipdare.search.core.query_by import FriendQueryBy, GeneralQueryBy 

32from flipdare.search.core.query_options import QueryOptions 

33 

34__all__ = ["QueryBuilder", "GeneralQueryBuilder", "FriendQueryBuilder", "QueryBuilderFactory"] 

35 

36type QueryBuilderType = GeneralQuery | FriendQuery 

37 

38 

39class QueryBuilderFactory: 

40 @staticmethod 

41 def create( 

42 request: SearchRequestSchema, 

43 uid: str | None = None, 

44 ) -> QueryBuilder[QueryBuilderType]: 

45 collection = request["collection"] 

46 match collection: 

47 case SearchCollections.GENERAL: 

48 return GeneralQueryBuilder.from_request(request, uid) 

49 case SearchCollections.FRIEND: 

50 return FriendQueryBuilder.from_request(request, uid) 

51 

52 

53class QueryBuilder[T: GeneralQuery | FriendQuery](ABC): 

54 query_str: str 

55 sort_type: SearchSortType 

56 page: int 

57 collection: SearchCollections 

58 relationship: Relationship | None 

59 obj_types: list[SearchObjType] | None 

60 auto_complete: bool 

61 

62 def __init__( 

63 self, 

64 collection: SearchCollections, 

65 query_str: str, 

66 page: int, 

67 sort_type: SearchSortType, 

68 relationship: Relationship | None = None, 

69 obj_types: list[SearchObjType] | None = None, 

70 auto_complete: bool = False, 

71 ) -> None: 

72 self.query_str = query_str 

73 self.page = page 

74 self.sort_type = sort_type 

75 self.collection = collection 

76 

77 self.relationship = relationship 

78 self.auto_complete = auto_complete 

79 

80 self.obj_types = obj_types 

81 

82 @classmethod 

83 def from_request(cls, request: SearchRequestSchema, uid: str | None = None) -> Self: 

84 relationship: Relationship | None = None 

85 relation_type = request.get("relation_type") 

86 if uid is not None and relation_type is not None: 

87 relationship = Relationship(uid=uid, relation_type=relation_type) 

88 

89 return cls( 

90 collection=request["collection"], 

91 query_str=request["query"], 

92 page=request["page_num"], 

93 sort_type=request["sort_type"], 

94 relationship=relationship, 

95 obj_types=request.get("obj_types"), 

96 auto_complete=request.get("auto_complete", False), 

97 ) 

98 

99 @abstractmethod 

100 def build(self) -> T: ... 

101 

102 

103class GeneralQueryBuilder(QueryBuilder[GeneralQuery]): 

104 def __init__( 

105 self, 

106 **kwargs: Any, 

107 ) -> None: 

108 kwargs.pop("collection", None) # Ensure collection is not passed to super 

109 

110 super().__init__(collection=SearchCollections.GENERAL, **kwargs) 

111 

112 @override 

113 def build(self) -> GeneralQuery: 

114 obj_types = self.obj_types 

115 relationship = self.relationship 

116 auto_complete = self.auto_complete 

117 query = self.query_str 

118 sort_type = self.sort_type 

119 page = self.page 

120 

121 if IS_DEBUG: 

122 msg = f"Building relationship={relationship} obj_types={obj_types}, auto_complete={auto_complete}" 

123 LOG().debug(msg) 

124 

125 filter_by: ComplexGeneralFilter | None = None 

126 if relationship is None: 

127 filter_by = ComplexGeneralFilter(obj_types=obj_types) 

128 else: 

129 relation_type = relationship.relation_type 

130 uid = relationship.uid 

131 if relation_type.is_mine: 

132 # we dont pass the relationship, 

133 # we just filter for the user's own UID since that's what "mine" means. 

134 filter_by = ComplexGeneralFilter( 

135 obj_types=obj_types, 

136 filters={GeneralDocumentKey.UID: uid}, 

137 ) 

138 else: 

139 filter_by = ComplexGeneralFilter( 

140 obj_types=obj_types, 

141 relationship=relationship, 

142 ) 

143 

144 options = QueryOptions.general() if not auto_complete else QueryOptions.auto() 

145 query_by = GeneralQueryBy([GeneralDocumentKey.KEYWORDS, GeneralDocumentKey.TAGS]) 

146 return GeneralQuery.query( 

147 query_str=query, 

148 query_by=query_by, 

149 filter_by=filter_by, 

150 query_options=options, 

151 sort_type=sort_type, 

152 page=page, 

153 ) 

154 

155 

156class FriendQueryBuilder(QueryBuilder[FriendQuery]): 

157 def __init__( 

158 self, 

159 uid: str, 

160 **kwargs: Any, 

161 ) -> None: 

162 kwargs.pop("collection", None) # Ensure collection is not passed to super 

163 

164 super().__init__(collection=SearchCollections.FRIEND, **kwargs) 

165 # friend requires a uid even without a relationship, 

166 # since we need to know whose friends to search. 

167 self._uid = uid 

168 

169 @override 

170 def build(self) -> FriendQuery: 

171 relationship = self.relationship 

172 auto_complete = self.auto_complete 

173 query = self.query_str 

174 sort_type = self.sort_type 

175 page = self.page 

176 

177 if relationship is None or relationship.relation_type.is_mine: 

178 filter_by = ComplexFriendFilter(uid=self._uid) 

179 else: # relationship.relation_type.is_friends 

180 filter_by = ComplexFriendFilter( 

181 uid=self._uid, 

182 filters={FriendDocumentKey.FRIEND_UID: self._uid}, 

183 exclude_uid=True, 

184 ) 

185 

186 if IS_DEBUG: 

187 msg = f"Building query for relationship={relationship} auto_complete={auto_complete}" 

188 LOG().debug(msg) 

189 

190 options = QueryOptions.friend() if not auto_complete else QueryOptions.auto() 

191 query_by = FriendQueryBy([FriendDocumentKey.KEYWORDS, FriendDocumentKey.FRIEND_KEYWORDS]) 

192 return FriendQuery.query( 

193 query_str=query, 

194 query_by=query_by, 

195 filter_by=filter_by, 

196 query_options=options, 

197 sort_type=sort_type, 

198 page=page, 

199 )