From 7872c3b523326095c80aca0191c998b38402294a Mon Sep 17 00:00:00 2001
From: Ephraim Anierobi <splendidzigy24@gmail.com>
Date: Wed, 21 Dec 2022 10:46:58 +0100
Subject: [PATCH 1/7] Add general-purpose "notifier" concept to DAGs

This makes it easy for users to setup notifications for their DAGs using on_*_callbacks
It's extensible and we can add it to more providers. Implemented a SlackNotifier in this
phase.

In the course of this, I extracted a 'Templater' class from AbstractBaseOperator
and have both the Notifier & ABO inherit from it. This is necessary in other to avoid
code repetition.
---
 airflow/models/abstractoperator.py     | 146 ++-----------------
 airflow/notifications/__init__.py      |  16 +++
 airflow/notifications/notifier.py      | 108 ++++++++++++++
 airflow/providers/slack/notifier.py    |  80 +++++++++++
 airflow/template/__init__.py           |  16 +++
 airflow/template/templater.py          | 187 +++++++++++++++++++++++++
 tests/notifications/__init__.py        |  16 +++
 tests/notifications/test_notifier.py   |  52 +++++++
 tests/notifications/test_notifier.txt  |   1 +
 tests/providers/slack/test_notifier.py | 111 +++++++++++++++
 tests/template/__init__.py             |  16 +++
 tests/template/test_templater.py       |  62 ++++++++
 12 files changed, 679 insertions(+), 132 deletions(-)
 create mode 100644 airflow/notifications/__init__.py
 create mode 100644 airflow/notifications/notifier.py
 create mode 100644 airflow/providers/slack/notifier.py
 create mode 100644 airflow/template/__init__.py
 create mode 100644 airflow/template/templater.py
 create mode 100644 tests/notifications/__init__.py
 create mode 100644 tests/notifications/test_notifier.py
 create mode 100644 tests/notifications/test_notifier.txt
 create mode 100644 tests/providers/slack/test_notifier.py
 create mode 100644 tests/template/__init__.py
 create mode 100644 tests/template/test_templater.py

diff --git a/airflow/models/abstractoperator.py b/airflow/models/abstractoperator.py
index d8f901c744a7d..d952f68acd92a 100644
--- a/airflow/models/abstractoperator.py
+++ b/airflow/models/abstractoperator.py
@@ -26,10 +26,8 @@
 from airflow.exceptions import AirflowException
 from airflow.models.expandinput import NotFullyPopulated
 from airflow.models.taskmixin import DAGNode
+from airflow.template.templater import Templater
 from airflow.utils.context import Context
-from airflow.utils.helpers import render_template_as_native, render_template_to_string
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.mixins import ResolveMixin
 from airflow.utils.session import NEW_SESSION, provide_session
 from airflow.utils.state import State, TaskInstanceState
 from airflow.utils.task_group import MappedTaskGroup
@@ -75,7 +73,7 @@ class NotMapped(Exception):
     """Raise if a task is neither mapped nor has any parent mapped groups."""
 
 
-class AbstractOperator(LoggingMixin, DAGNode):
+class AbstractOperator(Templater, DAGNode):
     """Common implementation for operators, including unmapped and mapped.
 
     This base class is more about sharing implementations, not defining a common
@@ -95,10 +93,6 @@ class AbstractOperator(LoggingMixin, DAGNode):
 
     # Defines the operator level extra links.
     operator_extra_links: Collection[BaseOperatorLink]
-    # For derived classes to define which fields will get jinjaified.
-    template_fields: Collection[str]
-    # Defines which files extensions to look for in the templated fields.
-    template_ext: Sequence[str]
 
     owner: str
     task_id: str
@@ -152,48 +146,6 @@ def dag_id(self) -> str:
     def node_id(self) -> str:
         return self.task_id
 
-    def get_template_env(self) -> jinja2.Environment:
-        """Fetch a Jinja template environment from the DAG or instantiate empty environment if no DAG."""
-        # This is imported locally since Jinja2 is heavy and we don't need it
-        # for most of the functionalities. It is imported by get_template_env()
-        # though, so we don't need to put this after the 'if dag' check.
-        from airflow.templates import SandboxedEnvironment
-
-        dag = self.get_dag()
-        if dag:
-            return dag.get_template_env(force_sandboxed=False)
-        return SandboxedEnvironment(cache_size=0)
-
-    def prepare_template(self) -> None:
-        """Hook triggered after the templated fields get replaced by their content.
-
-        If you need your operator to alter the content of the file before the
-        template is rendered, it should override this method to do so.
-        """
-
-    def resolve_template_files(self) -> None:
-        """Getting the content of files for template_field / template_ext."""
-        if self.template_ext:
-            for field in self.template_fields:
-                content = getattr(self, field, None)
-                if content is None:
-                    continue
-                elif isinstance(content, str) and any(content.endswith(ext) for ext in self.template_ext):
-                    env = self.get_template_env()
-                    try:
-                        setattr(self, field, env.loader.get_source(env, content)[0])  # type: ignore
-                    except Exception:
-                        self.log.exception("Failed to resolve template field %r", field)
-                elif isinstance(content, list):
-                    env = self.get_template_env()
-                    for i, item in enumerate(content):
-                        if isinstance(item, str) and any(item.endswith(ext) for ext in self.template_ext):
-                            try:
-                                content[i] = env.loader.get_source(env, item)[0]  # type: ignore
-                            except Exception:
-                                self.log.exception("Failed to get source %s", item)
-        self.prepare_template()
-
     def get_direct_relative_ids(self, upstream: bool = False) -> set[str]:
         """Get direct relative IDs to the current task, upstream or downstream."""
         if upstream:
@@ -577,6 +529,17 @@ def render_template_fields(
         """
         raise NotImplementedError()
 
+    def _render(self, template, context, dag: DAG | None = None):
+        if dag is None:
+            dag = self.get_dag()
+        return super()._render(template, context, dag=dag)
+
+    def get_template_env(self, dag: DAG | None = None) -> jinja2.Environment:
+        """Get the template environment for rendering templates."""
+        if dag is None:
+            dag = self.get_dag()
+        return super().get_template_env(dag=dag)
+
     @provide_session
     def _do_render_template_fields(
         self,
@@ -588,6 +551,7 @@ def _do_render_template_fields(
         *,
         session: Session = NEW_SESSION,
     ) -> None:
+        """Override the base to use custom error logging."""
         for attr_name in template_fields:
             try:
                 value = getattr(parent, attr_name)
@@ -615,85 +579,3 @@ def _do_render_template_fields(
                 raise
             else:
                 setattr(parent, attr_name, rendered_content)
-
-    def render_template(
-        self,
-        content: Any,
-        context: Context,
-        jinja_env: jinja2.Environment | None = None,
-        seen_oids: set[int] | None = None,
-    ) -> Any:
-        """Render a templated string.
-
-        If *content* is a collection holding multiple templated strings, strings
-        in the collection will be templated recursively.
-
-        :param content: Content to template. Only strings can be templated (may
-            be inside a collection).
-        :param context: Dict with values to apply on templated content
-        :param jinja_env: Jinja environment. Can be provided to avoid
-            re-creating Jinja environments during recursion.
-        :param seen_oids: template fields already rendered (to avoid
-            *RecursionError* on circular dependencies)
-        :return: Templated content
-        """
-        # "content" is a bad name, but we're stuck to it being public API.
-        value = content
-        del content
-
-        if seen_oids is not None:
-            oids = seen_oids
-        else:
-            oids = set()
-
-        if id(value) in oids:
-            return value
-
-        if not jinja_env:
-            jinja_env = self.get_template_env()
-
-        if isinstance(value, str):
-            if any(value.endswith(ext) for ext in self.template_ext):  # A filepath.
-                template = jinja_env.get_template(value)
-            else:
-                template = jinja_env.from_string(value)
-            dag = self.get_dag()
-            if dag and dag.render_template_as_native_obj:
-                return render_template_as_native(template, context)
-            return render_template_to_string(template, context)
-
-        if isinstance(value, ResolveMixin):
-            return value.resolve(context)
-
-        # Fast path for common built-in collections.
-        if value.__class__ is tuple:
-            return tuple(self.render_template(element, context, jinja_env, oids) for element in value)
-        elif isinstance(value, tuple):  # Special case for named tuples.
-            return value.__class__(*(self.render_template(el, context, jinja_env, oids) for el in value))
-        elif isinstance(value, list):
-            return [self.render_template(element, context, jinja_env, oids) for element in value]
-        elif isinstance(value, dict):
-            return {k: self.render_template(v, context, jinja_env, oids) for k, v in value.items()}
-        elif isinstance(value, set):
-            return {self.render_template(element, context, jinja_env, oids) for element in value}
-
-        # More complex collections.
-        self._render_nested_template_fields(value, context, jinja_env, oids)
-        return value
-
-    def _render_nested_template_fields(
-        self,
-        value: Any,
-        context: Context,
-        jinja_env: jinja2.Environment,
-        seen_oids: set[int],
-    ) -> None:
-        if id(value) in seen_oids:
-            return
-        seen_oids.add(id(value))
-        try:
-            nested_template_fields = value.template_fields
-        except AttributeError:
-            # content has no inner template fields
-            return
-        self._do_render_template_fields(value, nested_template_fields, context, jinja_env, seen_oids)
diff --git a/airflow/notifications/__init__.py b/airflow/notifications/__init__.py
new file mode 100644
index 0000000000000..13a83393a9124
--- /dev/null
+++ b/airflow/notifications/__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/notifications/notifier.py b/airflow/notifications/notifier.py
new file mode 100644
index 0000000000000..b366ec89c2364
--- /dev/null
+++ b/airflow/notifications/notifier.py
@@ -0,0 +1,108 @@
+# 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.
+
+from __future__ import annotations
+
+from abc import abstractmethod
+from typing import TYPE_CHECKING, Sequence
+
+import jinja2
+
+from airflow.template.templater import Templater
+from airflow.utils.context import Context, context_merge
+
+if TYPE_CHECKING:
+    from airflow import DAG
+
+
+class Notifier(Templater):
+    """
+    Base Notifier class for sending notifications
+
+    :param message: The message to send
+
+    The message or template_file can be used to send a notification. If both are set, the message
+    will be used.
+    """
+
+    template_fields: Sequence[str] = ("message",)
+    template_ext: Sequence[str] = (".txt",)
+
+    def __init__(
+        self,
+        message: str | None = "This is a default message",
+    ):
+        super().__init__()
+        self.message = message
+        self.resolve_template_files()
+
+    def _update_context(self, context: Context) -> Context:
+        """
+        Add additional context to the context
+
+        :param context: The airflow context
+        """
+        additional_context = {}
+        for field in self.template_fields:
+            additional_context[field] = getattr(self, field)
+        context_merge(context, additional_context)
+        return context
+
+    def _render(self, template, context, dag: DAG | None = None):
+        dag = context["dag"] if dag is None else dag
+        return super()._render(template, context, dag)
+
+    def render_template_fields(
+        self,
+        context: Context,
+        jinja_env: jinja2.Environment | None = None,
+    ) -> None:
+        """Template all attributes listed in *self.template_fields*.
+
+        This mutates the attributes in-place and is irreversible.
+
+        :param context: Context dict with values to apply on content.
+        :param jinja_env: Jinja environment to use for rendering.
+        """
+        context = self._update_context(context)
+        dag = context["dag"]
+        if not jinja_env:
+            jinja_env = self.get_template_env(dag=dag)
+        self._do_render_template_fields(self, self.template_fields, context, jinja_env, set())
+
+    @abstractmethod
+    def notify(self, context: Context) -> None:
+        """
+        Sends a notification
+
+        subclasses should always call super().notify(context) first to ensure
+        that the template is rendered with the context.
+
+        :param context: The airflow context
+        """
+        self.render_template_fields(context)
+
+    def __call__(self, context: Context) -> None:
+        """
+        Send a notification
+
+        :param context: The airflow context
+        """
+        try:
+            self.notify(context)
+        except Exception as e:
+            self.log.exception("Failed to send notification: %s", e)
diff --git a/airflow/providers/slack/notifier.py b/airflow/providers/slack/notifier.py
new file mode 100644
index 0000000000000..0472b512745ee
--- /dev/null
+++ b/airflow/providers/slack/notifier.py
@@ -0,0 +1,80 @@
+# 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.
+
+from __future__ import annotations
+
+import json
+
+from airflow.compat.functools import cached_property
+from airflow.notifications.notifier import Notifier
+from airflow.providers.slack.hooks.slack import SlackHook
+
+
+class SlackNotifier(Notifier):
+    """
+    Slack Notifier
+
+    :param slack_conn_id: Slack API token (https://api.slack.com/web). Optional
+    :param channel: The channel to send the message to. Optional
+    :param username: The username to send the message as. Optional
+    :param icon_url: The icon to use for the message. Optional
+    :param attachments: A list of attachments to send with the message. Optional
+    :param blocks: A list of blocks to send with the message. Optional
+    """
+
+    template_fields = ("message", "channel", "username", "attachments", "blocks")
+
+    def __init__(
+        self,
+        *,
+        slack_conn_id: str | None = "slack_api_default",
+        channel: str = "#general",
+        username: str = "Airflow",
+        icon_url: str = "https://raw.githubusercontent.com/apache/"
+        "airflow/main/airflow/www/static/pin_100.png",
+        attachments: list | None = None,
+        blocks: list | None = None,
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self.slack_conn_id = slack_conn_id
+        self.channel = channel
+        self.username = username
+        self.icon_url = icon_url
+        self.attachments = attachments or []
+        self.blocks = blocks or []
+
+    @cached_property
+    def hook(self) -> SlackHook:
+        """Slack Hook."""
+        return SlackHook(slack_conn_id=self.slack_conn_id)
+
+    def notify(self, context):
+        """Send a message to a Slack Channel"""
+        super().notify(context)
+        api_call_params = {
+            "channel": self.channel,
+            "username": self.username,
+            "text": self.message,
+            "icon_url": self.icon_url,
+            "attachments": json.dumps(self.attachments),
+            "blocks": json.dumps(self.blocks),
+        }
+        self.hook.call("chat.postMessage", json=api_call_params)
+
+
+send_slack_notification = SlackNotifier
diff --git a/airflow/template/__init__.py b/airflow/template/__init__.py
new file mode 100644
index 0000000000000..13a83393a9124
--- /dev/null
+++ b/airflow/template/__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/template/templater.py b/airflow/template/templater.py
new file mode 100644
index 0000000000000..e68015a4e703f
--- /dev/null
+++ b/airflow/template/templater.py
@@ -0,0 +1,187 @@
+# 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.
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Collection, Iterable, Sequence
+
+from airflow.utils.context import Context
+from airflow.utils.helpers import render_template_as_native, render_template_to_string
+from airflow.utils.log.logging_mixin import LoggingMixin
+from airflow.utils.mixins import ResolveMixin
+from airflow.utils.session import NEW_SESSION, provide_session
+
+if TYPE_CHECKING:
+    import jinja2
+    from sqlalchemy.orm import Session
+
+    from airflow import DAG
+
+
+class Templater(LoggingMixin):
+    """This renders the template fields of object."""
+
+    # For derived classes to define which fields will get jinjaified.
+    template_fields: Collection[str]
+    # Defines which files extensions to look for in the templated fields.
+    template_ext: Sequence[str]
+
+    def get_template_env(self, dag: DAG | None = None) -> jinja2.Environment:
+        """Fetch a Jinja template environment from the DAG or instantiate empty environment if no DAG."""
+        # This is imported locally since Jinja2 is heavy and we don't need it
+        # for most of the functionalities. It is imported by get_template_env()
+        # though, so we don't need to put this after the 'if dag' check.
+        from airflow.templates import SandboxedEnvironment
+
+        if dag:
+            return dag.get_template_env(force_sandboxed=False)
+        return SandboxedEnvironment(cache_size=0)
+
+    def prepare_template(self) -> None:
+        """Hook triggered after the templated fields get replaced by their content.
+
+        If you need your object to alter the content of the file before the
+        template is rendered, it should override this method to do so.
+        """
+
+    def resolve_template_files(self) -> None:
+        """Getting the content of files for template_field / template_ext."""
+        if self.template_ext:
+            for field in self.template_fields:
+                content = getattr(self, field, None)
+                if content is None:
+                    continue
+                elif isinstance(content, str) and any(content.endswith(ext) for ext in self.template_ext):
+                    env = self.get_template_env()
+                    try:
+                        setattr(self, field, env.loader.get_source(env, content)[0])  # type: ignore
+                    except Exception:
+                        self.log.exception("Failed to resolve template field %r", field)
+                elif isinstance(content, list):
+                    env = self.get_template_env()
+                    for i, item in enumerate(content):
+                        if isinstance(item, str) and any(item.endswith(ext) for ext in self.template_ext):
+                            try:
+                                content[i] = env.loader.get_source(env, item)[0]  # type: ignore
+                            except Exception:
+                                self.log.exception("Failed to get source %s", item)
+        self.prepare_template()
+
+    @provide_session
+    def _do_render_template_fields(
+        self,
+        parent: Any,
+        template_fields: Iterable[str],
+        context: Context,
+        jinja_env: jinja2.Environment,
+        seen_oids: set[int],
+        *,
+        session: Session = NEW_SESSION,
+    ) -> None:
+        for attr_name in template_fields:
+            value = getattr(parent, attr_name)
+            rendered_content = self.render_template(
+                value,
+                context,
+                jinja_env,
+                seen_oids,
+            )
+            if rendered_content:
+                setattr(parent, attr_name, rendered_content)
+
+    def _render(self, template, context, dag: DAG | None = None) -> Any:
+        if dag and dag.render_template_as_native_obj:
+            return render_template_as_native(template, context)
+        return render_template_to_string(template, context)
+
+    def render_template(
+        self,
+        content: Any,
+        context: Context,
+        jinja_env: jinja2.Environment | None = None,
+        seen_oids: set[int] | None = None,
+    ) -> Any:
+        """Render a templated string.
+
+        If *content* is a collection holding multiple templated strings, strings
+        in the collection will be templated recursively.
+
+        :param content: Content to template. Only strings can be templated (may
+            be inside a collection).
+        :param context: Dict with values to apply on templated content
+        :param jinja_env: Jinja environment. Can be provided to avoid
+            re-creating Jinja environments during recursion.
+        :param seen_oids: template fields already rendered (to avoid
+            *RecursionError* on circular dependencies)
+        :return: Templated content
+        """
+        # "content" is a bad name, but we're stuck to it being public API.
+        value = content
+        del content
+
+        if seen_oids is not None:
+            oids = seen_oids
+        else:
+            oids = set()
+
+        if id(value) in oids:
+            return value
+
+        if not jinja_env:
+            jinja_env = self.get_template_env()
+
+        if isinstance(value, str):
+            if any(value.endswith(ext) for ext in self.template_ext):  # A filepath.
+                template = jinja_env.get_template(value)
+            else:
+                template = jinja_env.from_string(value)
+            return self._render(template, context)
+        if isinstance(value, ResolveMixin):
+            return value.resolve(context)
+
+        # Fast path for common built-in collections.
+        if value.__class__ is tuple:
+            return tuple(self.render_template(element, context, jinja_env, oids) for element in value)
+        elif isinstance(value, tuple):  # Special case for named tuples.
+            return value.__class__(*(self.render_template(el, context, jinja_env, oids) for el in value))
+        elif isinstance(value, list):
+            return [self.render_template(element, context, jinja_env, oids) for element in value]
+        elif isinstance(value, dict):
+            return {k: self.render_template(v, context, jinja_env, oids) for k, v in value.items()}
+        elif isinstance(value, set):
+            return {self.render_template(element, context, jinja_env, oids) for element in value}
+
+        # More complex collections.
+        self._render_nested_template_fields(value, context, jinja_env, oids)
+        return value
+
+    def _render_nested_template_fields(
+        self,
+        value: Any,
+        context: Context,
+        jinja_env: jinja2.Environment,
+        seen_oids: set[int],
+    ) -> None:
+        if id(value) in seen_oids:
+            return
+        seen_oids.add(id(value))
+        try:
+            nested_template_fields = value.template_fields
+        except AttributeError:
+            # content has no inner template fields
+            return
+        self._do_render_template_fields(value, nested_template_fields, context, jinja_env, seen_oids)
diff --git a/tests/notifications/__init__.py b/tests/notifications/__init__.py
new file mode 100644
index 0000000000000..13a83393a9124
--- /dev/null
+++ b/tests/notifications/__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/tests/notifications/test_notifier.py b/tests/notifications/test_notifier.py
new file mode 100644
index 0000000000000..338c7e5365432
--- /dev/null
+++ b/tests/notifications/test_notifier.py
@@ -0,0 +1,52 @@
+# 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.
+
+from __future__ import annotations
+
+import jinja2
+import pytest
+
+from airflow.notifications.notifier import Notifier
+from airflow.operators.empty import EmptyOperator
+from airflow.utils.context import Context
+
+
+class TestNotifier:
+    def test_render_message_with_message(self, dag_maker):
+        with dag_maker("test_render_message_with_message") as dag:
+            EmptyOperator(task_id="test_id")
+
+        notifier = Notifier(message="Hello {{ dag.dag_id }}")
+        context: Context = {"dag": dag}
+        notifier.render_template_fields(context)
+        assert notifier.message == "Hello test_render_message_with_message"
+
+    def test_render_message_with_template(self, dag_maker, caplog):
+        with dag_maker("test_render_message_with_template") as dag:
+            EmptyOperator(task_id="test_id")
+        notifier = Notifier(message="test.txt")
+        context: Context = {"dag": dag}
+        with pytest.raises(jinja2.exceptions.TemplateNotFound):
+            notifier.render_template_fields(context)
+
+    def test_render_message_with_template_works(self, dag_maker, caplog):
+        with dag_maker("test_render_message_with_template_works") as dag:
+            EmptyOperator(task_id="test_id")
+        notifier = Notifier(message="test_notifier.txt")
+        context: Context = {"dag": dag}
+        notifier.render_template_fields(context)
+        assert notifier.message == "Hello test_render_message_with_template_works"
diff --git a/tests/notifications/test_notifier.txt b/tests/notifications/test_notifier.txt
new file mode 100644
index 0000000000000..b99a96436555b
--- /dev/null
+++ b/tests/notifications/test_notifier.txt
@@ -0,0 +1 @@
+Hello {{dag.dag_id}}
diff --git a/tests/providers/slack/test_notifier.py b/tests/providers/slack/test_notifier.py
new file mode 100644
index 0000000000000..16e3a9024dea7
--- /dev/null
+++ b/tests/providers/slack/test_notifier.py
@@ -0,0 +1,111 @@
+# 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.
+
+from __future__ import annotations
+
+from unittest import mock
+
+import pytest
+
+from airflow.models import Connection
+from airflow.operators.empty import EmptyOperator
+from airflow.providers.slack.hooks.slack import SlackHook
+from airflow.providers.slack.notifier import SlackNotifier, send_slack_notification
+
+SLACK_API_DEFAULT_CONN_ID = SlackHook.default_conn_name
+
+
+@pytest.fixture(scope="module", autouse=True)
+def slack_api_connections():
+    """Create tests connections."""
+    connections = [
+        Connection(
+            conn_id=SLACK_API_DEFAULT_CONN_ID,
+            conn_type="slack",
+            password="xoxb-1234567890123-09876543210987-AbCdEfGhIjKlMnOpQrStUvWx",
+        ),
+    ]
+
+    conn_uris = {f"AIRFLOW_CONN_{c.conn_id.upper()}": c.get_uri() for c in connections}
+
+    with mock.patch.dict("os.environ", values=conn_uris):
+        yield
+
+
+class TestSlackNotifier:
+    @mock.patch("airflow.providers.slack.notifier.SlackHook")
+    def test_slack_notifier(self, mock_slack_hook, slack_api_connections, dag_maker):
+        with dag_maker("test_slack_notifier") as dag:
+            EmptyOperator(task_id="task1")
+        notifier = send_slack_notification(message="test")
+        notifier.notify(context={"dag": dag})
+        mock_slack_hook.return_value.call.assert_called_once_with(
+            "chat.postMessage",
+            json={
+                "channel": "#general",
+                "username": "Airflow",
+                "text": "test",
+                "icon_url": "https://raw.githubusercontent.com/apache/airflow/main/airflow/www/static"
+                "/pin_100.png",
+                "attachments": "[]",
+                "blocks": "[]",
+            },
+        )
+
+    @mock.patch("airflow.providers.slack.notifier.SlackHook")
+    def test_slack_notifier_with_notifier_class(self, mock_slack_hook, slack_api_connections, dag_maker):
+        with dag_maker("test_slack_notifier") as dag:
+            EmptyOperator(task_id="task1")
+        notifier = SlackNotifier(message="test")
+        notifier.notify(context={"dag": dag})
+        mock_slack_hook.return_value.call.assert_called_once_with(
+            "chat.postMessage",
+            json={
+                "channel": "#general",
+                "username": "Airflow",
+                "text": "test",
+                "icon_url": "https://raw.githubusercontent.com/apache/airflow/main/airflow/www/static"
+                "/pin_100.png",
+                "attachments": "[]",
+                "blocks": "[]",
+            },
+        )
+
+    @mock.patch("airflow.providers.slack.notifier.SlackHook")
+    def test_slack_notifier_templated(self, mock_slack_hook, slack_api_connections, dag_maker):
+        with dag_maker("test_slack_notifier") as dag:
+            EmptyOperator(task_id="task1")
+
+        notifier = send_slack_notification(
+            message="test {{ username }}",
+            channel="#test-{{dag.dag_id}}",
+            attachments=[{"image_url": "{{ dag.dag_id }}.png"}],
+        )
+        context = {"dag": dag}
+        notifier.notify(context)
+        mock_slack_hook.return_value.call.assert_called_once_with(
+            "chat.postMessage",
+            json={
+                "channel": "#test-test_slack_notifier",
+                "username": "Airflow",
+                "text": "test Airflow",
+                "icon_url": "https://raw.githubusercontent.com/apache/airflow/main/airflow/www/static"
+                "/pin_100.png",
+                "attachments": '[{"image_url": "test_slack_notifier.png"}]',
+                "blocks": "[]",
+            },
+        )
diff --git a/tests/template/__init__.py b/tests/template/__init__.py
new file mode 100644
index 0000000000000..13a83393a9124
--- /dev/null
+++ b/tests/template/__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/tests/template/test_templater.py b/tests/template/test_templater.py
new file mode 100644
index 0000000000000..7f6a85009061f
--- /dev/null
+++ b/tests/template/test_templater.py
@@ -0,0 +1,62 @@
+# 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.
+
+from __future__ import annotations
+
+import jinja2
+
+from airflow import DAG
+from airflow.template.templater import Templater
+from airflow.utils.context import Context
+
+
+class TestTemplater:
+    def test_get_template_env(self):
+        # Test get_template_env when a DAG is provided
+        templater = Templater()
+        dag = DAG(dag_id="test_dag", render_template_as_native_obj=True)
+        env = templater.get_template_env(dag)
+        assert isinstance(env, jinja2.Environment)
+        assert not env.sandboxed
+
+        # Test get_template_env when no DAG is provided
+        templater = Templater()
+        env = templater.get_template_env()
+        assert isinstance(env, jinja2.Environment)
+        assert env.sandboxed
+
+    def test_prepare_template(self):
+        # Test that prepare_template is a no-op
+        templater = Templater()
+        templater.prepare_template()
+
+    def test_resolve_template_files_logs_exception(self, caplog):
+        templater = Templater()
+        templater.message = "template_file.txt"
+        templater.template_fields = ["message"]
+        templater.template_ext = [".txt"]
+        templater.resolve_template_files()
+        assert "Failed to resolve template field 'message'" in caplog.text
+
+    def test_render_template(self):
+        context = Context({"name": "world"})  # type: ignore
+        templater = Templater()
+        templater.message = "Hello {{ name }}"
+        templater.template_fields = ["message"]
+        templater.template_ext = [".txt"]
+        rendered_content = templater.render_template(templater.message, context)
+        assert rendered_content == "Hello world"

From 85c32e05b19965d5a95aa9df1399168cb419a4de Mon Sep 17 00:00:00 2001
From: Ephraim Anierobi <splendidzigy24@gmail.com>
Date: Tue, 27 Dec 2022 10:43:30 +0100
Subject: [PATCH 2/7] Renames and a fixup not to require a call to super in
 subclasses

---
 .../{notifier.py => basenotifier.py}          | 29 +++++--------------
 .../providers/slack/notifications/__init__.py | 16 ++++++++++
 .../slack_notifier.py}                        | 21 +++++++-------
 ...{test_notifier.py => test_basenotifier.py} | 24 +++++++++++----
 .../providers/slack/notifications/__init__.py | 16 ++++++++++
 .../test_slack_notifier.py}                   | 20 ++++++-------
 6 files changed, 80 insertions(+), 46 deletions(-)
 rename airflow/notifications/{notifier.py => basenotifier.py} (82%)
 create mode 100644 airflow/providers/slack/notifications/__init__.py
 rename airflow/providers/slack/{notifier.py => notifications/slack_notifier.py} (85%)
 rename tests/notifications/{test_notifier.py => test_basenotifier.py} (76%)
 create mode 100644 tests/providers/slack/notifications/__init__.py
 rename tests/providers/slack/{test_notifier.py => notifications/test_slack_notifier.py} (86%)

diff --git a/airflow/notifications/notifier.py b/airflow/notifications/basenotifier.py
similarity index 82%
rename from airflow/notifications/notifier.py
rename to airflow/notifications/basenotifier.py
index b366ec89c2364..5ebf96a173a12 100644
--- a/airflow/notifications/notifier.py
+++ b/airflow/notifications/basenotifier.py
@@ -29,25 +29,14 @@
     from airflow import DAG
 
 
-class Notifier(Templater):
-    """
-    Base Notifier class for sending notifications
+class BaseNotifier(Templater):
+    """BaseNotifier class for sending notifications"""
 
-    :param message: The message to send
+    template_fields: Sequence[str] = ()
+    template_ext: Sequence[str] = ()
 
-    The message or template_file can be used to send a notification. If both are set, the message
-    will be used.
-    """
-
-    template_fields: Sequence[str] = ("message",)
-    template_ext: Sequence[str] = (".txt",)
-
-    def __init__(
-        self,
-        message: str | None = "This is a default message",
-    ):
+    def __init__(self):
         super().__init__()
-        self.message = message
         self.resolve_template_files()
 
     def _update_context(self, context: Context) -> Context:
@@ -78,7 +67,6 @@ def render_template_fields(
         :param context: Context dict with values to apply on content.
         :param jinja_env: Jinja environment to use for rendering.
         """
-        context = self._update_context(context)
         dag = context["dag"]
         if not jinja_env:
             jinja_env = self.get_template_env(dag=dag)
@@ -89,12 +77,9 @@ def notify(self, context: Context) -> None:
         """
         Sends a notification
 
-        subclasses should always call super().notify(context) first to ensure
-        that the template is rendered with the context.
-
         :param context: The airflow context
         """
-        self.render_template_fields(context)
+        ...
 
     def __call__(self, context: Context) -> None:
         """
@@ -102,6 +87,8 @@ def __call__(self, context: Context) -> None:
 
         :param context: The airflow context
         """
+        context = self._update_context(context)
+        self.render_template_fields(context)
         try:
             self.notify(context)
         except Exception as e:
diff --git a/airflow/providers/slack/notifications/__init__.py b/airflow/providers/slack/notifications/__init__.py
new file mode 100644
index 0000000000000..13a83393a9124
--- /dev/null
+++ b/airflow/providers/slack/notifications/__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/slack/notifier.py b/airflow/providers/slack/notifications/slack_notifier.py
similarity index 85%
rename from airflow/providers/slack/notifier.py
rename to airflow/providers/slack/notifications/slack_notifier.py
index 0472b512745ee..9fbf698913488 100644
--- a/airflow/providers/slack/notifier.py
+++ b/airflow/providers/slack/notifications/slack_notifier.py
@@ -20,15 +20,16 @@
 import json
 
 from airflow.compat.functools import cached_property
-from airflow.notifications.notifier import Notifier
+from airflow.notifications.basenotifier import BaseNotifier
 from airflow.providers.slack.hooks.slack import SlackHook
 
 
-class SlackNotifier(Notifier):
+class SlackNotifier(BaseNotifier):
     """
-    Slack Notifier
+    Slack BaseNotifier
 
-    :param slack_conn_id: Slack API token (https://api.slack.com/web). Optional
+    :param slack_conn_id: Slack API token (https://api.slack.com/web).
+    :param text: The content of the message
     :param channel: The channel to send the message to. Optional
     :param username: The username to send the message as. Optional
     :param icon_url: The icon to use for the message. Optional
@@ -36,22 +37,23 @@ class SlackNotifier(Notifier):
     :param blocks: A list of blocks to send with the message. Optional
     """
 
-    template_fields = ("message", "channel", "username", "attachments", "blocks")
+    template_fields = ("text", "channel", "username", "attachments", "blocks")
 
     def __init__(
         self,
         *,
-        slack_conn_id: str | None = "slack_api_default",
+        slack_conn_id: str = "slack_api_default",
+        text: str = "This is a default message",
         channel: str = "#general",
         username: str = "Airflow",
         icon_url: str = "https://raw.githubusercontent.com/apache/"
         "airflow/main/airflow/www/static/pin_100.png",
         attachments: list | None = None,
         blocks: list | None = None,
-        **kwargs,
     ):
-        super().__init__(**kwargs)
+        super().__init__()
         self.slack_conn_id = slack_conn_id
+        self.text = text
         self.channel = channel
         self.username = username
         self.icon_url = icon_url
@@ -65,11 +67,10 @@ def hook(self) -> SlackHook:
 
     def notify(self, context):
         """Send a message to a Slack Channel"""
-        super().notify(context)
         api_call_params = {
             "channel": self.channel,
             "username": self.username,
-            "text": self.message,
+            "text": self.text,
             "icon_url": self.icon_url,
             "attachments": json.dumps(self.attachments),
             "blocks": json.dumps(self.blocks),
diff --git a/tests/notifications/test_notifier.py b/tests/notifications/test_basenotifier.py
similarity index 76%
rename from tests/notifications/test_notifier.py
rename to tests/notifications/test_basenotifier.py
index 338c7e5365432..574d577f75f94 100644
--- a/tests/notifications/test_notifier.py
+++ b/tests/notifications/test_basenotifier.py
@@ -20,17 +20,31 @@
 import jinja2
 import pytest
 
-from airflow.notifications.notifier import Notifier
+from airflow.notifications.basenotifier import BaseNotifier
 from airflow.operators.empty import EmptyOperator
 from airflow.utils.context import Context
 
 
-class TestNotifier:
+class MockNotifier(BaseNotifier):
+    """MockNotifier class for testing"""
+
+    template_fields = ("message",)
+    template_ext = (".txt",)
+
+    def __init__(self, message: str | None = "This is a test message"):
+        super().__init__()
+        self.message = message
+
+    def notify(self, context: Context) -> None:
+        pass
+
+
+class TestBaseNotifier:
     def test_render_message_with_message(self, dag_maker):
         with dag_maker("test_render_message_with_message") as dag:
             EmptyOperator(task_id="test_id")
 
-        notifier = Notifier(message="Hello {{ dag.dag_id }}")
+        notifier = MockNotifier(message="Hello {{ dag.dag_id }}")
         context: Context = {"dag": dag}
         notifier.render_template_fields(context)
         assert notifier.message == "Hello test_render_message_with_message"
@@ -38,7 +52,7 @@ def test_render_message_with_message(self, dag_maker):
     def test_render_message_with_template(self, dag_maker, caplog):
         with dag_maker("test_render_message_with_template") as dag:
             EmptyOperator(task_id="test_id")
-        notifier = Notifier(message="test.txt")
+        notifier = MockNotifier(message="test.txt")
         context: Context = {"dag": dag}
         with pytest.raises(jinja2.exceptions.TemplateNotFound):
             notifier.render_template_fields(context)
@@ -46,7 +60,7 @@ def test_render_message_with_template(self, dag_maker, caplog):
     def test_render_message_with_template_works(self, dag_maker, caplog):
         with dag_maker("test_render_message_with_template_works") as dag:
             EmptyOperator(task_id="test_id")
-        notifier = Notifier(message="test_notifier.txt")
+        notifier = MockNotifier(message="test_notifier.txt")
         context: Context = {"dag": dag}
         notifier.render_template_fields(context)
         assert notifier.message == "Hello test_render_message_with_template_works"
diff --git a/tests/providers/slack/notifications/__init__.py b/tests/providers/slack/notifications/__init__.py
new file mode 100644
index 0000000000000..13a83393a9124
--- /dev/null
+++ b/tests/providers/slack/notifications/__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/tests/providers/slack/test_notifier.py b/tests/providers/slack/notifications/test_slack_notifier.py
similarity index 86%
rename from tests/providers/slack/test_notifier.py
rename to tests/providers/slack/notifications/test_slack_notifier.py
index 16e3a9024dea7..bbcebf20e69a6 100644
--- a/tests/providers/slack/test_notifier.py
+++ b/tests/providers/slack/notifications/test_slack_notifier.py
@@ -24,7 +24,7 @@
 from airflow.models import Connection
 from airflow.operators.empty import EmptyOperator
 from airflow.providers.slack.hooks.slack import SlackHook
-from airflow.providers.slack.notifier import SlackNotifier, send_slack_notification
+from airflow.providers.slack.notifications.slack_notifier import SlackNotifier, send_slack_notification
 
 SLACK_API_DEFAULT_CONN_ID = SlackHook.default_conn_name
 
@@ -47,12 +47,12 @@ def slack_api_connections():
 
 
 class TestSlackNotifier:
-    @mock.patch("airflow.providers.slack.notifier.SlackHook")
+    @mock.patch("airflow.providers.slack.notifications.slack_notifier.SlackHook")
     def test_slack_notifier(self, mock_slack_hook, slack_api_connections, dag_maker):
         with dag_maker("test_slack_notifier") as dag:
             EmptyOperator(task_id="task1")
-        notifier = send_slack_notification(message="test")
-        notifier.notify(context={"dag": dag})
+        notifier = send_slack_notification(text="test")
+        notifier(context={"dag": dag})
         mock_slack_hook.return_value.call.assert_called_once_with(
             "chat.postMessage",
             json={
@@ -66,12 +66,12 @@ def test_slack_notifier(self, mock_slack_hook, slack_api_connections, dag_maker)
             },
         )
 
-    @mock.patch("airflow.providers.slack.notifier.SlackHook")
+    @mock.patch("airflow.providers.slack.notifications.slack_notifier.SlackHook")
     def test_slack_notifier_with_notifier_class(self, mock_slack_hook, slack_api_connections, dag_maker):
         with dag_maker("test_slack_notifier") as dag:
             EmptyOperator(task_id="task1")
-        notifier = SlackNotifier(message="test")
-        notifier.notify(context={"dag": dag})
+        notifier = SlackNotifier(text="test")
+        notifier(context={"dag": dag})
         mock_slack_hook.return_value.call.assert_called_once_with(
             "chat.postMessage",
             json={
@@ -85,18 +85,18 @@ def test_slack_notifier_with_notifier_class(self, mock_slack_hook, slack_api_con
             },
         )
 
-    @mock.patch("airflow.providers.slack.notifier.SlackHook")
+    @mock.patch("airflow.providers.slack.notifications.slack_notifier.SlackHook")
     def test_slack_notifier_templated(self, mock_slack_hook, slack_api_connections, dag_maker):
         with dag_maker("test_slack_notifier") as dag:
             EmptyOperator(task_id="task1")
 
         notifier = send_slack_notification(
-            message="test {{ username }}",
+            text="test {{ username }}",
             channel="#test-{{dag.dag_id}}",
             attachments=[{"image_url": "{{ dag.dag_id }}.png"}],
         )
         context = {"dag": dag}
-        notifier.notify(context)
+        notifier(context)
         mock_slack_hook.return_value.call.assert_called_once_with(
             "chat.postMessage",
             json={

From 6a6241716fe9d2bb7752defbdcbd08c10135ba00 Mon Sep 17 00:00:00 2001
From: Ephraim Anierobi <splendidzigy24@gmail.com>
Date: Wed, 28 Dec 2022 08:13:04 +0100
Subject: [PATCH 3/7] Raise compat exception and add docs

---
 airflow/providers/slack/CHANGELOG.rst         |  1 +
 .../slack/notifications/slack_notifier.py     |  8 ++-
 docs/apache-airflow-providers-slack/index.rst |  3 +
 .../slack_notifier_howto_guide.rst            | 56 +++++++++++++++++++
 4 files changed, 67 insertions(+), 1 deletion(-)
 create mode 100644 docs/apache-airflow-providers-slack/notifications/slack_notifier_howto_guide.rst

diff --git a/airflow/providers/slack/CHANGELOG.rst b/airflow/providers/slack/CHANGELOG.rst
index c63f65fcbfb7a..9025908d1e39f 100644
--- a/airflow/providers/slack/CHANGELOG.rst
+++ b/airflow/providers/slack/CHANGELOG.rst
@@ -41,6 +41,7 @@ Features
 ~~~~~~~~
 
 * ``Implements SqlToSlackApiFileOperator (#26374)``
+* ``Added SlackNotifier (#28569)``
 
 Bug Fixes
 ~~~~~~~~~
diff --git a/airflow/providers/slack/notifications/slack_notifier.py b/airflow/providers/slack/notifications/slack_notifier.py
index 9fbf698913488..81b2602d83c84 100644
--- a/airflow/providers/slack/notifications/slack_notifier.py
+++ b/airflow/providers/slack/notifications/slack_notifier.py
@@ -20,7 +20,13 @@
 import json
 
 from airflow.compat.functools import cached_property
-from airflow.notifications.basenotifier import BaseNotifier
+from airflow.exceptions import AirflowOptionalProviderFeatureException
+
+try:
+    from airflow.notifications.basenotifier import BaseNotifier
+except ImportError as e:
+    raise AirflowOptionalProviderFeatureException(e)
+
 from airflow.providers.slack.hooks.slack import SlackHook
 
 
diff --git a/docs/apache-airflow-providers-slack/index.rst b/docs/apache-airflow-providers-slack/index.rst
index d0a86c31f9397..ba1a0f4b958e0 100644
--- a/docs/apache-airflow-providers-slack/index.rst
+++ b/docs/apache-airflow-providers-slack/index.rst
@@ -26,6 +26,7 @@ Content
     :caption: System tests
 
     System Tests <_api/tests/system/providers/slack/index>
+
 .. toctree::
     :maxdepth: 1
     :caption: Guides
@@ -37,6 +38,8 @@ Content
     :caption: Guides
 
     SqlToSlackOperator types <operators/sql_to_slack>
+    Slack Notifications <notifications/slack_notifier_howto_guide>
+
 
 .. toctree::
     :maxdepth: 1
diff --git a/docs/apache-airflow-providers-slack/notifications/slack_notifier_howto_guide.rst b/docs/apache-airflow-providers-slack/notifications/slack_notifier_howto_guide.rst
new file mode 100644
index 0000000000000..674095777906c
--- /dev/null
+++ b/docs/apache-airflow-providers-slack/notifications/slack_notifier_howto_guide.rst
@@ -0,0 +1,56 @@
+ .. 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.
+
+How-to Guide for Slack notifications
+====================================
+
+Introduction
+------------
+Slack notifier (:class:`airflow.providers.slack.notifications.slack_notifier.SlackNotifier`) allows users to send
+messages to a slack channel using the various ``on_*_callbacks`` at both the DAG level and Task level
+
+Example Code:
+-------------
+
+.. code-block:: python
+
+    from datetime import datetime
+    from airflow import DAG
+    from airflow.operators.bash import BashOperator
+    from airflow.providers.slack.notifications.slack_notifier import send_slack_notification
+
+    with DAG(
+        start_date=datetime(2023, 1, 1),
+        on_success_callback=[
+            send_slack_notification(
+                text="The DAG {{ dag.dag_id }} succeeded",
+                channel="#general",
+                username="Airflow",
+            )
+        ],
+    ):
+        BashOperator(
+            task_id="mytask",
+            on_failure_callback=[
+                send_slack_notification(
+                    text="The task {{ ti.task_id }} failed",
+                    channel="#general",
+                    username="Airflow",
+                )
+            ],
+            bash_command="fail",
+        )

From 3cba59bdea4aa7968e6e088ef87b0d90f435bea4 Mon Sep 17 00:00:00 2001
From: Ephraim Anierobi <splendidzigy24@gmail.com>
Date: Wed, 28 Dec 2022 15:17:37 +0100
Subject: [PATCH 4/7] Ignore import error due to optional provider feature

---
 scripts/in_container/verify_providers.py | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/scripts/in_container/verify_providers.py b/scripts/in_container/verify_providers.py
index 87d79ee6ca123..93f71afb26f83 100755
--- a/scripts/in_container/verify_providers.py
+++ b/scripts/in_container/verify_providers.py
@@ -33,6 +33,8 @@
 
 from rich.console import Console
 
+from airflow.exceptions import AirflowOptionalProviderFeatureException
+
 console = Console(width=400, color_system="standard")
 
 AIRFLOW_SOURCES_ROOT = Path(__file__).parents[2].resolve()
@@ -360,6 +362,9 @@ def onerror(_):
                             imported_classes.append(class_name)
                 if w:
                     all_warnings.extend(w)
+            except AirflowOptionalProviderFeatureException:
+                # We ignore optional features
+                ...
             except Exception:
                 exception_str = traceback.format_exc()
                 tracebacks.append((modinfo.name, exception_str))

From 016c6e818088278a0c08d74957ff47ff4b923f18 Mon Sep 17 00:00:00 2001
From: Ephraim Anierobi <splendidzigy24@gmail.com>
Date: Wed, 28 Dec 2022 18:36:42 +0100
Subject: [PATCH 5/7] fixup! Ignore import error due to optional provider
 feature

---
 airflow/providers/slack/notifications/slack_notifier.py | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/airflow/providers/slack/notifications/slack_notifier.py b/airflow/providers/slack/notifications/slack_notifier.py
index 81b2602d83c84..e2fe130c97835 100644
--- a/airflow/providers/slack/notifications/slack_notifier.py
+++ b/airflow/providers/slack/notifications/slack_notifier.py
@@ -24,8 +24,10 @@
 
 try:
     from airflow.notifications.basenotifier import BaseNotifier
-except ImportError as e:
-    raise AirflowOptionalProviderFeatureException(e)
+except ImportError:
+    raise AirflowOptionalProviderFeatureException(
+        "Failed to import BaseNotifier. This feature is only available in Airflow versions >= 2.6.0"
+    )
 
 from airflow.providers.slack.hooks.slack import SlackHook
 

From 67fc2880bae02bf2cc85612a483acbd447bc39d6 Mon Sep 17 00:00:00 2001
From: Ephraim Anierobi <splendidzigy24@gmail.com>
Date: Tue, 3 Jan 2023 09:31:39 +0100
Subject: [PATCH 6/7] Apply suggestions from code review

Co-authored-by: Tzu-ping Chung <uranusjr@gmail.com>
---
 airflow/notifications/basenotifier.py | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/airflow/notifications/basenotifier.py b/airflow/notifications/basenotifier.py
index 5ebf96a173a12..629927ce7d209 100644
--- a/airflow/notifications/basenotifier.py
+++ b/airflow/notifications/basenotifier.py
@@ -45,14 +45,12 @@ def _update_context(self, context: Context) -> Context:
 
         :param context: The airflow context
         """
-        additional_context = {}
-        for field in self.template_fields:
-            additional_context[field] = getattr(self, field)
+        additional_context = ((f, getattr(self, f)) for f in self.template_fields)
         context_merge(context, additional_context)
         return context
 
     def _render(self, template, context, dag: DAG | None = None):
-        dag = context["dag"] if dag is None else dag
+        dag = dag or context["dag"]
         return super()._render(template, context, dag)
 
     def render_template_fields(

From 84206ace11868a9fa45bd8d54072f9bf0ee24f0b Mon Sep 17 00:00:00 2001
From: Ephraim Anierobi <splendidzigy24@gmail.com>
Date: Tue, 3 Jan 2023 09:58:17 +0100
Subject: [PATCH 7/7] fixup! Apply suggestions from code review

---
 .../slack/notifications/slack_notifier.py          | 14 ++++++++------
 airflow/template/templater.py                      |  6 +++++-
 .../slack/notifications/test_slack_notifier.py     |  6 +++---
 3 files changed, 16 insertions(+), 10 deletions(-)

diff --git a/airflow/providers/slack/notifications/slack_notifier.py b/airflow/providers/slack/notifications/slack_notifier.py
index e2fe130c97835..b48ed3670b06f 100644
--- a/airflow/providers/slack/notifications/slack_notifier.py
+++ b/airflow/providers/slack/notifications/slack_notifier.py
@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 import json
+from typing import Sequence
 
 from airflow.compat.functools import cached_property
 from airflow.exceptions import AirflowOptionalProviderFeatureException
@@ -31,6 +32,8 @@
 
 from airflow.providers.slack.hooks.slack import SlackHook
 
+ICON_URL: str = "https://raw.githubusercontent.com/apache/airflow/2.5.0/airflow/www/static/pin_100.png"
+
 
 class SlackNotifier(BaseNotifier):
     """
@@ -54,10 +57,9 @@ def __init__(
         text: str = "This is a default message",
         channel: str = "#general",
         username: str = "Airflow",
-        icon_url: str = "https://raw.githubusercontent.com/apache/"
-        "airflow/main/airflow/www/static/pin_100.png",
-        attachments: list | None = None,
-        blocks: list | None = None,
+        icon_url: str = ICON_URL,
+        attachments: Sequence = (),
+        blocks: Sequence = (),
     ):
         super().__init__()
         self.slack_conn_id = slack_conn_id
@@ -65,8 +67,8 @@ def __init__(
         self.channel = channel
         self.username = username
         self.icon_url = icon_url
-        self.attachments = attachments or []
-        self.blocks = blocks or []
+        self.attachments = attachments
+        self.blocks = blocks
 
     @cached_property
     def hook(self) -> SlackHook:
diff --git a/airflow/template/templater.py b/airflow/template/templater.py
index e68015a4e703f..83c4e763f833b 100644
--- a/airflow/template/templater.py
+++ b/airflow/template/templater.py
@@ -33,7 +33,11 @@
 
 
 class Templater(LoggingMixin):
-    """This renders the template fields of object."""
+    """
+    This renders the template fields of object.
+
+    :meta private:
+    """
 
     # For derived classes to define which fields will get jinjaified.
     template_fields: Collection[str]
diff --git a/tests/providers/slack/notifications/test_slack_notifier.py b/tests/providers/slack/notifications/test_slack_notifier.py
index bbcebf20e69a6..d6f353c04f00b 100644
--- a/tests/providers/slack/notifications/test_slack_notifier.py
+++ b/tests/providers/slack/notifications/test_slack_notifier.py
@@ -59,7 +59,7 @@ def test_slack_notifier(self, mock_slack_hook, slack_api_connections, dag_maker)
                 "channel": "#general",
                 "username": "Airflow",
                 "text": "test",
-                "icon_url": "https://raw.githubusercontent.com/apache/airflow/main/airflow/www/static"
+                "icon_url": "https://raw.githubusercontent.com/apache/airflow/2.5.0/airflow/www/static"
                 "/pin_100.png",
                 "attachments": "[]",
                 "blocks": "[]",
@@ -78,7 +78,7 @@ def test_slack_notifier_with_notifier_class(self, mock_slack_hook, slack_api_con
                 "channel": "#general",
                 "username": "Airflow",
                 "text": "test",
-                "icon_url": "https://raw.githubusercontent.com/apache/airflow/main/airflow/www/static"
+                "icon_url": "https://raw.githubusercontent.com/apache/airflow/2.5.0/airflow/www/static"
                 "/pin_100.png",
                 "attachments": "[]",
                 "blocks": "[]",
@@ -103,7 +103,7 @@ def test_slack_notifier_templated(self, mock_slack_hook, slack_api_connections,
                 "channel": "#test-test_slack_notifier",
                 "username": "Airflow",
                 "text": "test Airflow",
-                "icon_url": "https://raw.githubusercontent.com/apache/airflow/main/airflow/www/static"
+                "icon_url": "https://raw.githubusercontent.com/apache/airflow/2.5.0/airflow/www/static"
                 "/pin_100.png",
                 "attachments": '[{"image_url": "test_slack_notifier.png"}]',
                 "blocks": "[]",
