Coverage for functions \ flipdare \ service \ _error_mixin.py: 93%

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 typing import Any, NoReturn, Protocol, runtime_checkable 

14import flask 

15from flipdare.app_types import DatabaseDict 

16from flipdare.backend.app_logger import AppLogger 

17from flipdare.app_log import LOG 

18from flipdare.core.app_response import AppErrorResponse 

19from flipdare.generated.schema.error.error_code_schema import ErrorCodeSchema 

20from flipdare.generated.schema.error_schema import ErrorSchema 

21from flipdare.message.user_error_code import UserErrorCode 

22from flipdare.result.app_result import AppResult 

23from flipdare.error.app_error import AppError 

24from flipdare.error.error_context import ErrorContext 

25from flipdare.generated.shared.app_error_code import AppErrorCode 

26from flipdare.generated.shared.backend.app_job_type import AppJobType 

27from flipdare.generated.shared.firestore_collections import FirestoreCollections 

28from flipdare.message.error_message import ErrorMessage 

29 

30 

31@runtime_checkable 

32class ErrorMixinRequirements(Protocol): 

33 

34 # NOTE: this must be implemented in the protocol 

35 @property 

36 def app_logger(self) -> AppLogger: ... 

37 

38 # 

39 # ErrorMixin methods 

40 # 

41 def callable_request_error( 

42 self, 

43 endpoint: str, 

44 notify_admin: bool = True, 

45 error_code: AppErrorCode = AppErrorCode.SERVER, 

46 message: str | None = None, 

47 error: Exception | None = None, 

48 ) -> ErrorSchema: ... 

49 

50 def http_request_error( 

51 self, 

52 endpoint: str, 

53 notify_admin: bool = True, 

54 error_code: AppErrorCode = AppErrorCode.SERVER, 

55 message: str | None = None, 

56 error: Exception | None = None, 

57 ) -> flask.Response: ... 

58 

59 def request_error( 

60 self, 

61 endpoint: str, 

62 notify_admin: bool = True, 

63 error_code: AppErrorCode = AppErrorCode.SERVER, 

64 message: str | None = None, 

65 error: Exception | None = None, 

66 ) -> AppError: ... 

67 

68 def job_error( 

69 self, 

70 job_type: AppJobType, 

71 error_code: AppErrorCode, 

72 message: str, 

73 doc_id: str | None = None, 

74 data: DatabaseDict | None = None, 

75 stack_str: str | None = None, 

76 error: Exception | None = None, 

77 notify_admin: bool = True, 

78 ) -> None: ... 

79 

80 

81class ErrorMixin: 

82 __slots__ = () 

83 

84 def log_and_throw( 

85 self: ErrorMixinRequirements, 

86 endpoint: str, 

87 error_code: AppErrorCode = AppErrorCode.SERVER, 

88 message: str | None = None, 

89 cause: Any | None = None, 

90 ) -> NoReturn: 

91 # NOTE: this needs to be handled somewhere, if its called 

92 # NOTE: on a request handler path. 

93 LOG().error(f"Server error in request {endpoint}: {cause}") 

94 message = message if message is not None else ErrorMessage.INTERNAL_ERROR 

95 raise AppError.from_context( 

96 ErrorContext.server_error( 

97 endpoint, 

98 error_code=error_code, 

99 message=message, 

100 cause=cause, 

101 ), 

102 ) 

103 

104 def http_validation_error( 

105 self: ErrorMixinRequirements, 

106 endpoint: str, 

107 error: Exception, 

108 user_error_code: str | None = None, 

109 notify_admin: bool = True, 

110 ) -> flask.Response: 

111 LOG().error(f"HTTP Validation error for {endpoint}: {error}") 

112 if user_error_code is None: 

113 user_error_code = UserErrorCode.fallback_code(error) 

114 

115 msg = ErrorMessage.INVALID_REQUEST.formatted(ErrorCodeSchema(code=user_error_code)) 

116 return self.http_request_error( 

117 endpoint=endpoint, 

118 message=msg, 

119 error_code=AppErrorCode.INVALID_INPUT, 

120 notify_admin=notify_admin, 

121 ) 

122 

123 def http_request_error( 

124 self: ErrorMixinRequirements, 

125 endpoint: str, 

126 notify_admin: bool = True, 

127 error_code: AppErrorCode = AppErrorCode.SERVER, 

128 message: str | None = None, 

129 error: Exception | None = None, 

130 ) -> flask.Response: 

131 LOG().error(f"HTTP Request error for {endpoint}: {error}") 

132 

133 error = self.request_error( 

134 endpoint=endpoint, 

135 notify_admin=notify_admin, 

136 error_code=error_code, 

137 message=message, 

138 error=error, 

139 ) 

140 

141 return AppErrorResponse.from_context(error.context).raw_response() 

142 

143 def callable_validation_error( 

144 self: ErrorMixinRequirements, 

145 endpoint: str, 

146 error: Exception, 

147 user_error_code: str | None = None, 

148 notify_admin: bool = True, 

149 ) -> ErrorSchema: 

150 LOG().error(f"Callable Validation error for {endpoint}: {error}") 

151 if user_error_code is None: 

152 user_error_code = UserErrorCode.fallback_code(error) 

153 

154 msg = ErrorMessage.INVALID_REQUEST.formatted(ErrorCodeSchema(code=user_error_code)) 

155 return self.callable_request_error( 

156 endpoint=endpoint, 

157 message=msg, 

158 error_code=AppErrorCode.INVALID_INPUT, 

159 notify_admin=notify_admin, 

160 ) 

161 

162 def callable_request_error( 

163 self: ErrorMixinRequirements, 

164 endpoint: str, 

165 notify_admin: bool = True, 

166 error_code: AppErrorCode = AppErrorCode.SERVER, 

167 message: str | None = None, 

168 error: Exception | None = None, 

169 ) -> ErrorSchema: 

170 LOG().error(f"Callable Request error for {endpoint}: {message}") 

171 

172 error = self.request_error( 

173 endpoint=endpoint, 

174 notify_admin=notify_admin, 

175 error_code=error_code, 

176 message=message, 

177 error=error, 

178 ) 

179 

180 return AppErrorResponse.from_context(error.context).to_dict() 

181 

182 def job_error( 

183 self: ErrorMixinRequirements, 

184 job_type: AppJobType, 

185 error_code: AppErrorCode, 

186 message: str, 

187 doc_id: str | None = None, 

188 data: DatabaseDict | None = None, 

189 stack_str: str | None = None, 

190 error: Exception | None = None, 

191 notify_admin: bool = True, 

192 ) -> None: 

193 LOG().error(message) 

194 

195 self.app_logger.job_error( 

196 job_type=job_type, 

197 error_code=error_code, 

198 message=message, 

199 ex_error=error, 

200 notify_admin=notify_admin, 

201 doc_id=doc_id, 

202 data=data, 

203 stack_str=stack_str, 

204 ) 

205 

206 def cron_result_error( 

207 self: ErrorMixinRequirements, 

208 job_type: AppJobType, 

209 result: AppResult[Any], 

210 collection: FirestoreCollections, 

211 error_code: AppErrorCode, 

212 message: str, 

213 notify_admin: bool = True, 

214 ) -> None: 

215 LOG().error(message) 

216 

217 self.app_logger.from_result( 

218 job_type=job_type, 

219 result=result, 

220 collection=collection, 

221 notify_admin=notify_admin, 

222 error_code=error_code, 

223 message=message, 

224 ) 

225 

226 def request_error( 

227 self: ErrorMixinRequirements, 

228 endpoint: str, 

229 notify_admin: bool = True, 

230 error_code: AppErrorCode = AppErrorCode.SERVER, 

231 message: str | None = None, 

232 error: Exception | None = None, 

233 ) -> AppError: 

234 LOG().error(f"Server error in request {endpoint}: {message}") 

235 

236 message = message if message is not None else ErrorMessage.INTERNAL_ERROR 

237 err = AppError.from_context( 

238 ErrorContext.server_error( 

239 endpoint, 

240 error_code=error_code, 

241 message=message, 

242 error=error, 

243 ), 

244 ) 

245 

246 self.app_logger.log_context(err.to_log_context(notify_admin=notify_admin)) 

247 return err