Deprecate smtp configs in airflow settings / local_settings

airflow/providers/smtp/notifications/smtp.py

@@ -25,30 +25,12 @@
 from airflow.notifications.basenotifier import BaseNotifier
 from airflow.providers.smtp.hooks.smtp import SmtpHook
 
-try:
-    from airflow.settings import SMTP_DEFAULT_TEMPLATED_HTML_CONTENT_PATH, SMTP_DEFAULT_TEMPLATED_SUBJECT
-except ImportError:
-    # This is a fallback for when the settings are not available - they were only added in 2.8.1,
-    # so we should be able to remove it when min airflow version for the SMTP provider is 2.9.0
-    # we do not raise deprecation warning here, because the user might be using 2.8.0 and the new provider
-    # deliberately, and we do not want to upgrade to newer version of Airflow so we should not raise the
-    # deprecation warning here. If the user will modify the settings in local_settings even for earlier
-    # versions of Airflow, they will be properly used as they will be imported above
-    SMTP_DEFAULT_TEMPLATED_HTML_CONTENT_PATH = (Path(__file__).parent / "templates" / "email.html").as_posix()
-    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 %}
-"""
-
 
 class SmtpNotifier(BaseNotifier):
     """
     SMTP Notifier.
 
-    Accepts keyword arguments. The only required argument is `to`. Examples:
+    Accepts keyword arguments. The only required arguments are `from_email` and `to`. Examples:
 
     .. code-block:: python
 
@@ -63,7 +45,10 @@ class SmtpNotifier(BaseNotifier):
             ),
         )
 
-    Default template is defined in airflow.settings but can be overridden in local_settings.py
+    Default template can be overridden via the following provider configuration data:
+        - templated_email_subject_path
+        - templated_html_content_path
+
 
     :param smtp_conn_id: The :ref:`smtp connection id <howto/connection:smtp>`
         that contains the information used to authenticate the client.
@@ -104,22 +89,34 @@ def __init__(
         self.smtp_conn_id = smtp_conn_id
         self.from_email = from_email or conf.get("smtp", "smtp_mail_from")
         self.to = to
-        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
 
+        smtp_default_templated_subject_path = conf.get(
+            "smtp",
+            "templated_email_subject_path",
+            fallback=(Path(__file__).parent / "templates" / "email_subject.jinja2").as_posix(),
+        )
+        self.subject = (
+            subject or Path(smtp_default_templated_subject_path).read_text().replace("\n", "").strip()
+        )
         # 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()
+            smtp_default_templated_html_content_path = conf.get(
+                "smtp",
+                "templated_html_content_path",
+                fallback=(Path(__file__).parent / "templates" / "email.html").as_posix(),
+            )
+            self.html_content = Path(smtp_default_templated_html_content_path).read_text()
 
     @cached_property
     def hook(self) -> SmtpHook:

airflow/providers/smtp/notifications/templates/email_subject.jinja2

@@ -0,0 +1,23 @@
+{#
+ 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.
+#}
+{% 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 %}

airflow/providers/smtp/provider.yaml

@@ -87,3 +87,19 @@ config:
         version_added: 1.3.0
         example: "default"
         default: ~
+      templated_email_subject_path:
+        description: |
+          Allows overriding of the standard templated email subject line when the SmtpNotifier is used.
+          Must provide a path to the template.
+        type: string
+        version_added: 1.6.1
+        example: "path/to/override/email_subject.html"
+        default: ~
+      templated_html_content_path:
+        description: |
+          Allows overriding of the standard templated email path when the SmtpNotifier is used. Must provide
+          a path to the template.
+        type: string
+        version_added: 1.6.1
+        example: "path/to/override/email.html"
+        default: ~

airflow/settings.py

@@ -621,19 +621,6 @@ 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.
 _ENABLE_AIP_44: bool = os.environ.get("AIRFLOW_ENABLE_AIP_44", "false").lower() in {

tests/providers/smtp/notifications/test_smtp.py

@@ -17,6 +17,7 @@
 
 from __future__ import annotations
 
+import tempfile
 from unittest import mock
 
 import pytest
@@ -30,6 +31,7 @@
     send_smtp_notification,
 )
 from airflow.utils import timezone
+from tests.test_utils.config import conf_vars
 
 pytestmark = pytest.mark.db_test
 
@@ -166,3 +168,42 @@ def test_notifier_with_defaults_sla(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_nondefault_conf_vars(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}
+
+        with tempfile.NamedTemporaryFile(mode="wt", suffix=".txt") as f_subject, tempfile.NamedTemporaryFile(
+            mode="wt", suffix=".txt"
+        ) as f_content:
+            f_subject.write("Task {{ ti.task_id }} failed")
+            f_subject.flush()
+
+            f_content.write("Mock content goes here")
+            f_content.flush()
+
+            with conf_vars(
+                {
+                    ("smtp", "templated_html_content_path"): f_content.name,
+                    ("smtp", "templated_email_subject_path"): f_subject.name,
+                }
+            ):
+                notifier = SmtpNotifier(
+                    from_email=conf.get("smtp", "smtp_mail_from"),
+                    to="[email protected]",
+                )
+                notifier(context)
+                mock_smtphook_hook.return_value.__enter__().send_email_smtp.assert_called_once_with(
+                    from_email=conf.get("smtp", "smtp_mail_from"),
+                    to="[email protected]",
+                    subject="Task op failed",
+                    html_content="Mock content goes here",
+                    smtp_conn_id="smtp_default",
+                    files=None,
+                    cc=None,
+                    bcc=None,
+                    mime_subtype="mixed",
+                    mime_charset="utf-8",
+                    custom_headers=None,
+                )