""" 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 ` interface, and is based on the `smtplib `_ 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: ``(, )``. If a single string is passed it will be converted to ``()``. 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 " \ "('', '')" 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 its 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)