Coverage for functions \ flipdare \ firestore \ db_bridge.py: 47%

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""" 

14Generic database operations for PersistedWrapper/AppBaseModel subclasses. 

15 

16This module provides a generic way to perform common CRUD operations on any 

17PersistedWrapper/AppBaseModel subclass using AppDb or AppSubDb instances, reducing code duplication 

18across admin and service classes. All database operations return PersistedWrapper 

19to guarantee doc_id exists. 

20""" 

21 

22from typing import Any 

23 

24from flipdare.app_log import LOG 

25from flipdare.constants import IS_DEBUG 

26from flipdare.result.app_result import AppResult 

27from flipdare.firestore._app_db import AppDb 

28from flipdare.firestore.core.app_base_model import AppBaseModel 

29from flipdare.generated.shared.app_error_code import AppErrorCode 

30from flipdare.wrapper import PersistedWrapper 

31 

32__all__ = ["DbBridge"] 

33 

34 

35class DbBridge[W: PersistedWrapper[Any], D: AppDb[Any, Any]]: 

36 """ 

37 Generic database operations handler for PersistedWrapper/AppBaseModel subclasses. 

38 

39 Provides common get and update operations with consistent error handling 

40 and AppResult wrapping. 

41 

42 Example usage: 

43 user_ops = DbBridge(user_db, "User") 

44 result = user_ops.get("user_123") 

45 if result.has_errors: 

46 # handle error 

47 user = result.generated 

48 

49 # Update 

50 user.update_field("some_field", "new_value") 

51 update_result = user_ops.update(user) 

52 """ 

53 

54 def __init__(self, db: D, model_name: str) -> None: 

55 """ 

56 Initialize generic database operations. 

57 

58 Args: 

59 db: The AppDb instance for this model type 

60 model_name: Human-readable name for the model (e.g., "User", "Flag", "FlagAction") 

61 

62 """ 

63 self._db = db 

64 self._model_name = model_name 

65 

66 @property 

67 def db(self) -> D: 

68 """Get the underlying database instance.""" 

69 return self._db 

70 

71 @property 

72 def model_name(self) -> str: 

73 """Get the model name.""" 

74 return self._model_name 

75 

76 def get(self, doc_id: str) -> AppResult[W]: 

77 """ 

78 Get a model by document ID. 

79 

80 Args: 

81 doc_id: The document ID to retrieve 

82 

83 Returns: 

84 AppResult with specific wrapper type in .generated if successful, errors otherwise 

85 

86 """ 

87 result = AppResult[W](doc_id=doc_id) 

88 debug_msg = f"{self._model_name} {doc_id}" 

89 

90 try: 

91 model = self._db.get(doc_id) 

92 if model is None: 

93 msg = f"{debug_msg} not found." 

94 result.add_error(AppErrorCode.NOT_FOUND, msg) 

95 return result 

96 

97 result.generated = model 

98 return result 

99 except Exception as e: 

100 msg = f"Exception retrieving {debug_msg}: {e}" 

101 LOG().error(msg) 

102 result.add_error(AppErrorCode.DATABASE_EX, msg) 

103 return result 

104 

105 def update(self, model: W) -> AppResult[W]: 

106 """ 

107 Update a PersistedWrapper with doc_id set that has changes. 

108 

109 Args: 

110 model: The wrapper instance to update (must have doc_id set) 

111 

112 Returns: 

113 AppResult with updated wrapper of specific type in .generated if successful, errors otherwise 

114 

115 """ 

116 doc_id = model.doc_id 

117 debug_name = f"{self.model_name}:{doc_id}" 

118 result = AppResult[W](doc_id=model.doc_id) 

119 

120 try: 

121 if not model.has_changes: 

122 msg = f"No updates found for {debug_name}." 

123 LOG().warning(msg) 

124 result.add_error(AppErrorCode.UNEXPECTED_CODE_PATH, msg) 

125 return result 

126 

127 updates = model.get_updates() 

128 if IS_DEBUG: 

129 LOG().debug(f"Updating {debug_name} with changes: {updates}") 

130 

131 updated_model = self._db.update(model.doc_id, updates) 

132 if updated_model is None: 

133 msg = f"Failed to update {debug_name}." 

134 result.add_error(AppErrorCode.DATABASE_EX, msg) 

135 return result 

136 

137 result.generated = updated_model 

138 return result 

139 except Exception as e: 

140 msg = f"Exception updating {debug_name}: {e}" 

141 LOG().error(msg) 

142 result.add_error(AppErrorCode.DATABASE_EX, msg) 

143 return result 

144 

145 def create(self, model: AppBaseModel) -> AppResult[W]: 

146 """ 

147 Create a new wrapper in the database. 

148 

149 Args: 

150 model: The wrapper instance to create 

151 

152 Returns: 

153 AppResult with specific wrapper type of created model in .generated if successful, errors otherwise 

154 

155 """ 

156 result = AppResult[W]() 

157 

158 try: 

159 created_model = self._db.create(model) 

160 result.generated = created_model 

161 result.doc_id = created_model.doc_id 

162 return result 

163 except Exception as e: 

164 msg = f"Exception creating {self.model_name}: {e}" 

165 LOG().error(msg) 

166 result.add_error(AppErrorCode.DATABASE_EX, msg) 

167 return result 

168 

169 def delete(self, doc_id: str) -> AppResult[None]: 

170 """ 

171 Delete a model from the database. 

172 

173 Args: 

174 doc_id: The document ID to delete 

175 

176 Returns: 

177 AppResult with success or error status 

178 

179 """ 

180 result = AppResult[None](doc_id=doc_id) 

181 

182 try: 

183 self._db.delete(doc_id) 

184 return result 

185 except Exception as e: 

186 msg = f"Exception deleting {self.model_name} {doc_id}: {e}" 

187 LOG().error(msg) 

188 result.add_error(AppErrorCode.DATABASE_EX, msg) 

189 return result 

190 

191 def exists(self, doc_id: str) -> bool: 

192 """ 

193 Check if a model exists in the database. 

194 

195 Args: 

196 doc_id: The document ID to check 

197 

198 Returns: 

199 True if the document exists, False otherwise 

200 

201 """ 

202 try: 

203 return self._db.exists(doc_id) 

204 except Exception: 

205 return False 

206 

207 def get_bulk(self, doc_ids: list[str]) -> AppResult[list[W]]: 

208 """ 

209 Get multiple wrappers by their document IDs. 

210 

211 Args: 

212 doc_ids: List of document IDs to retrieve 

213 

214 Returns: 

215 AppResult with list of PersistedWrapper in .generated if successful, errors otherwise 

216 

217 """ 

218 result = AppResult[list[W]]() 

219 

220 try: 

221 models = self._db.get_bulk(doc_ids) 

222 result.generated = models 

223 return result 

224 except Exception as e: 

225 msg = f"Exception retrieving bulk {self.model_name}s: {e}" 

226 LOG().error(msg) 

227 result.add_error(AppErrorCode.DATABASE_EX, msg) 

228 return result