Coverage for functions \ flipdare \ error \ app_error.py: 90%

105 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# 

12 

13from __future__ import annotations 

14 

15from typing import Any, Self, TypeGuard, override 

16 

17from flipdare.app_types import SchemaDict 

18from flipdare.error.app_error_protocol import AppErrorProtocol 

19from flipdare.error.error_context import ErrorContext 

20 

21from flipdare.error.log_context import LogContext 

22from flipdare.generated import AppLogCategory, AppErrorCode, AppJobType, ErrorSchema 

23from flipdare.generated.shared.backend.system_log_type import SystemLogType 

24 

25__all__ = [ 

26 "ErrorGuard", 

27 "AppError", 

28 "DatabaseError", 

29 "JobError", 

30 "ServerError", 

31 "CodePathError", 

32 "AuthError", 

33 "UserNotFoundError", 

34 "SearchError", 

35] 

36 

37 

38class ErrorGuard: 

39 @staticmethod 

40 def is_error(data: SchemaDict) -> TypeGuard[ErrorSchema]: 

41 return "code" in data and "category" in data 

42 

43 

44class AppError(Exception): 

45 # NOTE: you should use the new chaining 

46 # NOTE: i.e. raise AppError(...) from e 

47 # NOTE: to set the cause properly 

48 CODE: AppErrorProtocol = AppErrorCode.SERVER 

49 

50 def __init__( 

51 self, 

52 source: str, # either a url or a job type or some other source of the error 

53 message: str, 

54 error_code: AppErrorProtocol | None = None, 

55 http_code: int | None = None, 

56 title: str | None = None, 

57 cause: str | None = None, 

58 error: Exception | None = None, 

59 ) -> None: 

60 self._source = source 

61 self._message = message 

62 self._error_code = error_code or self.CODE 

63 self._override_http_code = http_code 

64 self._override_title = title 

65 self._cause_message = cause 

66 self._error = error 

67 

68 super().__init__(message) 

69 

70 @classmethod 

71 def from_context(cls, ctx: ErrorContext) -> Self: 

72 return cls( 

73 source=ctx.endpoint, 

74 title=ctx.title, 

75 message=ctx.message, 

76 error_code=ctx.error_code, 

77 http_code=ctx.http_code, 

78 cause=ctx.cause, 

79 error=ctx.error, 

80 ) 

81 

82 @property 

83 def source(self) -> str: 

84 """Get the source of this error (e.g. URL, job type, etc.).""" 

85 return self._source 

86 

87 @property 

88 def cause(self) -> BaseException | None: 

89 if self._error is not None: 

90 return self._error 

91 

92 return self.__cause__ 

93 

94 @property 

95 def http_code(self) -> int: 

96 return ( 

97 self._override_http_code 

98 if self._override_http_code is not None 

99 else self._error_code.http_code 

100 ) 

101 

102 @property 

103 def error_code(self) -> AppErrorProtocol: 

104 return self._error_code 

105 

106 @property 

107 def category(self) -> AppLogCategory: 

108 return self._error_code.category 

109 

110 @property 

111 def title(self) -> str: 

112 return ( 

113 self._override_title 

114 if self._override_title is not None 

115 else self._error_code.category.label 

116 ) 

117 

118 @property 

119 def message(self) -> str: 

120 return self._message 

121 

122 @property 

123 def cause_message(self) -> str | None: 

124 if self._cause_message is not None: 

125 return self._cause_message 

126 

127 cause = self.cause 

128 return str(cause) if cause else None 

129 

130 @property 

131 def context(self) -> ErrorContext: 

132 return ErrorContext( 

133 endpoint=self.source, 

134 title=self.title, 

135 message=self.message, 

136 error_code=self.error_code, 

137 cause=self.cause_message, 

138 ) 

139 

140 def to_log_context(self, notify_admin: bool = True) -> LogContext: 

141 from flipdare.error.message_format import AppErrorMsgFormat 

142 

143 return LogContext( 

144 log_type=SystemLogType.ERROR, 

145 called_by=self.source, 

146 category=self.category, 

147 message=self.message, 

148 error_code=self.error_code, 

149 formatter=AppErrorMsgFormat(self), 

150 notify_admin=notify_admin, 

151 ) 

152 

153 def to_dict(self) -> ErrorSchema: 

154 return self.context.to_dict() 

155 

156 def copy_with(self, **kwargs: Any) -> Self: 

157 cls = self.__class__ 

158 return cls( 

159 source=kwargs.get("source", self.source), 

160 message=kwargs.get("message", self.message), 

161 error_code=kwargs.get("error_code", self.error_code), 

162 http_code=kwargs.get("http_code", self.http_code), 

163 title=kwargs.get("title", self.title), 

164 cause=kwargs.get("cause", self.cause_message), 

165 error=kwargs.get("error", self.cause), 

166 ) 

167 

168 @override 

169 def __repr__(self) -> str: 

170 return self.context.__repr__() 

171 

172 @override 

173 def __str__(self) -> str: 

174 return self.context.__str__() 

175 

176 

177class UserNotFoundError(AppError): 

178 CODE = AppErrorCode.USER_NOT_FOUND 

179 

180 

181class SearchError(AppError): 

182 CODE = AppErrorCode.SEARCH 

183 

184 

185class CodePathError(AppError): 

186 def __init__(self, message: str, guard_check_failed: bool = False) -> None: 

187 super().__init__( 

188 source="code", 

189 error_code=( 

190 AppErrorCode.GUARD if guard_check_failed else AppErrorCode.UNEXPECTED_CODE_PATH 

191 ), 

192 message=message, 

193 ) 

194 

195 

196class AuthError(AppError): 

197 def __init__(self, url: str, message: str, is_error: bool = False) -> None: 

198 super().__init__( 

199 error_code=AppErrorCode.AUTH if is_error else AppErrorCode.PERMISSION_DENIED, 

200 source=url, 

201 message=message, 

202 ) 

203 

204 

205class ServerError(AppError): 

206 CODE = AppErrorCode.SERVER 

207 

208 def __init__( 

209 self, 

210 message: str, 

211 error_code: AppErrorProtocol = AppErrorCode.SERVER, 

212 error: Exception | None = None, 

213 ) -> None: 

214 super().__init__(source="system", error_code=error_code, message=message, error=error) 

215 

216 

217class DatabaseError(AppError): 

218 def __init__( 

219 self, 

220 message: str, 

221 error_code: AppErrorProtocol = AppErrorCode.DATABASE, 

222 collection_name: str | None = None, 

223 document_id: str | None = None, 

224 error: Exception | None = None, 

225 ) -> None: 

226 self._collection_name = collection_name 

227 self._document_id = document_id 

228 

229 super().__init__( 

230 source=collection_name or "database", 

231 message=message, 

232 error_code=error_code, 

233 error=error, 

234 ) 

235 

236 @property 

237 def collection_name(self) -> str | None: 

238 """Get the collection name associated with this exception, if any.""" 

239 return self._collection_name 

240 

241 @property 

242 def document_id(self) -> str | None: 

243 """Get the document ID associated with this exception, if any.""" 

244 return self._document_id 

245 

246 

247class JobError(AppError): 

248 def __init__( 

249 self, 

250 message: str, 

251 job_type: AppJobType, 

252 error_code: AppErrorProtocol, 

253 ) -> None: 

254 self._job_type = job_type 

255 super().__init__(source=job_type.value, message=message, error_code=error_code) 

256 

257 @property 

258 def job_type(self) -> AppJobType: 

259 """Get the job type associated with this exception, if any.""" 

260 return self._job_type