Coverage for functions \ flipdare \ result \ job_result.py: 95%
83 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#
14from dataclasses import dataclass
15from typing import Any, ClassVar, NoReturn, Self, override
16from flipdare.result.output_result import OutputResult
17from flipdare.app_types import DatabaseDict
18from flipdare.constants import NO_DOC_ID
19from flipdare.error.app_error_protocol import AppErrorProtocol
20from flipdare.result.app_result import AppResult
21from flipdare.generated.shared.app_error_code import AppErrorCode
22from flipdare.generated.shared.backend.app_job_type import AppJobType
23from flipdare.result.outcome import Outcome
24from flipdare.generated.shared.firestore_collections import FirestoreCollections
27@dataclass(kw_only=True)
28class JobResult[T](OutputResult):
29 _ERROR_MSG: ClassVar[str] = "You must set .result or .value before returning."
31 data: DatabaseDict | None = None
32 _app_result: AppResult[T] | None = None
34 @classmethod
35 def setup(
36 cls,
37 message: str,
38 duration: int = -1,
39 result: AppResult[T] | None = None,
40 error_code: AppErrorProtocol | None = None,
41 doc_id: str | None = None,
42 job_type: AppJobType | None = None,
43 collection: FirestoreCollections | None = None,
44 data: DatabaseDict | None = None,
45 ) -> Self:
46 error_code = error_code or (result.main_error if result else None) or None
47 return cls(
48 _app_result=result,
49 duration=duration,
50 data=data,
51 outcome=Outcome.ERROR if error_code else Outcome.OK,
52 error_code=error_code,
53 doc_id=doc_id or (result.doc_id if result else NO_DOC_ID),
54 job_type=job_type,
55 collection=collection,
56 message=message,
57 )
59 @classmethod
60 def from_result(
61 cls,
62 result: AppResult[T],
63 duration: int = -1,
64 message: str | None = None,
65 doc_id: str | None = None,
66 job_type: AppJobType | None = None,
67 collection: FirestoreCollections | None = None,
68 data: DatabaseDict | None = None,
69 ) -> Self:
70 """Create OutputAppResult from an AppResult. Outcome is determined by AppResult state (ok, error)."""
71 outcome = result.outcome
72 error_code: AppErrorProtocol | None = None
74 base_msg = message or result.message
76 match outcome:
77 case Outcome.OK:
78 message = base_msg or "Operation completed successfully."
79 case Outcome.SKIPPED:
80 message = base_msg or "Operation skipped."
81 case Outcome.WARNING:
82 # we dont set an error code, because its a warning ..
83 message = base_msg or "Operation completed with warnings."
84 case Outcome.ERROR:
85 error_code = result.main_error or AppErrorCode.SERVER
86 message = base_msg or f"Operation failed with error: {error_code}"
88 return cls(
89 _app_result=result,
90 duration=duration,
91 data=data,
92 outcome=outcome,
93 error_code=error_code,
94 doc_id=doc_id or result.doc_id,
95 job_type=job_type,
96 collection=collection,
97 message=message,
98 )
100 @classmethod
101 @override
102 def ok(
103 cls,
104 *,
105 duration: int = -1,
106 doc_id: str | None = None,
107 message: str = "Operation completed successfully.",
108 job_type: AppJobType | None = None,
109 collection: FirestoreCollections | None = None,
110 **kwargs: Any,
111 ) -> Self:
112 """Create successful OutputAppResult. Note: Requires doc_id to be provided (no default in practice)."""
113 return cls(
114 _app_result=AppResult[T].ok(doc_id=doc_id or NO_DOC_ID, message=message),
115 outcome=Outcome.OK,
116 duration=duration,
117 doc_id=doc_id or NO_DOC_ID,
118 job_type=job_type,
119 collection=collection,
120 message=message,
121 )
123 @classmethod
124 @override
125 def partial(
126 cls,
127 *,
128 error_code: AppErrorProtocol,
129 message: str,
130 duration: int,
131 doc_id: str | None = None,
132 job_type: AppJobType | None = None,
133 collection: FirestoreCollections | None = None,
134 **kwargs: Any,
135 ) -> Self:
136 """Create partial OutputAppResult. Note: Requires doc_id to be provided (no default in practice)."""
137 # note: partial are errors that are not critical enough to fail the entire job,
138 # note: but should still be logged and monitored.
139 # note: They often indicate issues that need attention, but do not necessarily require immediate action.
140 return cls(
141 _app_result=AppResult[T].error(
142 doc_id=doc_id or NO_DOC_ID, message=message, error_code=error_code
143 ),
144 outcome=Outcome.WARNING,
145 duration=duration,
146 error_code=error_code,
147 doc_id=doc_id or NO_DOC_ID,
148 job_type=job_type,
149 collection=collection,
150 message=message,
151 )
153 @classmethod
154 def skip_doc(
155 cls,
156 doc_id: str,
157 message: str,
158 duration: int = -1,
159 job_type: AppJobType | None = None,
160 collection: FirestoreCollections | None = None,
161 data: DatabaseDict | None = None,
162 ) -> Self:
163 """Create skipped OutputAppResult for a specific document."""
164 return cls(
165 _app_result=AppResult[T].skip(doc_id=doc_id, message=message or "Operation skipped."),
166 data=data,
167 duration=duration,
168 outcome=Outcome.SKIPPED,
169 doc_id=doc_id,
170 job_type=job_type,
171 collection=collection,
172 message=message,
173 )
175 @classmethod
176 def skip_job(
177 cls,
178 job_type: AppJobType,
179 message: str,
180 duration: int = -1,
181 collection: FirestoreCollections | None = None,
182 data: DatabaseDict | None = None,
183 ) -> Self:
184 return cls(
185 _app_result=AppResult[T].skip(doc_id=NO_DOC_ID, message=message),
186 duration=duration,
187 data=data,
188 outcome=Outcome.SKIPPED,
189 doc_id=NO_DOC_ID,
190 job_type=job_type,
191 collection=collection,
192 message=message,
193 )
195 # --------------------------------------------------------------------------------------------
196 # STATE UPDATE METHODS
197 # --------------------------------------------------------------------------------------------
199 @override
200 def set_ok(
201 self,
202 message: str,
203 doc_id: str | None = None,
204 ) -> None:
205 current_doc_id = self.app_result.doc_id
206 self._app_result = AppResult[T].ok(doc_id=current_doc_id, message=message)
207 self.outcome = Outcome.OK
208 self.doc_id = doc_id if doc_id is not None else current_doc_id
209 self.error_code = None
210 self.message = message
211 self.data = None
213 @override # type: ignore[override]
214 def set_error(
215 self,
216 error_code: AppErrorCode,
217 message: str,
218 *,
219 app_result: AppResult[Any] | None = None,
220 data: DatabaseDict | None = None,
221 ) -> None:
222 self._app_result = app_result
223 self.data = data
224 self.outcome = Outcome.ERROR
225 self.error_code = error_code
226 self.message = message
228 # --------------------------------------------------------------------------------------------
229 # Access state
230 # --------------------------------------------------------------------------------------------
231 @property
232 def is_finalized(self) -> bool:
233 return self._app_result is not None
235 @property
236 def app_result(self) -> AppResult[T]:
237 if self._app_result is None:
238 self._finalize_error()
239 return self._app_result
241 @app_result.setter
242 def app_result(self, val: AppResult[T]) -> None:
243 self._app_result = val
245 # --------------------------------------------------------------------------------------------
246 # Properties
247 # --------------------------------------------------------------------------------------------
249 @property
250 def should_log(self) -> bool:
251 return not (self.outcome.is_ok or self.outcome.is_skipped)
253 def _finalize_error(self) -> NoReturn:
254 msg = f"OutputResult for {self.doc_id} was never finalized! {self._ERROR_MSG}"
255 raise ValueError(msg)