Coverage for functions \ flipdare \ search \ doc \ _search_document.py: 77%

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 datetime import datetime 

17from typing import Any, Self, cast 

18from google.cloud.firestore_v1.transforms import Sentinel 

19from pydantic import TypeAdapter, ValidationError 

20from flipdare.app_globals import truncate_string 

21from flipdare.app_log import LOG 

22from flipdare.app_types import SearchDict 

23from flipdare.constants import IS_DEBUG 

24from flipdare.error.data_load_error import DataLoadError 

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

26from flipdare.util.time_util import TypesenseTime 

27 

28__all__ = ["SearchDocument", "TS_C"] 

29 

30 

31# Type alias for timestamp types 

32# NOTE: should only be used internally within SearchDocument and its subclasses 

33type TS_C = int | Sentinel | datetime | str | None 

34 

35_created_at: str = "created_at" 

36_updated_at: str = "updated_at" 

37 

38 

39class SearchDocument[TSchema](ABC): 

40 __slots__ = ("_changed", "_data", "_doc_id", "_original") 

41 

42 # NOTE: Subclasses MUST set this 

43 SCHEMA_CLASS: type[TSchema] # pylint: disable=declare-non-slot 

44 

45 def __init__(self, data: TSchema, doc_id: str | None = None) -> None: 

46 

47 self._changed: set[str] = set() 

48 

49 if doc_id is not None: 

50 self._doc_id = doc_id 

51 

52 # Store data with schema type for proper TypedDict typing 

53 self._data: TSchema = data 

54 self._original: SearchDict = cast("SearchDict", data) 

55 

56 @classmethod 

57 @abstractmethod 

58 def from_payload( 

59 cls, 

60 doc_id: str, 

61 payload: SearchDict, 

62 original: Self | None = None, 

63 ) -> Self: ... 

64 

65 @staticmethod 

66 def parse( 

67 schema_cls: type[TSchema], 

68 payload: SearchDict, 

69 original: SearchDocument[Any] | None = None, 

70 ) -> TSchema: 

71 

72 from flipdare.error.message_format import ValidationErrorMsgFormat 

73 

74 # NOTE: typesense only returns the fields updated 

75 if original is not None: 

76 original_payload = original.payload() 

77 for key in original_payload: 

78 if key not in payload: 

79 payload[key] = original_payload[key] 

80 

81 # Validate with Pydantic 

82 try: 

83 adapter = TypeAdapter(schema_cls) 

84 return adapter.validate_python(payload) 

85 

86 except ValidationError as e: 

87 formatter = ValidationErrorMsgFormat( 

88 class_type=schema_cls, 

89 error=e, 

90 ) 

91 LOG().error(str(formatter)) 

92 

93 raise DataLoadError.model( 

94 class_name=formatter.class_name, 

95 missing_code=formatter.user_error_code, 

96 error=e, 

97 ) from e 

98 

99 # methods should be overriden 

100 @property 

101 @abstractmethod 

102 def uid(self) -> str: ... 

103 

104 @property 

105 @abstractmethod 

106 def obj_type(self) -> SearchObjType: ... 

107 

108 @property 

109 @abstractmethod 

110 def created_at(self) -> int: ... 

111 

112 @property 

113 @abstractmethod 

114 def updated_at(self) -> int: ... 

115 

116 # main 

117 @property 

118 def doc_id(self) -> str | None: 

119 return getattr(self, "_doc_id", None) 

120 

121 @doc_id.setter 

122 def doc_id(self, value: str) -> None: 

123 self._doc_id = value 

124 

125 # helpers 

126 @property 

127 def created_at_dt(self) -> datetime: 

128 dt = TypesenseTime.to_utc_datetime(self.created_at) 

129 if dt is None: 

130 raise ValueError(f"Failed to convert created_at to datetime in {self.debug_str}") 

131 return dt 

132 

133 @property 

134 def updated_at_dt(self) -> datetime: 

135 dt = TypesenseTime.to_utc_datetime(self.updated_at) 

136 if dt is None: 

137 raise ValueError(f"Failed to convert updated_at to datetime in {self.debug_str}") 

138 return dt 

139 

140 # keeping track of changes.. 

141 @property 

142 def has_changed(self) -> bool: 

143 return len(self._changed) > 0 

144 

145 def updates(self) -> SearchDict | None: 

146 if not self.has_changed: 

147 return None 

148 

149 data_dict = cast("SearchDict", self._data) 

150 updates: SearchDict = {} 

151 for key in self._changed: 

152 if key == _created_at: 

153 continue # Never update created_at 

154 

155 updates[key] = data_dict.get(key) 

156 

157 # Always include updated_at when there are changes 

158 if len(updates) > 0: 

159 updates[_updated_at] = data_dict.get(_updated_at) 

160 

161 return updates 

162 

163 def payload(self) -> SearchDict: 

164 """Return complete document payload (all fields).""" 

165 return cast("SearchDict", self._data).copy() 

166 

167 def update_payload(self, payload_values: SearchDict, updated_at: int | None = None) -> None: 

168 has_changed = False 

169 # watch the 'ref' reference.. we are modifying self._data in place. 

170 data_ref = cast("SearchDict", self._data) 

171 

172 for key, value in payload_values.items(): 

173 if key == _created_at: 

174 continue 

175 if key == _updated_at: 

176 if updated_at is None: 

177 updated_at = value 

178 continue 

179 

180 old_value = data_ref.get(key) 

181 if (old_value is None and value is None) or old_value == value: 

182 if IS_DEBUG: 

183 LOG().debug( 

184 f'Skipping update for "{key}" since value is unchanged {old_value}' 

185 ) 

186 continue 

187 

188 data_ref[key] = value 

189 if key not in self._changed: 

190 has_changed = True 

191 self._changed.add(key) 

192 

193 if not has_changed: 

194 return 

195 

196 if updated_at is not None: 

197 data_ref[_updated_at] = updated_at 

198 

199 def payload_equal(self, other: SearchDocument[Any]) -> bool: 

200 # we dont override __eq__ because the doc_id may be different for same data 

201 other_payload = other.payload() 

202 self_payload = self.payload() 

203 

204 return self_payload == other_payload 

205 

206 def _get_field(self, key: str) -> Any: 

207 """Get field value from data dict.""" 

208 # Cast to dict for .get() operation 

209 return cast("SearchDict", self._data).get(key) 

210 

211 def _set_field(self, key: str, value: Any, is_update: bool = False) -> None: 

212 """Set field value in data dict and track changes.""" 

213 data_ref = cast("SearchDict", self._data) 

214 old_value = data_ref.get(key) 

215 if (old_value is None and value is None) or old_value == value: 

216 if IS_DEBUG: 

217 LOG().debug(f'Skipping update for "{key}" since value is unchanged {old_value}') 

218 return 

219 

220 data_ref[key] = value 

221 

222 if not is_update: 

223 return 

224 

225 # Track change and update timestamp 

226 data_ref[_updated_at] = TypesenseTime.server_timestamp() 

227 self._changed.add(key) 

228 

229 @property 

230 def debug_str(self) -> str: 

231 msg = self.doc_id if self.doc_id is not None else "NoDocID" 

232 payload = self.payload() 

233 for key in sorted(payload.keys()): 

234 msg += f"[{key}: {payload[key]}]," 

235 

236 return truncate_string(msg, 124) 

237 

238 # Backward compatibility methods (deprecated) 

239 def _get_attr(self, key: str) -> Any: 

240 """Deprecated: Use _get_field instead.""" 

241 return self._get_field(key) 

242 

243 def _set_attr(self, key: str, value: Any, is_update: bool = False) -> None: 

244 """Deprecated: Use _set_field instead.""" 

245 self._set_field(key, value, is_update)