python-mylib/mylib/report.py

156 lines
5 KiB
Python
Raw Normal View History

2021-04-19 14:47:30 +02:00
""" Report """
2021-04-19 14:55:07 +02:00
import atexit
2021-04-19 14:47:30 +02:00
import logging
from mylib.config import ConfigurableObject, StringOption
from mylib.email import EmailClient
log = logging.getLogger(__name__)
2021-05-19 19:19:57 +02:00
class Report(ConfigurableObject): # pylint: disable=useless-object-inheritance
"""Logging report"""
2021-04-19 14:47:30 +02:00
_config_name = "report"
_config_comment = "Email report"
_defaults = {
"recipient": None,
"subject": "Report",
"loglevel": "WARNING",
"logformat": "%(asctime)s - %(levelname)s - %(message)s",
"just_try": False,
}
2021-04-19 14:47:30 +02:00
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)
2021-04-19 14:47:30 +02:00
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)
2021-04-19 14:47:30 +02:00
if self.add_logging_handler:
logging.getLogger().addHandler(self.handler)
if self._send_at_exit:
self.send_at_exit()
2021-04-19 14:47:30 +02:00
def get_handler(self):
2024-03-15 09:52:23 +01:00
"""Retrieve logging handler"""
2021-04-19 14:47:30 +02:00
return self.handler
def write(self, msg):
"""Write a message"""
2021-04-19 14:47:30 +02:00
self.content.append(msg)
def get_content(self):
"""Read the report content"""
2021-04-19 14:47:30 +02:00
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")
2021-04-19 14:47:30 +02:00
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"
)
2021-04-19 14:47:30 +02:00
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")
2021-04-19 14:47:30 +02:00
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
2021-04-19 14:55:07 +02:00
def send_at_exit(self, **kwargs):
"""Send report at exit"""
2021-04-19 14:55:07 +02:00
atexit.register(self.send, **kwargs)