apache · potiuk · Dec 21, 2023 · Dec 14, 2023 · Dec 18, 2023 · Dec 18, 2023
@@ -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,19 @@ 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(to="myemail@myemail.com"))
+
+        EmptyOperator(
+            task_id="task",
+            on_failure_callback=SmtpNotifier(to="myemail@myemail.com", from_email="myemail@myemail.com"),
+        )
+
+    Default template is defined in airflow.settings but can be overridden in local_settings.py
+
     :param smtp_conn_id: The :ref:`smtp connection id <howto/connection:smtp>`
         that contains the information used to authenticate the client.
     """
@@ -47,31 +62,43 @@ class SmtpNotifier(BaseNotifier):
 
     def __init__(
         self,
-        from_email: str | None,
+        *,
         to: str | Iterable[str],
-        subject: str,
-        html_content: str,
+        from_email: str | None = None,
+        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,
         mime_subtype: str = "mixed",
         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
         self.mime_subtype = mime_subtype
         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."""

@@ -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.
@@ -0,0 +1,58 @@
+<!DOCTYPE html>
+<html>
+    <head>
+        <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
+        <meta name="viewport" content="width=device-width">
+    </head>
+<body>
+    <table role="presentation">
+        {% if ti is defined %}
+        <tr>
+            <td>Run ID:</td>
+            <td>{{ ti.run_id }}</td>
+        </tr>
+        <tr>
+            <td>Try:</td>
+            <td>{{ ti.try_number }} of {{ ti.max_tries + 1 }}</td>
+        </tr>
+        <tr>
+            <td>Task State:</td>
+            <td>{{ ti.state }}</td>
+        </tr>
+        <tr>
+            <td>Host:</td>
+            <td>{{ ti.hostname }}</td>
+        </tr>
+        <tr>
+            <td>Log Link:</td>
+            <td><a href="{{ ti.log_url }}" style="text-decoration:underline;">{{ ti.log_url }}</a></td>
+        </tr>
+        <tr>
+            <td>Mark Success Link:</td>
+            <td><a href="{{ ti.mark_success_url }}" style="text-decoration:underline;">{{ ti.mark_success_url }}</a></td>
+        </tr>
+        {% elif slas is defined %}
+        <tr>
+            <td>Dag:</td>
+            <td>{{ dag.dag_id }}</td>
+        </tr>
+        <tr>
+            <td>Task List:</td>
+            <td>{{ task_list }}</td>
+        </tr>
+        <tr>
+            <td>Blocking Task List:</td>
+            <td>{{ blocking_task_list }}</td>
+        </tr>
+        <tr>
+            <td>SLAs:</td>
+            <td>{{ slas }}</td>
+        </tr>
+        <tr>
+            <td>Blocking TI's</td>
+            <td>{{ blocking_tis }}</td>
+        </tr>
+        {% endif %}
+    </table>
+</body>
+</html>
diff --git 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
@@ -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
@@ -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,54 @@ 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(
+            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="""<!DOCTYPE html>\n<html>\n    <head>\n        <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />\n        <meta name="viewport" content="width=device-width">\n    </head>\n<body>\n    <table role="presentation">\n        \n        <tr>\n            <td>Run ID:</td>\n            <td>test</td>\n        </tr>\n        <tr>\n            <td>Try:</td>\n            <td>1 of 1</td>\n        </tr>\n        <tr>\n            <td>Task State:</td>\n            <td>None</td>\n        </tr>\n        <tr>\n            <td>Host:</td>\n            <td></td>\n        </tr>\n        <tr>\n            <td>Log Link:</td>\n            <td><a href="http://localhost:8080/log?execution_date=2018-01-01T00%3A00%3A00%2B00%3A00&task_id=op&dag_id=dag&map_index=-1" style="text-decoration:underline;">http://localhost:8080/log?execution_date=2018-01-01T00%3A00%3A00%2B00%3A00&task_id=op&dag_id=dag&map_index=-1</a></td>\n        </tr>\n        <tr>\n            <td>Mark Success Link:</td>\n            <td><a href="http://localhost:8080/confirm?task_id=op&dag_id=dag&dag_run_id=test&upstream=false&downstream=false&state=success" style="text-decoration:underline;">http://localhost:8080/confirm?task_id=op&dag_id=dag&dag_run_id=test&upstream=false&downstream=false&state=success</a></td>\n        </tr>\n        \n    </table>\n</body>\n</html>""",
+            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(
+            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="""<!DOCTYPE html>\n<html>\n    <head>\n        <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />\n        <meta name="viewport" content="width=device-width">\n    </head>\n<body>\n    <table role="presentation">\n        \n        <tr>\n            <td>Dag:</td>\n            <td>test_notifier</td>\n        </tr>\n        <tr>\n            <td>Task List:</td>\n            <td>[]</td>\n        </tr>\n        <tr>\n            <td>Blocking Task List:</td>\n            <td>[]</td>\n        </tr>\n        <tr>\n            <td>SLAs:</td>\n            <td>[(\'test_notifier\', \'op\', \'2018-01-01T00:00:00+00:00\')]</td>\n        </tr>\n        <tr>\n            <td>Blocking TI\'s</td>\n            <td>[]</td>\n        </tr>\n        \n    </table>\n</body>\n</html>""",
+            smtp_conn_id="smtp_default",
+            files=None,
+            cc=None,
+            bcc=None,
+            mime_subtype="mixed",
+            mime_charset="utf-8",
+            custom_headers=None,
+        )