Coverage for functions \ flipdare \ task \ command \ _base_command.py: 39%

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 typing import Any 

14from collections.abc import Callable, Mapping 

15from flipdare.backend.app_logger import AppLogger 

16from flipdare.app_log import LOG 

17from flipdare.constants import IS_DEBUG 

18from flipdare.result.output_result import OutputResult 

19from flipdare.mailer.admin_mailer import AdminMailer 

20from flipdare.mailer._jinja_email_template import JinjaEmailTemplate 

21from flipdare.firestore.backend.exchange_rate_db import ExchangeRateDb 

22from flipdare.generated.shared.app_error_code import AppErrorCode 

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

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

25from flipdare.util.time_util import TimeUtil 

26 

27 

28class BaseCommand[TSchema: Mapping[str, Any]]: 

29 def __init__( 

30 self, 

31 job_type: AppJobType, 

32 schema_class: type[TSchema], 

33 command_callback: Callable[[], None | OutputResult], 

34 email_callback: Callable[[str, Exception | None], JinjaEmailTemplate[Any]], 

35 app_logger: AppLogger | None = None, 

36 mailer: AdminMailer | None = None, 

37 exchange_rate_db: ExchangeRateDb | None = None, 

38 ) -> None: 

39 self._job_type = job_type 

40 self._schema_class = schema_class 

41 self._command_callback = command_callback 

42 self._email_callback = email_callback 

43 self._app_logger = app_logger 

44 self._mailer = mailer 

45 self._exchange_rate_db = exchange_rate_db 

46 

47 def run_command(self) -> OutputResult: 

48 job_type = self.job_type 

49 debug_str = f"Command: {job_type.label} - {job_type.description}" 

50 ex_error: Exception | None = None 

51 

52 start = TimeUtil.get_current_utc_dt() 

53 

54 try: 

55 result = self._command_callback() 

56 if isinstance(result, OutputResult): 

57 message = result.message 

58 else: 

59 message = f"Successfully ran {self.command_name} command." 

60 except Exception as ex: 

61 message = f"Failed to run {self.command_name} command." 

62 ex_error = ex 

63 

64 template = self._email_callback(message, ex_error) 

65 ok = self.send_template(email_template=template) 

66 

67 end = TimeUtil.get_current_utc_dt() 

68 duration = TimeUtil.duration_in_seconds(start, end) 

69 if ok and ex_error is None: 

70 if IS_DEBUG: 

71 LOG().debug(f"Successfully ran command: {debug_str}") 

72 

73 return OutputResult.ok(job_type=job_type, message=message, duration=duration) 

74 

75 # handle error 

76 LOG().error(f"Error running command: {debug_str} - {message}") 

77 return OutputResult.error( 

78 error_code=AppErrorCode.COMMAND, 

79 duration=duration, 

80 job_type=job_type, 

81 message=message, 

82 ) 

83 

84 @property 

85 def app_logger(self) -> AppLogger: 

86 from flipdare.services import get_app_logger 

87 

88 if self._app_logger is None: 

89 self._app_logger = get_app_logger() 

90 return self._app_logger 

91 

92 @property 

93 def mailer(self) -> AdminMailer: 

94 from flipdare.services import get_admin_mailer 

95 

96 if self._mailer is None: 

97 self._mailer = get_admin_mailer() 

98 return self._mailer 

99 

100 @property 

101 def exchange_rate_db(self) -> ExchangeRateDb: 

102 from flipdare.services import get_exchange_rate_db 

103 

104 if self._exchange_rate_db is None: 

105 self._exchange_rate_db = get_exchange_rate_db() 

106 return self._exchange_rate_db 

107 

108 @property 

109 def keys(self) -> list[str]: 

110 return list(self._schema_class.__annotations__.keys()) 

111 

112 @property 

113 def command_name(self) -> str: 

114 return self.job_type.label 

115 

116 @property 

117 def job_type(self) -> AppJobType: 

118 return self._job_type 

119 

120 @property 

121 def priority(self) -> AppReportPriority: 

122 return self._job_type.priority 

123 

124 @property 

125 def command_description(self) -> str: 

126 return self.job_type.description 

127 

128 @property 

129 def _debug_label(self) -> str: 

130 return f"{self.command_name}/{self.command_description}\n" 

131 

132 def send_template(self, email_template: JinjaEmailTemplate[Any]) -> bool: 

133 LOG().info(f"Sending {self.command_name} report email to admins.") 

134 try: 

135 self.mailer.send(email_template=email_template) 

136 return True 

137 except Exception as ex: 

138 cause = f"Failed to send {self.command_name} report email: {ex}" 

139 self.log_error(cause) 

140 return False 

141 

142 def log_error(self, cause: str) -> None: 

143 LOG().error(f"{self.command_name} Report Error: {cause}") 

144 self.app_logger.system_error( 

145 job_type=self.job_type, 

146 error_code=AppErrorCode.SERVER_REPORT, 

147 message=cause, 

148 notify_admin=True, 

149 )