Coverage for functions\flipdare\result\app

1#!/usr/bin/env python

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.

9# This software includes third-party components licensed under MIT,

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details.

11#

13import sys

14from typing import TYPE_CHECKING, Any, TypedDict, override

15from collections import Counter, UserDict, defaultdict

16from dataclasses import dataclass

17from flipdare.app_types import DatabaseDict

18from flipdare.constants import NO_DOC_ID

19from flipdare.error.app_error_protocol import AppErrorProtocol

20from flipdare.generated.shared.app_error_code import AppErrorCode

21from flipdare.result.outcome import Outcome

22from flipdare.util.debug_util import stringify_debug

24if TYPE_CHECKING:

25 from flipdare.firestore.context._model_context import ModelContext

27__all__ = ["AppResult"]

29# internal_methods set for O(1) lookups

30_INTERNAL_METHODS = {"__init__", "ok", "error", "skip", "fail", "_base_task_name"}

32type ResultExtraType = DatabaseDict | str

35class AppResultError(TypedDict):

36 error_code: AppErrorProtocol

37 message: str

40def _base_task_name() -> str:

41 idx = 1

42 while True:

43 try:

44 frame = sys._getframe(idx)

45 code = frame.f_code

46 func_name = code.co_name

48 # 1. Skip internal factory methods

49 if func_name in _INTERNAL_METHODS:

50 idx += 1

51 continue

53 # 2. Use co_qualname (Python 3.11+) to get 'Class.method' directly

54 # This is significantly faster than inspect.stack()

55 qualname = code.co_qualname

57 # Handle GenericAlias noise or Global functions

58 if "GenericAlias" in qualname or func_name == "__call__":

59 idx += 1

60 continue

62 return qualname

64 except ValueError:

65 # Reached the end of the stack

66 return "Unknown.task"

69class AppResult[R = None]:

70 __slots__ = (

71 "_data",

72 "_errors",

73 "_extra_error_info",

74 "_generated",

75 "_main_task",

76 "_outcome",

77 "_task_names",

78 "_warnings",

79 "context",

80 "doc_id",

81 "message",

82 )

84 def __init__(

85 self,

86 *,

87 outcome: Outcome = Outcome.OK,

88 task_name: str | None = None,

89 doc_id: str = NO_DOC_ID,

90 data: DatabaseDict | None = None,

91 message: str | None = None,

92 ) -> None:

93 # context

94 self._main_task = _base_task_name()

95 if task_name is not None:

96 self._main_task = f"{self._main_task} - {task_name}"

97 self.doc_id = doc_id

98 self.message = message

99 self.context: ModelContext | None = None

100

101 # result

102 self._outcome = outcome

103 self._data = data

104 self._generated: R | None = None

105

106 # Internal Containers

107 self._task_names = [self._main_task]

108 self._warnings = _WarningEntries()

109 self._errors = _ErrorEntries()

110 self._extra_error_info: dict[AppErrorProtocol, ResultExtraType] = {}

111

112 @classmethod

113 def from_id_data(cls, data: DatabaseDict) -> "AppResult[R]":

114 doc_id: str = NO_DOC_ID

115 if "doc_id" in data:

116 doc_id = data["doc_id"]

117

118 result: AppResult[R] = AppResult(

119 doc_id=doc_id,

120 task_name=_base_task_name(),

121 data=data,

122 outcome=Outcome.OK,

123 )

124 if result.doc_id == NO_DOC_ID:

125 result.add_error(

126 AppErrorCode.MISSING_ID,

127 "Data is missing 'doc_id' field.",

128 )

129

130 return result

131

132 @classmethod

133 def ok(

134 cls,

135 doc_id: str,

136 message: str | None = None,

137 data: DatabaseDict | None = None,

138 ) -> "AppResult[R]":

139 return cls(doc_id=doc_id, data=data, outcome=Outcome.OK, message=message)

140

141 @classmethod

142 def skip(

143 cls,

144 doc_id: str,

145 message: str | None = None,

146 data: DatabaseDict | None = None,

147 ) -> "AppResult[R]":

148 return cls(doc_id=doc_id, data=data, outcome=Outcome.SKIPPED, message=message)

149

150 @classmethod

151 def error(

152 cls,

153 error_code: AppErrorProtocol,

154 message: str,

155 doc_id: str = NO_DOC_ID,

156 data: DatabaseDict | None = None,

157 ) -> "AppResult[R]":

158 res = cls(doc_id=doc_id, data=data, outcome=Outcome.ERROR, message=message)

159 res.add_error(error_code, message)

160 return res

161

162 def set_ok(self, message: str | None = None) -> None:

163 self._outcome = Outcome.OK

164 self.message = message

165 # clear the errors...

166 self._reset_errors()

167

168 def set_skipped(self, message: str | None = None) -> None:

169 self._outcome = Outcome.SKIPPED

170 self.message = message

171 # clear the errors since they won't be relevant for a skipped result

172 self._reset_errors()

173

174 def set_error(self, error_code: AppErrorCode, message: str) -> None:

175 self._outcome = Outcome.ERROR

176 self.message = message

177 self.add_error(error_code, message)

178

179 # --- Simplified Status Checks ---

180 @property

181 def is_error(self) -> bool:

182 return len(self._errors) > 0 or self._outcome.is_error

183

184 @property

185 def is_warning(self) -> bool:

186 return len(self._warnings) > 0 or self._outcome.is_warning

187

188 @property

189 def is_skipped(self) -> bool:

190 return self._outcome.is_skipped

191

192 @property

193 def is_ok(self) -> bool:

194 return self._outcome.is_ok

195

196 @property

197 def is_context_error(self) -> bool:

198 return self.context is not None and self.is_error

199

200 # --- Getters ---

201 @property

202 def main_task(self) -> str:

203 return self._main_task

204

205 @property

206 def task_names(self) -> list[str]:

207 return self._task_names

208

209 @property

210 def generated(self) -> R | None:

211 return self._generated

212

213 @generated.setter

214 def generated(self, value: R) -> None:

215 self._generated = value

216 if value is not None:

217 doc_id = getattr(value, "doc_id", None)

218 if doc_id is not None:

219 self.doc_id = doc_id

220

221 @property

222 def main_error(self) -> AppErrorProtocol | None:

223 if not self.is_error:

224 return None

225

226 return self._errors.main_error_type()

227

228 @property

229 def outcome(self) -> Outcome:

230 return self._outcome

231

232 def add_error(

233 self,

234 error_code: AppErrorProtocol,

235 message: str,

236 extra: DatabaseDict | str | None = None,

237 ) -> None:

238 self._errors.add(self.main_task, message, error_code)

239 if extra is not None:

240 self._extra_error_info[error_code] = extra

241 self._outcome = Outcome.ERROR # escalate to error if not already

242

243 def add_exception(

244 self,

245 error_code: AppErrorCode,

246 ex: Exception,

247 extra: DatabaseDict | str | None = None,

248 ) -> None:

249 self.add_error(error_code, f"{type(ex).__name__}: {ex}", extra)

250 self._outcome = Outcome.ERROR # escalate to error if not already

251

252 def add_warning(self, message: str) -> None:

253 self._warnings.add(self.main_task, message)

254 # warnings indicate partial failure

255 if not self._outcome.is_error:

256 self._outcome = Outcome.WARNING

257

258 # --- Internal ---

259 def _reset_errors(self) -> None:

260 self._errors = _ErrorEntries()

261 self._warnings = _WarningEntries()

262 self._extra_error_info.clear()

263

264 # --- Data Export ---

265 @property

266 def error_str(self) -> str:

267 if not self.is_error:

268 return "No errors."

269 return "\n".join(

270 f"\t{err.error_code.display_title}: {err.message}"

271 for entries in self._errors.values()

272 for err in entries

273 )

274

275 @property

276 def errors(self) -> list[AppResultError]:

277 return [

278 AppResultError({"error_code": record.error_code, "message": record.message})

279 for entries in self._errors.values()

280 for record in entries

281 ]

282

283 @property

284 def warnings(self) -> list[str]:

285 return [record.message for entries in self._warnings.values() for record in entries]

286

287 def merge(self, other: "AppResult[Any]") -> None:

288 """Merge another result into this one."""

289 # NOTE: We do not merge generated data or context,

290 # NOTE: as those are typically specific to individual tasks.

291 self._errors.update(other._errors)

292 self._warnings.update(other._warnings)

293 self._extra_error_info.update(other._extra_error_info)

294

295 # if other has a more severe outcome, escalate to it

296 replace_message = False

297 if other._outcome.is_error:

298 self._outcome = Outcome.ERROR

299 replace_message = True

300 elif other._outcome.is_warning and not self._outcome.is_error:

301 self._outcome = Outcome.WARNING

302 replace_message = True

303 elif other._outcome.is_skipped and self._outcome.is_ok:

304 self._outcome = Outcome.SKIPPED

305 replace_message = True

306

307 if replace_message and other.message is not None:

308 self.message = other.message

309

310 # --- Logging ---

311 def to_dict(self) -> dict[str, Any]:

312 # for using in app_log_db

313 error_dict: dict[str, Any] = {

314 "main_task": self.main_task,

315 "doc_id": self.doc_id,

316 "result_value": self.outcome.name,

317 }

318 if self.is_error:

319 error_dict["errors"] = {}

320 for task, errs in self._errors.items():

321 error_dict["errors"][task] = [str(err) for err in errs]

322 if self.is_warning:

323 error_dict["warnings"] = {}

324 for task, warns in self._warnings.items():

325 error_dict["warnings"][task] = [str(warn) for warn in warns]

326 return error_dict

327

328 @property

329 def formatted(self) -> str:

330 result = "\n" + "-" * 40 + "\n"

331 result += f" Doc Id: {self.doc_id}\n"

332 result += f" Error: {self.is_error} - Skipped: {self.is_skipped} - Warning: {self.is_warning}\n"

333 result += f" Main Error Type: {self.main_error}\n"

334 result += f" Main Task: {self.main_task}\n"

335 result += f" Result Value: {self.outcome}\n"

336

337 if not self.is_error and not self.is_warning:

338 return result

339

340 headers = ["Type", "Task", "Message"]

341

342 from tabulate import tabulate

343

344 extra_info_str = ""

345

346 entries: list[list[str]] = []

347 if self.is_error:

348 entries.append(["Errors", "", ""])

349 for task, errs in self._errors.items():

350 for err in errs:

351 extra_info_str = ""

352 if err.error_code in self._extra_error_info:

353 extra_info = self._extra_error_info[err.error_code]

354 extra_info_str += (

355 extra_info

356 if isinstance(extra_info, str)

357 else stringify_debug(extra_info)

358 )

359 extra_info_str += "\n\n"

360 entries.append(

361 [err.error_code.display_title, task, err.message],

362 )

363 if self.is_warning:

364 entries.append(["Warnings", "", ""])

365 for task, warns in self._warnings.items():

366 for warn in warns:

367 entries.extend([["WARNING", task, warn.message]])

368

369 if len(entries) <= 0:

370 return result

371

372 result += "\n"

373 result += tabulate(entries, headers=headers, tablefmt="plain")

374 result += "\n" + "-" * 40

375 if extra_info_str:

376 result += "\nExtra Info:\n" + extra_info_str + "\n" + "-" * 40

377 result += "\n"

378

379 return result

380

381 @override

382 def __repr__(self) -> str:

383 return f"<AppResult {self.outcome.name} for {self.main_task} (doc:{self.doc_id})>"

384

385

386@dataclass

387class _Entry:

388 """Base entry for warnings without error types."""

389

390 message: str

391

392 @override

393 def __str__(self) -> str:

394 return self.message

395

396

397@dataclass

398class _ErrorEntry:

399 """Error entry with error type classification."""

400

401 message: str

402 error_code: AppErrorProtocol

403

404 @override

405 def __str__(self) -> str:

406 return f"{self.error_code.display_title}: {self.message}"

407

408

409class _Entries[T](UserDict[str, list[T]]):

410 """Generic container. UserDict makes it compatible with Hamcrest/dict matchers."""

411

412 def __init__(self) -> None:

413 # UserDict stores everything in self.data

414 self.data: dict[str, list[T]] = defaultdict(list)

415

416 # __len__, __contains__, __getitem__, __setitem__, __iter__, and .items()

417 # are all provided by UserDict automatically.

418

419

420class _ErrorEntries(_Entries[_ErrorEntry]):

421 """Container for error entries only (always have error_type)."""

422

423 def main_error_type(self) -> AppErrorProtocol | None:

424 if not self.data:

425 return None

426

427 # Simplified counting using collections.Counter

428 counts = Counter(entry.error_code for entries in self.data.values() for entry in entries)

429

430 # most_common(1) returns [(value, count)] or []

431 return counts.most_common(1)[0][0] if counts else None

432

433 def has_error_type(self, error_code: AppErrorProtocol) -> bool:

434 return any(

435 entry.error_code == error_code for entries in self.data.values() for entry in entries

436 )

437

438 def add(self, key: str, message: str, error_code: AppErrorProtocol) -> None:

439 # defaultdict(list) handles the 'if key not in self.data' check for you

440 self.data[key].append(_ErrorEntry(message=message, error_code=error_code))

441

442

443class _WarningEntries(_Entries[_Entry]):

444 """Container for warning entries."""

445

446 def add(self, key: str, message: str) -> None:

447 self.data[key].append(_Entry(message=message))

Coverage for functions \ flipdare \ result \ app_result.py: 93%

244 statements