mirror of
https://github.com/datafolklabs/cement.git
synced 2026-02-06 11:37:06 +00:00
235 lines
7.7 KiB
Python
235 lines
7.7 KiB
Python
"""
|
|
Cement smtp extension module.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
import os
|
|
import smtplib
|
|
from email.header import Header
|
|
from email.mime.multipart import MIMEMultipart
|
|
from email.mime.base import MIMEBase
|
|
from email.mime.text import MIMEText
|
|
from email import encoders
|
|
from typing import Any, Dict, Union, Tuple, TYPE_CHECKING
|
|
from ..core import mail
|
|
from ..utils import fs
|
|
from ..utils.misc import minimal_logger, is_true
|
|
|
|
if TYPE_CHECKING:
|
|
from ..core.foundation import App # pragma: nocover
|
|
|
|
LOG = minimal_logger(__name__)
|
|
|
|
|
|
class SMTPMailHandler(mail.MailHandler):
|
|
|
|
"""
|
|
This class implements the :ref:`IMail <cement.core.mail>`
|
|
interface, and is based on the `smtplib
|
|
<http://docs.python.org/dev/library/smtplib.html>`_ standard library.
|
|
|
|
"""
|
|
|
|
class Meta(mail.MailHandler.Meta):
|
|
|
|
"""Handler meta-data."""
|
|
|
|
#: Unique identifier for this handler
|
|
label = 'smtp'
|
|
|
|
#: Configuration default values
|
|
config_defaults = {
|
|
'to': [],
|
|
'from_addr': 'noreply@localhost',
|
|
'cc': [],
|
|
'bcc': [],
|
|
'subject': None,
|
|
'subject_prefix': None,
|
|
'host': 'localhost',
|
|
'port': '25',
|
|
'timeout': 30,
|
|
'ssl': False,
|
|
'tls': False,
|
|
'auth': False,
|
|
'username': None,
|
|
'password': None,
|
|
'files': None,
|
|
}
|
|
|
|
_meta: Meta # type: ignore
|
|
|
|
def _get_params(self, **kw: Any) -> Dict[str, Any]:
|
|
params = dict()
|
|
|
|
# some keyword args override configuration defaults
|
|
for item in ['to', 'from_addr', 'cc', 'bcc', 'subject',
|
|
'subject_prefix', 'files']:
|
|
config_item = self.app.config.get(self._meta.config_section, item)
|
|
params[item] = kw.get(item, config_item)
|
|
|
|
# others don't
|
|
other_params = ['ssl', 'tls', 'host', 'port', 'auth', 'username',
|
|
'password', 'timeout']
|
|
for item in other_params:
|
|
params[item] = self.app.config.get(self._meta.config_section,
|
|
item)
|
|
|
|
return params
|
|
|
|
def send(self, body: Union[str, Tuple[str, str]], **kw: Any) -> bool:
|
|
"""
|
|
Send an email message via SMTP. Keyword arguments override
|
|
configuration defaults (cc, bcc, etc).
|
|
|
|
Args:
|
|
body (tuple): The message body to send. Tuple is treated as:
|
|
``(<text>, <html>)``. If a single string is passed it will be
|
|
converted to ``(<text>)``. At minimum, a text version is
|
|
required.
|
|
|
|
Keyword Args:
|
|
to (list): List of recipients (generally email addresses)
|
|
from_addr (str): Address (generally email) of the sender
|
|
cc (list): List of CC Recipients
|
|
bcc (list): List of BCC Recipients
|
|
subject (str): Message subject line
|
|
subject_prefix (str): Prefix for message subject line (useful to
|
|
override if you want to remove/change the default prefix).
|
|
files (list): List of file paths to attach to the message. Can be
|
|
``[ '/path/to/file.ext', ... ]`` or alternative filename can
|
|
be defined by passing a list of tuples in the form of
|
|
``[ ('alt-name.ext', '/path/to/file.ext'), ...]``
|
|
|
|
Returns:
|
|
bool:``True`` if message is sent successfully, ``False`` otherwise
|
|
|
|
Example:
|
|
|
|
.. code-block:: python
|
|
|
|
# Using all configuration defaults
|
|
app.mail.send('This is my message body')
|
|
|
|
# Overriding configuration defaults
|
|
app.mail.send('My message body'
|
|
from_addr='me@example.com',
|
|
to=['john@example.com'],
|
|
cc=['jane@example.com', 'rita@example.com'],
|
|
subject='This is my subject',
|
|
)
|
|
|
|
"""
|
|
params = self._get_params(**kw)
|
|
|
|
if is_true(params['ssl']):
|
|
server = smtplib.SMTP_SSL(params['host'],
|
|
params['port'],
|
|
params['timeout'])
|
|
LOG.debug(f"{self._meta.label} : initiating smtp over ssl")
|
|
|
|
else:
|
|
server = smtplib.SMTP(params['host'], # type: ignore
|
|
params['port'],
|
|
params['timeout'])
|
|
LOG.debug(f"{self._meta.label} : initiating smtp")
|
|
|
|
if self.app.debug is True:
|
|
server.set_debuglevel(9)
|
|
|
|
if is_true(params['tls']):
|
|
LOG.debug(f"{self._meta.label} : initiating tls")
|
|
server.starttls()
|
|
|
|
if is_true(params['auth']):
|
|
server.login(params['username'], params['password'])
|
|
|
|
res = self._send_message(server, body, **params)
|
|
server.quit()
|
|
return res
|
|
|
|
def _send_message(self,
|
|
server: Union[smtplib.SMTP, smtplib.SMTP_SSL],
|
|
body: Union[str, Tuple[str, str]],
|
|
**params: Any) -> bool:
|
|
msg = MIMEMultipart('alternative')
|
|
msg.set_charset('utf-8')
|
|
|
|
msg['From'] = params['from_addr']
|
|
msg['To'] = ', '.join(params['to'])
|
|
if params['cc']:
|
|
msg['Cc'] = ', '.join(params['cc'])
|
|
if params['bcc']:
|
|
msg['Bcc'] = ', '.join(params['bcc'])
|
|
if params['subject_prefix'] not in [None, '']:
|
|
subject = f"{params['subject_prefix']} {params['subject']}"
|
|
else:
|
|
subject = params['subject']
|
|
msg['Subject'] = Header(subject) # type: ignore
|
|
|
|
# add body as text and/or as html
|
|
partText = None
|
|
partHtml = None
|
|
|
|
if type(body) not in [str, tuple]:
|
|
error_msg = "Message body must be string or tuple " \
|
|
"('<text>', '<html>')"
|
|
raise TypeError(error_msg)
|
|
|
|
if isinstance(body, str):
|
|
partText = MIMEText(body)
|
|
elif isinstance(body, tuple):
|
|
# handle plain text
|
|
if len(body) >= 1:
|
|
partText = MIMEText(body[0], 'plain')
|
|
|
|
# handle html
|
|
if len(body) >= 2:
|
|
partHtml = MIMEText(body[1], 'html')
|
|
|
|
if partText:
|
|
msg.attach(partText)
|
|
if partHtml:
|
|
msg.attach(partHtml)
|
|
|
|
# attach files
|
|
if params['files']:
|
|
for in_path in params['files']:
|
|
part = MIMEBase('application', 'octet-stream')
|
|
|
|
# support for alternative file name if it's tuple
|
|
# like ('alt-name.ext', '/path/to/file.ext')
|
|
if isinstance(in_path, tuple):
|
|
if in_path[0] == in_path[1]:
|
|
# protect against the full path being passed in
|
|
alt_name = os.path.basename(in_path[0])
|
|
else:
|
|
alt_name = in_path[0]
|
|
path = in_path[1]
|
|
else:
|
|
alt_name = os.path.basename(in_path)
|
|
path = in_path
|
|
|
|
path = fs.abspath(path)
|
|
|
|
# add attachment
|
|
with open(path, 'rb') as file:
|
|
part.set_payload(file.read())
|
|
|
|
# encode and name
|
|
encoders.encode_base64(part)
|
|
part.add_header(
|
|
'Content-Disposition',
|
|
f'attachment; filename={alt_name}',
|
|
)
|
|
msg.attach(part)
|
|
|
|
server.send_message(msg)
|
|
|
|
# FIXME: how to check success? docs don't say return type
|
|
# - `[ext.scrub]` [Issue #724](https://github.com/datafolklabs/cement/issues/724)
|
|
return True
|
|
|
|
|
|
def load(app: App) -> None:
|
|
app.handler.register(SMTPMailHandler)
|