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
« 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#
13from __future__ import annotations
15from typing import Any, Self, TypeGuard, override
17from flipdare.app_types import SchemaDict
18from flipdare.error.app_error_protocol import AppErrorProtocol
19from flipdare.error.error_context import ErrorContext
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
25__all__ = [
26 "ErrorGuard",
27 "AppError",
28 "DatabaseError",
29 "JobError",
30 "ServerError",
31 "CodePathError",
32 "AuthError",
33 "UserNotFoundError",
34 "SearchError",
35]
38class ErrorGuard:
39 @staticmethod
40 def is_error(data: SchemaDict) -> TypeGuard[ErrorSchema]:
41 return "code" in data and "category" in data
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
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
68 super().__init__(message)
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 )
82 @property
83 def source(self) -> str:
84 """Get the source of this error (e.g. URL, job type, etc.)."""
85 return self._source
87 @property
88 def cause(self) -> BaseException | None:
89 if self._error is not None:
90 return self._error
92 return self.__cause__
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 )
102 @property
103 def error_code(self) -> AppErrorProtocol:
104 return self._error_code
106 @property
107 def category(self) -> AppLogCategory:
108 return self._error_code.category
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 )
118 @property
119 def message(self) -> str:
120 return self._message
122 @property
123 def cause_message(self) -> str | None:
124 if self._cause_message is not None:
125 return self._cause_message
127 cause = self.cause
128 return str(cause) if cause else None
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 )
140 def to_log_context(self, notify_admin: bool = True) -> LogContext:
141 from flipdare.error.message_format import AppErrorMsgFormat
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 )
153 def to_dict(self) -> ErrorSchema:
154 return self.context.to_dict()
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 )
168 @override
169 def __repr__(self) -> str:
170 return self.context.__repr__()
172 @override
173 def __str__(self) -> str:
174 return self.context.__str__()
177class UserNotFoundError(AppError):
178 CODE = AppErrorCode.USER_NOT_FOUND
181class SearchError(AppError):
182 CODE = AppErrorCode.SEARCH
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 )
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 )
205class ServerError(AppError):
206 CODE = AppErrorCode.SERVER
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)
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
229 super().__init__(
230 source=collection_name or "database",
231 message=message,
232 error_code=error_code,
233 error=error,
234 )
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
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
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)
257 @property
258 def job_type(self) -> AppJobType:
259 """Get the job type associated with this exception, if any."""
260 return self._job_type