Coverage for functions \ flipdare \ firestore \ core \ app_base_model.py: 91%

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 typing import Any, Self 

16 

17from pydantic import BaseModel 

18from pydantic.fields import FieldInfo 

19from pydantic_core import ValidationError 

20 

21from flipdare.app_log import LOG 

22 

23 

24class AppBaseModel(BaseModel): 

25 def __init__(self, **data: Any) -> None: 

26 try: 

27 super().__init__(**data) 

28 except ValidationError as e: 

29 msg = f"Validation error creating {self.__class__.__name__}\n" 

30 for error in e.errors(): 

31 msg += f" -> {error['loc'][0]}: {error['msg']}\n" 

32 

33 msg += f"\nError: {e}\n\n" 

34 LOG().error(msg) 

35 raise 

36 

37 @classmethod 

38 def from_dict(cls, data: dict[str, Any]) -> Self: 

39 """ 

40 Load model from dict with validation and error handling. 

41 

42 Args: 

43 data: Dictionary of model data 

44 

45 Returns: 

46 Instance of the model 

47 

48 Raises: 

49 DataLoadError: If validation fails 

50 CodePathError: If unexpected error occurs 

51 

52 """ 

53 from flipdare.app_log import LOG 

54 from flipdare.error.app_error import CodePathError 

55 from flipdare.error.data_load_error import DataLoadError 

56 from flipdare.error.message_format import ValidationErrorMsgFormat 

57 

58 try: 

59 return cls(**data) 

60 except ValidationError as e: 

61 formatter = ValidationErrorMsgFormat( 

62 class_type=cls, 

63 error=e, 

64 ) 

65 LOG().error(str(formatter)) 

66 raise DataLoadError.model( 

67 class_name=formatter.class_name, 

68 missing_code=formatter.user_error_code, 

69 error=e, 

70 ) from e 

71 except Exception as e: 

72 msg = f"Unexpected error creating model from dict: {e}" 

73 LOG().error(msg) 

74 raise CodePathError(message=msg) from e 

75 

76 @property 

77 def searchable_values(self) -> list[str]: 

78 """Should be overridden by subclasses that have searchable values""" 

79 return [] 

80 

81 def calculate_change_score(self, other: Any) -> float: # noqa: ARG002 

82 """ 

83 NOTE: This should be overridden by models where we want to calculate a change score for updates. 

84 """ 

85 from flipdare.constants import DEFAULT_CHANGE_SCORE 

86 

87 return DEFAULT_CHANGE_SCORE 

88 

89 def update(self, **kwargs: Any) -> bool: 

90 # NOTE: this does not update timestamps which need to be updated in the wrapper.. 

91 has_changed = False 

92 # Access the schema definition from the class, not the instance 

93 # model_definition: dict[str, FieldInfo] = type(self).model_fields 

94 

95 for key, value in kwargs.items(): 

96 if key not in self.__pydantic_fields__: 

97 continue 

98 

99 changed = self._update(key, value) 

100 if changed: 

101 has_changed = True 

102 

103 return has_changed 

104 

105 def _update(self, key: str, value: Any) -> bool: 

106 # Use .get() to satisfy subscriptable checks and handle missing keys safely 

107 # note: this should be only called internally, since it has 

108 # note: no real checks on key or value 

109 fields: dict[str, FieldInfo] = type(self).model_fields 

110 field_info = fields.get(key) 

111 

112 if field_info is None: 

113 return False 

114 

115 if value is None and field_info.is_required(): 

116 raise ValueError(f"Field '{key}' is mandatory and cannot be set to None.") 

117 

118 if getattr(self, key) != value: 

119 setattr(self, key, value) 

120 return True 

121 

122 return False 

123 

124 def to_json_dict(self) -> dict[str, Any]: 

125 """ 

126 Convert model to JSON-compatible dict with aliases. 

127 Handles Firestore Sentinel timestamps by converting to current UTC time. 

128 """ 

129 from google.cloud.firestore_v1.transforms import Sentinel 

130 from flipdare.util.time_util import TimeUtil 

131 

132 data: dict[str, Any] = self.model_dump(mode="json", by_alias=True, exclude={"id"}) 

133 

134 # Convert any sentinels to current utc time for JSON serialization 

135 for key in ["created_at", "updated_at"]: 

136 if key in data and isinstance(data[key], Sentinel): 

137 data[key] = TimeUtil.get_current_utc_dt() 

138 

139 return data 

140 

141 def to_dict(self) -> dict[str, Any]: 

142 """Convert model to dict with aliases, excluding id.""" 

143 return self.model_dump(by_alias=True, exclude={"id"}) 

144 

145 def to_dict_with_id(self) -> dict[str, Any]: 

146 """ 

147 Convert model to dict including id field. 

148 Used for debugging/logging purposes. 

149 

150 Raises: 

151 ValueError: If model has no id field or id is None 

152 

153 """ 

154 doc_id = getattr(self, "id", None) 

155 if doc_id is None: 

156 raise ValueError("Cannot convert to dict with id when id is None") 

157 

158 data: dict[str, Any] = self.model_dump(by_alias=False) 

159 data["id"] = doc_id 

160 return data 

161 

162 def debug_str(self) -> str: 

163 """Pretty prints fields, current values, and their defaults.""" 

164 lines = [f"<{self.__class__.__name__}>"] 

165 

166 for field_name, field_info in self.__class__.model_fields.items(): 

167 # Get current value (or fallback to default if not set) 

168 current_value = getattr(self, field_name) 

169 

170 # Identify the default 

171 default = field_info.default if field_info.default is not None else "Required" 

172 if field_info.default_factory: 

173 default = f"Factory({field_info.default_factory.__name__})" 

174 

175 alias = f" (alias: {field_info.alias})" if field_info.alias else "" 

176 

177 lines.append( 

178 f" • {field_name}{alias}:" 

179 f"\n Value: {current_value!r}" 

180 f"\n Default: {default}", 

181 ) 

182 

183 return "\n".join(lines)