Coverage for functions \ flipdare \ search \ db \ app_friend_search.py: 69%
128 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-05-08 12:22 +1000
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#
14from typing import Any, Self, override
15import typesense
16from typesense.types.document import SearchParameters
18from flipdare.app_env import get_app_environment
19from flipdare.app_log import LOG
20from flipdare.constants import (
21 IS_DEBUG,
22 MAX_SEARCH_FRIENDS_JOIN,
23 MAX_SEARCH_RESULTS_PER_PAGE,
24 TYPESENSE_HARD_PER_PAGE_LIMIT,
25)
26from flipdare.error.app_error import CodePathError
27from flipdare.generated.schema.search.friend_document_schema import FriendDocumentKey
28from flipdare.generated.shared.search.search_collections import SearchCollections
29from flipdare.generated.shared.search.search_sort_type import SearchSortType
30from flipdare.search.core.filter.friend_filter import SimpleFriendFilter
31from flipdare.search.core.query.friend_query import FriendQuery
32from flipdare.search.core.query_by import QueryBy
33from flipdare.search.db._app_search import AppSearch
34from flipdare.search.doc.friend_document import FriendDocument
36__all__ = ["AppFriendSearch"]
39class AppFriendSearch(AppSearch[FriendDocument, FriendDocumentKey]):
41 def __init__(
42 self,
43 client: typesense.Client,
44 per_page: int = MAX_SEARCH_RESULTS_PER_PAGE,
45 ) -> None:
46 super().__init__(
47 SearchCollections.FRIEND,
48 client,
49 FriendDocument,
50 per_page=per_page,
51 )
53 @classmethod
54 def custom(
55 cls,
56 client: typesense.Client,
57 collection: SearchCollections,
58 per_page: int = MAX_SEARCH_RESULTS_PER_PAGE,
59 ) -> Self:
60 """Create a custom instance with a different collection (test use only)."""
61 if get_app_environment().in_cloud:
62 msg = f"Custom collections are NOT allowed in the cloud.\n{get_app_environment().debug_str()}"
63 raise RuntimeError(msg)
65 # Create instance without calling __init__
66 instance = object.__new__(cls)
68 # Call parent __init__ directly with custom collection
69 AppSearch.__init__(
70 instance,
71 collection,
72 client,
73 FriendDocument,
74 per_page=per_page,
75 )
77 return instance
79 @override
80 def get_user(self, uid: str) -> list[FriendDocument] | None:
81 return self._get_all(uid)
83 @override
84 def delete_user(self, uid: str) -> None:
85 docs = self._get_all(uid)
86 if docs is None:
87 msg = f"No documents found for delete_user with uid={uid} in {self.collection.value}"
88 LOG().info(msg)
89 return
91 doc_ids = [doc.doc_id for doc in docs if doc.doc_id is not None]
92 self.batch_delete(doc_ids)
94 @override
95 def get_user_type(
96 self,
97 uid: str,
98 identifier: str, # friend_id
99 ) -> FriendDocument | None:
100 results = self._get(uid, FriendDocumentKey.FRIEND_UID, identifier)
101 if results is None or len(results) == 0:
102 return None
103 if len(results) > 1:
104 debug_str = f"{self.collection.value}/{FriendDocumentKey.FRIEND_UID}={identifier}"
105 msg = f"Multiple documents found for {debug_str}, returning first."
106 LOG().warning(msg)
108 return results[0]
110 @override
111 def delete_user_type(
112 self,
113 uid: str,
114 identifier: str, # friend_id
115 delete_reciprocal: bool = True, # whether to delete the reciprocal friend document (default True)
116 **kwargs: Any, # for future extensibility if we want to add additional filters for delete_user_type
117 ) -> FriendDocument | None:
118 deleted_doc = self._delete(uid, FriendDocumentKey.FRIEND_UID, identifier)
119 if not delete_reciprocal:
120 if IS_DEBUG:
121 msg = f"Deleted document (no reciprocal) for uid={uid} and friend_uid={identifier}"
122 LOG().debug(msg)
124 return deleted_doc
126 recip_deleted_doc = self._delete(identifier, FriendDocumentKey.FRIEND_UID, uid)
127 if deleted_doc is None and recip_deleted_doc is None:
128 if IS_DEBUG:
129 msg = f"No friend documents for uid={uid} and friend_uid={identifier}"
130 LOG().debug(msg)
131 return None
133 if recip_deleted_doc is None:
134 msg = f"No reciprocal friend document found for friend_uid={identifier} when deleting uid={uid}"
135 LOG().warning(msg)
136 return deleted_doc
137 else:
138 # deleted_doc is None
139 msg = f"Reciprocal friend document deleted for friend_uid={identifier} when deleting uid={uid}"
140 LOG().info(msg)
141 return recip_deleted_doc
143 def get_friend_uids(self, uid: str) -> list[str]:
144 """
145 Fetch all friend_uids for a given user, handling Typesense's
146 250-per-page limit through pagination.
147 """
148 all_friend_uids: list[str] = []
149 page: int = 1
150 per_page: int = TYPESENSE_HARD_PER_PAGE_LIMIT # Typesense maximum
151 # the check is >, so we DONT Add 1 (ai keeps recommending adding 1, but is WRONG.)
152 max_pages: int = MAX_SEARCH_FRIENDS_JOIN // per_page
154 while len(all_friend_uids) < MAX_SEARCH_FRIENDS_JOIN:
155 params: SearchParameters = {
156 "q": "*",
157 "query_by": "uid",
158 "filter_by": f"uid:={uid}",
159 "per_page": per_page,
160 "page": page,
161 }
162 try:
163 # Execute the internal search on the friend collection
164 payload = self.search(params)
165 hits = payload.hits
167 if len(hits) == 0:
168 break
170 # Extract friend_uids from the current page
171 all_friend_uids.extend(str(hit.document["friend_uid"]) for hit in hits)
173 # Check if there are more pages to fetch
174 if len(hits) < per_page:
175 break
177 page += 1
179 # Safety break for project-specific limits
180 if page > max_pages:
181 break
183 except Exception as e:
184 msg = f"Failed to fetch friends for uid {uid} at page {page}: {e}"
185 LOG().error(msg)
186 break
188 return all_friend_uids[:MAX_SEARCH_FRIENDS_JOIN] # Enforce the maximum limit
190 #
191 # internal helpers
192 #
194 def _get(
195 self,
196 uid: str,
197 field: FriendDocumentKey,
198 value: str,
199 sort_by_newest: bool = False,
200 ) -> list[FriendDocument] | None:
201 filter_by = SimpleFriendFilter(uid=uid, filters={field: value})
202 query = FriendQuery.query(
203 query_by=QueryBy[FriendDocumentKey](field),
204 query_str=value,
205 filter_by=filter_by,
206 sort_type=SearchSortType.NEWEST if sort_by_newest else SearchSortType.OLDEST,
207 )
209 op_str = f"get:uid={uid},{field}={value}"
210 if IS_DEBUG:
211 LOG().debug(
212 f"Searching for Friend document in {self.collection.value}/{op_str} with {query}",
213 )
215 result = self.search(query.search_params)
216 return self.process_result(op_str, result)
218 def _get_all(
219 self,
220 uid: str,
221 sort_by_newest: bool = False,
222 ) -> list[FriendDocument] | None:
223 sort_type = SearchSortType.NEWEST if sort_by_newest else SearchSortType.OLDEST
224 query_by = QueryBy[FriendDocumentKey](FriendDocumentKey.UID)
225 query = FriendQuery.query(
226 query_str="*",
227 query_by=query_by,
228 sort_type=sort_type,
229 )
231 op_str = f"get:uid={uid}"
232 if IS_DEBUG:
233 LOG().debug(
234 f"Searching for Friend document in {self.collection.value}/{op_str} with {query}",
235 )
237 result = self.search(query.search_params)
238 return self.process_result(op_str, result)
240 def _delete(
241 self,
242 uid: str,
243 field: FriendDocumentKey,
244 value: str,
245 ) -> FriendDocument | None:
246 col = self.collection
248 results = self._get(uid, field, value)
249 if results is None:
250 if IS_DEBUG:
251 LOG().debug(f"No documents found for delete in {col.value} with {field}={value}")
252 return None
254 if len(results) > 1:
255 msg = f"Use batch_delete to delete multiple documents for {col.value}:{field}={value}"
256 LOG().error(msg)
257 raise CodePathError(message=msg)
259 doc_id = results[0].doc_id
260 if doc_id is None:
261 # should never get here since doc_id should always be present, but handle just in case
262 LOG().error(f"Document ID is None for {col.value}:{field}={value}, cannot delete.")
263 return None
265 if IS_DEBUG:
266 LOG().debug(f"Deleting document with id={doc_id} for {col.value}:{field}={value}")
268 return self.delete(doc_id)