""" Report """ import atexit import logging from mylib.config import ConfigurableObject, StringOption from mylib.email import EmailClient log = logging.getLogger(__name__) class Report(ConfigurableObject): # pylint: disable=useless-object-inheritance """Logging report""" _config_name = "report" _config_comment = "Email report" _defaults = { "recipient": None, "subject": "Report", "loglevel": "WARNING", "logformat": "%(asctime)s - %(levelname)s - %(message)s", "just_try": False, } content = [] handler = None formatter = None email_client = None def __init__( self, email_client=None, add_logging_handler=False, send_at_exit=None, initialize=True, **kwargs, ): super().__init__(**kwargs) self.email_client = email_client self.add_logging_handler = add_logging_handler self._send_at_exit = send_at_exit self._attachment_files = [] self._attachment_payloads = [] if initialize: self.initialize() def configure(self, **kwargs): # pylint: disable=arguments-differ """Configure options on registered mylib.Config object""" section = super().configure( just_try_help=kwargs.pop("just_try_help", "Just-try mode: do not really send report"), **kwargs, ) section.add_option(StringOption, "recipient", comment="Report recipient email address") section.add_option( StringOption, "subject", default=self._defaults["subject"], comment="Report email subject", ) section.add_option( StringOption, "loglevel", default=self._defaults["loglevel"], comment='Report log level (as accept by python logging, for instance "INFO")', ) section.add_option( StringOption, "logformat", default=self._defaults["logformat"], comment='Report log level (as accept by python logging, for instance "INFO")', ) if not self.email_client: self.email_client = EmailClient(config=self._config) self.email_client.configure() return section def initialize(self, loaded_config=None): """Configuration initialized hook""" super().initialize(loaded_config=loaded_config) self.handler = logging.StreamHandler(self) loglevel = self._get_option("loglevel").upper() assert hasattr(logging, loglevel), f"Invalid report loglevel {loglevel}" self.handler.setLevel(getattr(logging, loglevel)) self.formatter = logging.Formatter(self._get_option("logformat")) self.handler.setFormatter(self.formatter) if self.add_logging_handler: logging.getLogger().addHandler(self.handler) if self._send_at_exit: self.send_at_exit() def get_handler(self): """Retrieve logging handler""" return self.handler def write(self, msg): """Write a message""" self.content.append(msg) def get_content(self): """Read the report content""" return "".join(self.content) def add_attachment_file(self, filepath): """Add attachment file""" self._attachment_files.append(filepath) def add_attachment_payload(self, payload): """Add attachment payload""" self._attachment_payloads.append(payload) def send(self, subject=None, rcpt_to=None, email_client=None, just_try=None): """Send report using an EmailClient""" if rcpt_to is None: rcpt_to = self._get_option("recipient") if not rcpt_to: log.debug("No report recipient, do not send report") return True if subject is None: subject = self._get_option("subject") assert subject, "You must provide report subject using Report.__init__ or Report.send" if email_client is None: email_client = self.email_client assert email_client, ( "You must provide an email client __init__(), send() or send_at_exit() methods argument" " email_client" ) content = self.get_content() if not content and not self._attachment_files and not self._attachment_payloads: log.debug("Report is empty, do not send it") return True msg = email_client.forge_message( rcpt_to, subject=subject, text_body=content, attachment_files=self._attachment_files, attachment_payloads=self._attachment_payloads, ) if email_client.send( rcpt_to, msg=msg, just_try=just_try if just_try is not None else self._just_try ): log.debug("Report sent to %s", rcpt_to) return True log.error("Fail to send report to %s", rcpt_to) return False def send_at_exit(self, **kwargs): """Send report at exit""" atexit.register(self.send, **kwargs)