FitTrackee/fittrackee/emails/email.py
2022-04-23 06:11:08 +02:00

142 lines
4.9 KiB
Python

import logging
import smtplib
import ssl
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from typing import Dict, Optional, Type, Union
from flask import Flask
from jinja2 import Environment, FileSystemLoader, select_autoescape
from urllib3.util import parse_url
from .exceptions import InvalidEmailUrlScheme
email_log = logging.getLogger('fittrackee_api_email')
email_log.setLevel(logging.DEBUG)
class EmailMessage:
def __init__(
self, sender: str, recipient: str, subject: str, html: str, text: str
) -> None:
self.sender = sender
self.recipient = recipient
self.subject = subject
self.html = html
self.text = text
def generate_message(self) -> MIMEMultipart:
message = MIMEMultipart('alternative')
message['Subject'] = self.subject
message['From'] = self.sender
message['To'] = self.recipient
part1 = MIMEText(self.text, 'plain')
part2 = MIMEText(self.html, 'html')
message.attach(part1)
message.attach(part2)
return message
class EmailTemplate:
def __init__(self, template_directory: str) -> None:
self._env = Environment(
autoescape=select_autoescape(['html', 'htm', 'xml']),
loader=FileSystemLoader(template_directory),
)
def get_content(
self, template_name: str, lang: str, part: str, data: Dict
) -> str:
template = self._env.get_template(f'{template_name}/{lang}/{part}')
return template.render(data)
def get_all_contents(self, template: str, lang: str, data: Dict) -> Dict:
output = {}
for part in ['subject.txt', 'body.txt', 'body.html']:
output[part] = self.get_content(template, lang, part, data)
return output
def get_message(
self, template: str, lang: str, sender: str, recipient: str, data: Dict
) -> MIMEMultipart:
output = self.get_all_contents(template, lang, data)
message = EmailMessage(
sender,
recipient,
output['subject.txt'],
output['body.html'],
output['body.txt'],
)
return message.generate_message()
class EmailService:
def __init__(self, app: Optional[Flask] = None) -> None:
self.host = 'localhost'
self.port = 25
self.use_tls = False
self.use_ssl = False
self.username = None
self.password = None
self.sender_email = 'no-reply@example.com'
self.email_template: Optional[EmailTemplate] = None
if app is not None:
self.init_email(app)
def init_email(self, app: Flask) -> None:
parsed_url = self.parse_email_url(app.config['EMAIL_URL'])
self.host = parsed_url['host']
self.port = parsed_url['port']
self.use_tls = parsed_url['use_tls']
self.use_ssl = parsed_url['use_ssl']
self.username = parsed_url['username']
self.password = parsed_url['password']
self.sender_email = app.config['SENDER_EMAIL']
self.email_template = EmailTemplate(app.config['TEMPLATES_FOLDER'])
@staticmethod
def parse_email_url(email_url: str) -> Dict:
parsed_url = parse_url(email_url)
if parsed_url.scheme != 'smtp':
raise InvalidEmailUrlScheme()
credentials = (
parsed_url.auth.split(':') if parsed_url.auth else [None, None]
)
return {
'host': parsed_url.host,
'port': 25 if parsed_url.port is None else parsed_url.port,
'use_tls': True if parsed_url.query == 'tls=True' else False,
'use_ssl': True if parsed_url.query == 'ssl=True' else False,
'username': credentials[0],
'password': credentials[1],
}
@property
def smtp(self) -> Type[Union[smtplib.SMTP_SSL, smtplib.SMTP]]:
return smtplib.SMTP_SSL if self.use_ssl else smtplib.SMTP
def send(
self, template: str, lang: str, recipient: str, data: Dict
) -> None:
if not self.email_template:
raise Exception('No email template defined.')
message = self.email_template.get_message(
template, lang, self.sender_email, recipient, data
)
connection_params = {}
if self.use_ssl or self.use_tls:
context = ssl.create_default_context()
if self.use_ssl:
connection_params.update({'context': context})
with self.smtp(
self.host, self.port, **connection_params # type: ignore
) as smtp:
if self.use_tls:
smtp.ehlo()
smtp.starttls(context=context)
smtp.ehlo()
if self.username and self.password:
smtp.login(self.username, self.password) # type: ignore
smtp.sendmail(self.sender_email, recipient, message.as_string())
smtp.quit()