import logging
from logging import Handler, LogRecord
from logging.handlers import SMTPHandler, BufferingHandler
from textwrap import dedent
from typing import List
from redmail.email.sender import EmailSender
class _EmailHandlerMixin:
def set_sender(self,
host, port,
user_name=None, password=None,
sender=None, receivers=None,
subject=None):
"Create a simple default sender"
self.sender = EmailSender(
host=host, port=port,
user_name=user_name, password=password
)
self.sender.receivers = receivers
self.sender.sender = sender
if receivers is not None:
self.sender.receivers = receivers
elif user_name is not None:
self.sender.receivers = user_name
else:
raise ValueError("Missing receiver")
self.sender.text = self.default_text
self.sender.subject = subject or "Log record"
def get_subject(self, record):
"Get subject of the email"
if self.fmt_subject is not None:
return self.fmt_subject.format(
record=record, handler=self
)
class EmailHandler(_EmailHandlerMixin, Handler):
"""Logging handler for sending a log record as an email
Parameters
----------
level : int
Log level of the handler
sender : EmailSender
Sender instance to be used for sending
the log records.
fmt_subject : str, optional
Format of the email subject. ``record``
is passed as a format argument.
kwargs : dict
Keyword arguments for creating the
sender if ``sender`` was not passed.
Examples
--------
Minimal example:
.. code-block:: python
handler = EmailHandler(
host="smtp.myhost.com", port=0,
fmt_subject="Log: {record.levelname}",
sender="no-reply@example.com",
receivers=["me@example.com"],
)
Customized example:
.. code-block:: python
from redmail import EmailSender
email = EmailSender(
host="smtp.myhost.com",
port=0
)
email.sender = "no-reply@example.com"
email.receivers = ["me@example.com"]
email.html = '''
Record: {{ record.levelname }}
{{ record.msg }}
Info
- Path: {{ record.pathname }}
- Function: {{ record.funcName }}
- Line number: {{ record.lineno }}
'''
handler = EmailHandler(sender=email, fmt_subject="{record.name}}: {record.levelname}")
import logging
logger = logging.getLogger()
logger.addHandler(handler)
"""
def __init__(self, level:int=logging.NOTSET, sender:EmailSender=None, fmt_subject=None, **kwargs):
super().__init__(level)
if sender is not None:
self.sender = sender
else:
self.set_sender(**kwargs)
self.fmt_subject = fmt_subject
default_text = "{{ msg }}"
def emit(self, record:logging.LogRecord):
"Emit a record"
self.sender.send(
subject=self.get_subject(record),
body_params={
"record": record,
"msg": self.format(record),
"handler": self,
}
)
class MultiEmailHandler(_EmailHandlerMixin, BufferingHandler):
"""Logging handler for sending multiple log records as an email
Parameters
----------
capacity : int
Number of
sender : EmailSender
Sender instance to be used for sending
the log records.
fmt_subject : str, optional
Format of the email subject. ``record``
is passed as a format argument.
kwargs : dict
Keyword arguments for creating the
sender if ``sender`` was not passed.
Examples
--------
Minimal example:
.. code-block:: python
handler = MultiEmailHandler(
host="smtp.myhost.com", port=0,
fmt_subject="Log: {min_level_name} - {max_level_name}",
sender="no-reply@example.com",
receivers=["me@example.com"],
)
Customized example:
.. code-block:: python
from redmail import EmailSender
email = EmailSender(
host="smtp.myhost.com",
port=0
)
email.sender = "no-reply@example.com"
email.receivers = ["me@example.com"]
email.html = '''
Record: {{ record.levelname }}
{{ record.msg }}
Info
- Path: {{ record.pathname }}
- Function: {{ record.funcName }}
- Line number: {{ record.lineno }}
'''
handler = EmailHandler(sender=email, fmt_subject="{record.name}}: {record.levelname}")
import logging
logger = logging.getLogger()
logger.addHandler(handler)
"""
default_text = dedent("""
{% for record in records -%}
Log Record
----------------------------
{% set msg = handler.format(record) -%}
{{ msg }}
{% endfor %}""")[1:]
def __init__(self, capacity:int, sender:EmailSender=None, fmt_subject=None, **kwargs):
super().__init__(capacity)
if sender is not None:
self.sender = sender
else:
self.set_sender(**kwargs)
self.fmt_subject = fmt_subject
def flush(self):
"Flush (send) the records"
self.acquire()
try:
for rec in self.buffer:
# This creates msg, exc_text etc. to LogRecords
self.format(rec)
# For some reason logging does not create this attr unless having asctime in the format string
if self.formatter is None:
rec.asctime = logging.Formatter().formatTime(rec)
self.sender.send(
subject=self.get_subject(self.buffer),
body_params={
"records": self.buffer,
"handler": self
}
)
self.buffer = []
finally:
self.release()
def get_subject(self, records:List[LogRecord]):
"Get subject of the email"
if self.fmt_subject is not None:
min_level = min([record.levelno for record in records])
max_level = max([record.levelno for record in records])
return self.fmt_subject.format(
min_level_name=logging.getLevelName(min_level),
max_level_name=logging.getLevelName(max_level),
handler=self,
records=records
)