diff --git a/airflow/providers/smtp/notifications/smtp.py b/airflow/providers/smtp/notifications/smtp.py index 88262dc876836..ce977152e2aeb 100644 --- a/airflow/providers/smtp/notifications/smtp.py +++ b/airflow/providers/smtp/notifications/smtp.py @@ -18,8 +18,10 @@ from __future__ import annotations from functools import cached_property +from pathlib import Path from typing import Any, Iterable +from airflow.configuration import conf from airflow.notifications.basenotifier import BaseNotifier from airflow.providers.smtp.hooks.smtp import SmtpHook @@ -28,6 +30,23 @@ class SmtpNotifier(BaseNotifier): """ SMTP Notifier. + Accepts keyword arguments. The only required argument is `to`. Examples: + + .. code-block:: python + + EmptyOperator(task_id="task", on_failure_callback=SmtpNotifier(from_email=None, to="myemail@myemail.com")) + + EmptyOperator( + task_id="task", + on_failure_callback=SmtpNotifier( + from_email="myemail@myemail.com", + to="myemail@myemail.com", + subject="Task {{ ti.task_id }} failed", + ), + ) + + Default template is defined in airflow.settings but can be overridden in local_settings.py + :param smtp_conn_id: The :ref:`smtp connection id ` that contains the information used to authenticate the client. """ @@ -47,10 +66,12 @@ class SmtpNotifier(BaseNotifier): def __init__( self, + # TODO: Move from_email to keyword parameter in next major release so that users do not + # need to specify from_email. No argument here will lead to defaults from conf being used. from_email: str | None, to: str | Iterable[str], - subject: str, - html_content: str, + subject: str | None = None, + html_content: str | None = None, files: list[str] | None = None, cc: str | Iterable[str] | None = None, bcc: str | Iterable[str] | None = None, @@ -58,13 +79,16 @@ def __init__( mime_charset: str = "utf-8", custom_headers: dict[str, Any] | None = None, smtp_conn_id: str = SmtpHook.default_conn_name, + *, + template: str | None = None, ): + from airflow.settings import SMTP_DEFAULT_TEMPLATED_HTML_CONTENT_PATH, SMTP_DEFAULT_TEMPLATED_SUBJECT + super().__init__() self.smtp_conn_id = smtp_conn_id - self.from_email = from_email + self.from_email = from_email or conf.get("smtp", "smtp_mail_from") self.to = to - self.subject = subject - self.html_content = html_content + self.subject = subject or SMTP_DEFAULT_TEMPLATED_SUBJECT.replace("\n", "").strip() self.files = files self.cc = cc self.bcc = bcc @@ -72,6 +96,15 @@ def __init__( self.mime_charset = mime_charset self.custom_headers = custom_headers + # If html_content is passed, prioritize it. Otherwise, if template is passed, use + # it to populate html_content. Else, fall back to defaults defined in settings + if html_content is not None: + self.html_content = html_content + elif template is not None: + self.html_content = Path(template).read_text() + else: + self.html_content = Path(SMTP_DEFAULT_TEMPLATED_HTML_CONTENT_PATH).read_text() + @cached_property def hook(self) -> SmtpHook: """Smtp Events Hook.""" diff --git a/airflow/providers/smtp/notifications/templates/__init__.py b/airflow/providers/smtp/notifications/templates/__init__.py new file mode 100644 index 0000000000000..13a83393a9124 --- /dev/null +++ b/airflow/providers/smtp/notifications/templates/__init__.py @@ -0,0 +1,16 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. diff --git a/airflow/providers/smtp/notifications/templates/email.html b/airflow/providers/smtp/notifications/templates/email.html new file mode 100644 index 0000000000000..8046a99d71863 --- /dev/null +++ b/airflow/providers/smtp/notifications/templates/email.html @@ -0,0 +1,58 @@ + + + + + + + + + {% if ti is defined %} + + + + + + + + + + + + + + + + + + + + + + + + + {% elif slas is defined %} + + + + + + + + + + + + + + + + + + + + + {% endif %} +
Run ID:{{ ti.run_id }}
Try:{{ ti.try_number }} of {{ ti.max_tries + 1 }}
Task State:{{ ti.state }}
Host:{{ ti.hostname }}
Log Link:{{ ti.log_url }}
Mark Success Link:{{ ti.mark_success_url }}
Dag:{{ dag.dag_id }}
Task List:{{ task_list }}
Blocking Task List:{{ blocking_task_list }}
SLAs:{{ slas }}
Blocking TI's{{ blocking_tis }}
+ + diff --git a/airflow/settings.py b/airflow/settings.py index 1a38a59ed301c..e86a8d557348b 100644 --- a/airflow/settings.py +++ b/airflow/settings.py @@ -629,6 +629,18 @@ def initialize(): DAEMON_UMASK: str = conf.get("core", "daemon_umask", fallback="0o077") +SMTP_DEFAULT_TEMPLATED_SUBJECT = """ +{% if ti is defined %} +DAG {{ ti.dag_id }} - Task {{ ti.task_id }} - Run ID {{ ti.run_id }} in State {{ ti.state }} +{% elif slas is defined %} +SLA Missed for DAG {{ dag.dag_id }} - Task {{ slas[0].task_id }} +{% endif %} +""" + +SMTP_DEFAULT_TEMPLATED_HTML_CONTENT_PATH = os.path.join( + os.path.dirname(__file__), "providers", "smtp", "notifications", "templates", "email.html" +) + # AIP-44: internal_api (experimental) # This feature is not complete yet, so we disable it by default. diff --git a/setup.py b/setup.py index ee60c885207ae..9dad27f45e928 100644 --- a/setup.py +++ b/setup.py @@ -902,6 +902,8 @@ def parse_config_files(self, *args, **kwargs) -> None: self.package_data["airflow"].append(provider_relative_path) # Add python_kubernetes_script.jinja2 to package data self.package_data["airflow"].append("providers/cncf/kubernetes/python_kubernetes_script.jinja2") + # Add default email template to package data + self.package_data["airflow"].append("providers/smtp/notifications/templates/email.html") else: self.install_requires.extend( [ diff --git a/tests/providers/smtp/notifications/test_smtp.py b/tests/providers/smtp/notifications/test_smtp.py index 36c1ac780decd..0bbe14622688c 100644 --- a/tests/providers/smtp/notifications/test_smtp.py +++ b/tests/providers/smtp/notifications/test_smtp.py @@ -21,19 +21,22 @@ import pytest +from airflow.configuration import conf +from airflow.models import SlaMiss from airflow.operators.empty import EmptyOperator from airflow.providers.smtp.hooks.smtp import SmtpHook from airflow.providers.smtp.notifications.smtp import ( SmtpNotifier, send_smtp_notification, ) +from airflow.utils import timezone pytestmark = pytest.mark.db_test SMTP_API_DEFAULT_CONN_ID = SmtpHook.default_conn_name -class TestPagerdutyNotifier: +class TestSmtpNotifier: @mock.patch("airflow.providers.smtp.notifications.smtp.SmtpHook") def test_notifier(self, mock_smtphook_hook, dag_maker): with dag_maker("test_notifier") as dag: @@ -110,3 +113,56 @@ def test_notifier_templated(self, mock_smtphook_hook, dag_maker): mime_charset="utf-8", custom_headers=None, ) + + @mock.patch("airflow.providers.smtp.notifications.smtp.SmtpHook") + def test_notifier_with_defaults(self, mock_smtphook_hook, create_task_instance): + ti = create_task_instance(dag_id="dag", task_id="op", execution_date=timezone.datetime(2018, 1, 1)) + context = {"dag": ti.dag_run.dag, "ti": ti} + notifier = SmtpNotifier( + from_email=conf.get("smtp", "smtp_mail_from"), + to="test_reciver@test.com", + ) + notifier(context) + mock_smtphook_hook.return_value.__enter__().send_email_smtp.assert_called_once_with( + from_email=conf.get("smtp", "smtp_mail_from"), + to="test_reciver@test.com", + subject="DAG dag - Task op - Run ID test in State None", + html_content="""\n\n \n \n \n \n\n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n
Run ID:test
Try:1 of 1
Task State:None
Host:
Log Link:http://localhost:8080/log?execution_date=2018-01-01T00%3A00%3A00%2B00%3A00&task_id=op&dag_id=dag&map_index=-1
Mark Success Link:http://localhost:8080/confirm?task_id=op&dag_id=dag&dag_run_id=test&upstream=false&downstream=false&state=success
\n\n""", + smtp_conn_id="smtp_default", + files=None, + cc=None, + bcc=None, + mime_subtype="mixed", + mime_charset="utf-8", + custom_headers=None, + ) + + @mock.patch("airflow.providers.smtp.notifications.smtp.SmtpHook") + def test_notifier_with_defaults_sla(self, mock_smtphook_hook, dag_maker): + with dag_maker("test_notifier") as dag: + EmptyOperator(task_id="task1") + context = { + "dag": dag, + "slas": [SlaMiss(task_id="op", dag_id=dag.dag_id, execution_date=timezone.datetime(2018, 1, 1))], + "task_list": [], + "blocking_task_list": [], + "blocking_tis": [], + } + notifier = SmtpNotifier( + from_email=conf.get("smtp", "smtp_mail_from"), + to="test_reciver@test.com", + ) + notifier(context) + mock_smtphook_hook.return_value.__enter__().send_email_smtp.assert_called_once_with( + from_email=conf.get("smtp", "smtp_mail_from"), + to="test_reciver@test.com", + subject="SLA Missed for DAG test_notifier - Task op", + html_content="""\n\n \n \n \n \n\n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n \n
Dag:test_notifier
Task List:[]
Blocking Task List:[]
SLAs:[(\'test_notifier\', \'op\', \'2018-01-01T00:00:00+00:00\')]
Blocking TI\'s[]
\n\n""", + smtp_conn_id="smtp_default", + files=None, + cc=None, + bcc=None, + mime_subtype="mixed", + mime_charset="utf-8", + custom_headers=None, + )