Coverage for functions \ flipdare \ task \ report \ core \ _base_report.py: 89%

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 abc import ABC, abstractmethod 

14from typing import Any 

15 

16from flipdare.backend.app_logger import AppLogger 

17from flipdare.constants import IS_DEBUG, IS_TRACE 

18from flipdare.mailer.email_image import EmailImage 

19from flipdare.app_log import LOG 

20from flipdare.result.output_result import OutputResult 

21from flipdare.mailer.admin_mailer import AdminMailer 

22from flipdare.mailer._jinja_email_template import JinjaEmailTemplate 

23from flipdare.generated.shared.app_error_code import AppErrorCode 

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

25from flipdare.generated.shared.backend.app_report_priority import AppReportPriority 

26from flipdare.service._error_mixin import ErrorMixin 

27 

28 

29class BaseReport(ErrorMixin, ABC): 

30 __slots__ = ( 

31 "_app_logger", 

32 "_errors", 

33 "_job_type", 

34 "_mailer", 

35 "_priority", 

36 ) 

37 

38 def __init__( 

39 self, 

40 job_type: AppJobType, 

41 app_logger: AppLogger, 

42 mailer: AdminMailer, 

43 ) -> None: 

44 self._job_type = job_type 

45 self._app_logger = app_logger 

46 self._mailer = mailer 

47 self._priority = job_type.priority 

48 self._errors: list[str] = [] 

49 

50 @abstractmethod 

51 def build_template(self) -> JinjaEmailTemplate[Any] | OutputResult: ... 

52 

53 @property 

54 def report_name(self) -> str: 

55 return self.job_type.label 

56 

57 @property 

58 def app_logger(self) -> AppLogger: 

59 return self._app_logger 

60 

61 @property 

62 def mailer(self) -> AdminMailer: 

63 return self._mailer 

64 

65 @property 

66 def job_type(self) -> AppJobType: 

67 return self._job_type 

68 

69 @property 

70 def priority(self) -> AppReportPriority: 

71 return self._priority 

72 

73 @property 

74 def report_description(self) -> str: 

75 return self._job_type.description 

76 

77 @property 

78 def _debug_label(self) -> str: 

79 return f"{self.job_type.label}/{self.report_description}\n" 

80 

81 @property 

82 def errors(self) -> list[str]: 

83 return self._errors 

84 

85 def add_error(self, message: str) -> None: 

86 LOG().error(message) 

87 self._errors.insert(0, message) 

88 

89 @property 

90 def error_string(self) -> str | None: 

91 errors = self._errors 

92 if not errors: 

93 return None 

94 return "".join(f"\t{e}\n" for e in errors) 

95 

96 def create_and_send(self) -> OutputResult: 

97 result = self.build_template() 

98 if isinstance(result, OutputResult): 

99 return result 

100 

101 if IS_DEBUG: 

102 msg = f"Built template for {self.report_name} report images={len(result.images) if result.images else 0}" 

103 LOG().debug(msg) 

104 

105 ok = self.send_template(email_template=result, images=result.images) 

106 if not ok: 

107 msg = f"Failed to send report email for {self.report_name}" 

108 return self._error_result(msg=msg) 

109 else: 

110 return self._ok_result() 

111 

112 def send_template( 

113 self, 

114 email_template: JinjaEmailTemplate[Any], 

115 images: list[EmailImage] | None = None, 

116 ) -> bool: 

117 LOG().info(f"Sending {self.job_type.label} report email to admins.") 

118 try: 

119 self.mailer.send(email_template=email_template, images=images) 

120 return True 

121 except Exception as ex: 

122 msg = f"Failed to send {self.report_name}/{self.job_type.label} report email: {ex}" 

123 self._error_result(msg=msg) 

124 return False 

125 

126 def _ok_result(self) -> OutputResult: 

127 msg = f"Report {self.report_name} sent successfully" 

128 if IS_TRACE: 

129 LOG().trace(msg) 

130 

131 return OutputResult.ok( 

132 job_type=self.job_type, 

133 collection=self.job_type.default_collection, 

134 message=msg, 

135 ) 

136 

137 def _error_result(self, msg: str, code: AppErrorCode | None = None) -> OutputResult: 

138 LOG().error(msg) 

139 

140 code = code or AppErrorCode.REPORT_EMAIL 

141 return OutputResult.error( 

142 error_code=code, 

143 duration=-1, 

144 job_type=code, 

145 collection=self.job_type.default_collection, 

146 message=msg, 

147 )