import logging
from logging import Handler, LogRecord
from logging.handlers import SMTPHandler, BufferingHandler
from textwrap import dedent
from typing import List, Optional
import warnings
from redmail.email.sender import EmailSender
class _EmailHandlerMixin:
def __init__(self, email, kwargs):
if email is not None:
# Using copy to prevent modifying the sender
# if it is used somewhere else
email = email.copy()
self.email = email
self._set_email_kwargs(kwargs)
else:
self.set_email(**kwargs)
self._validate_email()
def set_email(self,
host, port,
username=None, password=None,
**kwargs):
"Create a simple default sender"
if "user_name" in kwargs and username is None:
warnings.warn("Argument user_name is replaced with username. Please use username instead.", FutureWarning)
username = kwargs.pop("user_name")
self.email = EmailSender(
host=host, port=port,
username=username, password=password
)
self._set_email_kwargs(kwargs)
def get_subject(self, record):
"Format subject of the email sender"
return self.email.subject.format(
record=record, handler=self
)
def _set_email_kwargs(self, kwargs:dict):
for attr, value in kwargs.items():
if not hasattr(self.email, attr):
raise AttributeError(f"EmailSender has no attribute {attr}")
setattr(self.email, attr, value)
# Set default message body if nothing specified
has_no_body = (
self.email.text is None
and self.email.text_template is None
and self.email.html is None
and self.email.html_template is None
)
if has_no_body:
self.email.text = self.default_text
def _validate_email(self):
"Validate the email has all required attributes for logging"
req_attrs = ('host', 'port', 'subject', 'receivers')
missing = []
for attr in req_attrs:
if getattr(self.email, attr) is None:
missing.append(attr)
if missing:
cls_name = type(self).__name__
raise TypeError(f'{cls_name} email sender missing attributes: {missing}')
class EmailHandler(_EmailHandlerMixin, Handler):
"""Logging handler for sending a log record as an email
Parameters
----------
level : int
Log level of the handler
email : EmailSender
Sender instance to be used for sending
the log records.
kwargs : dict
Keyword arguments for creating the
sender if ``email`` was not passed.
Examples
--------
Minimal example:
.. code-block:: python
handler = EmailHandler(
host="smtp.myhost.com", port=0,
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.email = "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(email=email)
import logging
logger = logging.getLogger()
logger.addHandler(handler)
"""
email: EmailSender
default_text = "{{ msg }}"
def __init__(self, level:int=logging.NOTSET, email:EmailSender=None, **kwargs):
_EmailHandlerMixin.__init__(self, email=email, kwargs=kwargs)
Handler.__init__(self, level)
def emit(self, record:logging.LogRecord):
"Emit a record (send email)"
self.email.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
email : EmailSender
Sender instance to be used for sending
the log records.
kwargs : dict
Keyword arguments for creating the
sender if ``email`` was not passed.
Examples
--------
Minimal example:
.. code-block:: python
handler = MultiEmailHandler(
host="smtp.myhost.com", port=0,
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)
import logging
logger = logging.getLogger()
logger.addHandler(handler)
"""
default_text = dedent("""
Log Recods:
{% for record in records -%}
{{ handler.format(record) }}
{% endfor %}""")[1:]
def __init__(self, capacity:Optional[int]=None, email:EmailSender=None, **kwargs):
_EmailHandlerMixin.__init__(self, email=email, kwargs=kwargs)
BufferingHandler.__init__(self, capacity)
def flush(self):
"Flush the records (send an email)"
self.acquire()
try:
msgs = []
for rec in self.buffer:
# This creates msg, exc_text etc. to the LogRecords
msgs.append(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.email.send(
subject=self.get_subject(self.buffer),
body_params={
"records": self.buffer,
"msgs": msgs,
"handler": self
}
)
self.buffer = []
finally:
self.release()
def shouldFlush(self, record):
"""Should the handler flush its buffer?
Returns true if the buffer is up to capacity. This method can be overridden to implement custom flushing strategies.
"""
if self.capacity is None:
# Only manual flushing
return False
else:
return super().shouldFlush(record)
def get_subject(self, records:List[LogRecord]):
"Get subject of the email"
if records:
min_level = min([record.levelno for record in records])
max_level = max([record.levelno for record in records])
fmt_kwds = {
"min_level_name": logging.getLevelName(min_level),
"max_level_name": logging.getLevelName(max_level),
}
else:
# No log records, getting something
fmt_kwds = {
"min_level_name": logging.getLevelName(logging.NOTSET),
"max_level_name": logging.getLevelName(logging.NOTSET),
}
return self.email.subject.format(
**fmt_kwds,
handler=self,
records=records
)