Source code for weaver.notify

import binascii
import hashlib
import logging
import os
import smtplib
from typing import TYPE_CHECKING

from mako.template import Template
from pyramid.settings import asbool

from weaver.datatype import Job
from weaver.utils import bytes2str, get_settings, str2bytes

    from weaver.typedefs import AnySettingsContainer

[docs]LOGGER = logging.getLogger(__name__)
__DEFAULT_TEMPLATE__ = """ <%doc> This is an example notification message to be sent by email when a job is done. It is formatted using the Mako template library ( The content must also include the message header. The provided variables are: to: Recipient's address job: weaver.datatype.Job object settings: application settings And every variable returned by the `weaver.datatype.Job.json` method: status: succeeded, failed logs: url to the logs jobID: example "617f23d3-f474-47f9-a8ec-55da9dd6ac71" result: url to the outputs duration: example "0:01:02" message: example "Job succeeded." percentCompleted: example 100 </%doc> From: Weaver To: ${to} Subject: Job ${job.process} ${job.status.title()} Content-Type: text/plain; charset=UTF-8 Dear user, Your job submitted on ${job.created.strftime("%Y/%m/%d %H:%M %Z")} to ${settings.get("weaver.url")} ${job.status}. % if job.status == "succeeded": You can retrieve the output(s) at the following link: ${job.results[0]["reference"]} % endif The logs are available here: ${logs} Regards, Weaver """
[docs]def notify_job_complete(job, to_email_recipient, container): # type: (Job, str, AnySettingsContainer) -> None """ Send email notification of a job completion. """ settings = get_settings(container) smtp_host = settings.get("weaver.wps_email_notify_smtp_host") from_addr = settings.get("weaver.wps_email_notify_from_addr") password = settings.get("weaver.wps_email_notify_password") port = settings.get("weaver.wps_email_notify_port") ssl = asbool(settings.get("weaver.wps_email_notify_ssl", True)) # an example template is located in # weaver/wps_restapi/templates/notification_email_example.mako template_dir = settings.get("weaver.wps_email_notify_template_dir") if not smtp_host or not port: raise ValueError("The email server configuration is missing.") # find appropriate template according to settings if not os.path.isdir(template_dir): LOGGER.warning("No default email template directory configured. Using default format.") template = Template(text=__DEFAULT_TEMPLATE__) # nosec: B702 else: default_name = settings.get("weaver.wps_email_notify_template_default", "default.mako") process_name = f"{job.process!s}.mako" default_template = os.path.join(template_dir, default_name) process_template = os.path.join(template_dir, process_name) if os.path.isfile(process_template): template = Template(filename=process_template) # nosec: B702 elif os.path.isfile(default_template): template = Template(filename=default_template) # nosec: B702 else: raise IOError(f"Template file doesn't exist: OneOf[{process_name!s}, {default_name!s}]") job_json = job.json(settings) contents = template.render(to=to_email_recipient, job=job, settings=settings, **job_json) message = f"{contents}".strip("\n") if ssl: server = smtplib.SMTP_SSL(smtp_host, port) else: server = smtplib.SMTP(smtp_host, port) server.ehlo() try: server.starttls() server.ehlo() except smtplib.SMTPException: pass try: if password: server.login(from_addr, password) result = server.sendmail(from_addr, to_email_recipient, message.encode("utf8")) finally: server.close() if result: code, error_message = result[to_email_recipient] raise IOError(f"Code: {code}, Message: {error_message}")
[docs]def encrypt_email(email, settings): if not email or not isinstance(email, str): raise TypeError(f"Invalid email: {email!s}") LOGGER.debug("Job email setup.") try: salt = str2bytes(settings.get("weaver.wps_email_encrypt_salt")) email = str2bytes(email) rounds = int(settings.get("weaver.wps_email_encrypt_rounds", 100000)) derived_key = hashlib.pbkdf2_hmac("sha256", email, salt, rounds) return bytes2str(binascii.hexlify(derived_key)) except Exception as ex: LOGGER.debug("Job email setup failed [%r].", ex) raise ValueError("Cannot register job, server not properly configured for notification email.")