Send E-Mail with output log and success status in Python

Question

I want to send an E-Mail automatically after each script run, whether it was successfull or not. The E-Mail should include logs and in case of an error the stack trace. The subject should contain a postfix that tells whether there the run was with errors ("ERROR") or without errors ("SUCCESS").

Currently I'm using using the follwing code, which does this in an ugly way:

if __name__ == '__main__':
    # setup logging and e-mail messaging
    logger = logging.getLogger(__name__)  # Creating an instance of logger
    logger.setLevel(logging.DEBUG)
    mail_handler = BufferingSMTPHandler(fromaddr='xyz@test.de',
                                        toaddr=['xyz@tzest.de'],
                                        subject='JOB: Testjob')
    mail_handler.setLevel(logging.INFO)  # Set minimum level to receive messages
    logger.addHandler(mail_handler)
    # rest of __main__ is at the end of the file!


def handle_exception(exc_type, exc_value, exc_traceback):
    if issubclass(exc_type, KeyboardInterrupt):
        sys.__excepthook__(exc_type, exc_value, exc_traceback)
        return

    logger.critical("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback))
    mail_handler.set_failed(True)
    mail_handler.flush()

if __name__ == '__main__':
    .... do work here
    logger.info('TEST')
    ....
    mail_handler.set_failed(False)  # reached end of the script, so success is assumed
    mail_handler.flush()

The BufferingSMTPHandler looks like this:

import logging
import logging.handlers
import smtplib


class BufferingSMTPHandler(logging.handlers.BufferingHandler):
    def __init__(self, subject, fromaddr, toaddr, mailhost='mysmtpserver', capacity=500):
        logging.handlers.BufferingHandler.__init__(self, capacity)
        self.mailhost = mailhost
        self.fromaddr = fromaddr
        self.toaddr = toaddr
        self.subject = subject
        self.failed = False

        self.setFormatter(logging.Formatter(
            '{levelname} - {asctime} - {lineno} - {name} -  In {funcName}: {message}',
            style='{'))  # Formatter to prettify logs

    def set_failed(self, failed):
        self.failed = failed

    def flush(self):  # Method to send emails
        if len(self.buffer) <= 0:
            return

        with smtplib.SMTP(self.mailhost) as smtp:
            if self.failed:
                subject = self.subject + ' - ERROR'
            else:
                subject = self.subject + ' - SUCCESS'
            body = ''

            for record in self.buffer:
                body += self.format(record) + '
'  # Populating body of the message with formatted logs

            msg = f'Subject: {subject}

{body}'

            smtp.sendmail(self.fromaddr, self.toaddr, msg)
            self.buffer.clear()  # Clearing buffer to allude sending same logs

Is there a way to do this without adding so much code to the main script?

E.g. in this kind of way (pseudo-code):

from status_mailer import AutoStatusMailer

if __name__ == '__main__':
  mailer = AutoStatusMailer(fromaddr='xyz@test.de', toaddr=['xyz@tzest.de'], subject='JOB: Testjob')
  logger = mailer.get_logger()
  .... do work here
  logger.info('TEST')
  ...
  ... status mail is automatically sent afterwards

Answer 1

I've put together a class called AutoStatusMailer based on information found on StackOverflow to solve the problem:

import atexit
import collections
import logging
import smtplib
import sys
from email.message import EmailMessage


class TailLogHandler(logging.Handler):
    def __init__(self, log_queue, formatter=None):
        logging.Handler.__init__(self)
        self.log_queue = log_queue
        if not formatter:
            formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
        self.setFormatter(formatter)

    def emit(self, record):
        self.log_queue.append(self.format(record))


class TailLogger(object):
    def __init__(self, maxlen, formatter=None):
        self._log_queue = collections.deque(maxlen=maxlen)
        self._log_handler = TailLogHandler(self._log_queue, formatter)

    def contents(self):
        return '
'.join(self._log_queue)

    @property
    def log_handler(self):
        return self._log_handler


class AutoStatusMailer:
    def __init__(self, fromaddr, toaddr, subject, mailhost, log_length=500, log_level=logging.DEBUG, log_console=True,
                 call_original_excepthook_on_error=False):
        self.mailhost = mailhost
        self.fromaddr = fromaddr
        self.toaddr = toaddr
        self.subject = subject
        self.failed = False
        self.call_original_excepthook_on_error = call_original_excepthook_on_error

        self.logger = logging.getLogger('auto_status_mail_logger')  # Creating an instance of logger
        self.logger.setLevel(log_level)

        if log_console:
            console_handler = logging.StreamHandler(sys.stdout)
            console_handler.setLevel(log_level)
            self.logger.addHandler(console_handler)

        self.tail = TailLogger(log_length)
        tail_handler = self.tail.log_handler
        tail_handler.setLevel(log_level)  # Set minimum level to receive messages
        self.logger.addHandler(tail_handler)

        self.original_excepthook = sys.excepthook
        sys.excepthook = self.handle_exception
        atexit.register(self.send_mail)

    def get_logger(self):
        return self.logger

    def handle_exception(self, exc_type, exc_value, exc_traceback):
        if issubclass(exc_type, KeyboardInterrupt):
            sys.__excepthook__(exc_type, exc_value, exc_traceback)
            return

        self.logger.critical("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback))
        self.failed = True

        if self.call_original_excepthook_on_error:
            self.original_excepthook(exc_type, exc_value, exc_traceback)

    def send_mail(self):
        with smtplib.SMTP(self.mailhost) as smtp:
            if self.failed:
                subject = self.subject + ' - ERROR'
            else:
                subject = self.subject + ' - SUCCESS'
            msg = EmailMessage()
            msg.set_content(self.tail.contents())
            msg['Subject'] = subject
            msg['From'] = self.fromaddr
            msg['To'] = ', '.join(self.toaddr)
            smtp.send_message(msg)

It is used like this:

from auto_status_mailer import AutoStatusMailer

auto_status_mailer = AutoStatusMailer(fromaddr='test@test.de',
                                      toaddr=['test@test.de', 'test2@test.de'],
                                      subject='JOB: My Job',
                                      mailhost='mri.server.lan')

logger = auto_status_mailer.get_logger()

logger.info("Test")
# raise ValueError("TEST")