Add Slack Incoming Webhook Notifier (#33966)

airflow/providers/slack/notifications/slack_webhook.py

@@ -0,0 +1,95 @@
+# 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 functools import cached_property
+
+from airflow.exceptions import AirflowOptionalProviderFeatureException
+from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook
+
+try:
+    from airflow.notifications.basenotifier import BaseNotifier
+except ImportError:
+    raise AirflowOptionalProviderFeatureException(
+        "Failed to import BaseNotifier. This feature is only available in Airflow versions >= 2.6.0"
+    )
+
+
+class SlackWebhookNotifier(BaseNotifier):
+    """
+    Slack Incoming Webhooks Notifier.
+
+    .. note::
+        ``SlackWebhookNotifier`` provide integration with Slack Incoming Webhooks,
+        and may not function accurately within Legacy Slack Integration Incoming Webhook.
+
+    :param slack_webhook_conn_id: :ref:`Slack Incoming Webhook <howto/connection:slack>`
+        connection id that has Incoming Webhook token in the password field.
+    :param text: The content of the message
+    :param blocks: A list of blocks to send with the message. Optional
+    :param unfurl_links: Option to indicate whether text url should unfurl. Optional
+    :param unfurl_media: Option to indicate whether media url should unfurl. Optional
+    :param timeout: The maximum number of seconds the client will wait to connect. Optional
+        and receive a response from Slack. If not set than default WebhookClient value will use.
+    :param proxy: Proxy to make the Slack Incoming Webhook call. Optional
+    :param attachments: A list of attachments to send with the message. Optional
+    """
+
+    template_fields = ("slack_webhook_conn_id", "text", "attachments", "blocks", "proxy", "timeout")
+
+    def __init__(
+        self,
+        *,
+        slack_webhook_conn_id: str = SlackWebhookHook.default_conn_name,
+        text: str,
+        blocks: list | None = None,
+        unfurl_links: bool | None = None,
+        unfurl_media: bool | None = None,
+        proxy: str | None = None,
+        timeout: int | None = None,
+        attachments: list | None = None,
+    ):
+        super().__init__()
+        self.slack_webhook_conn_id = slack_webhook_conn_id
+        self.text = text
+        self.attachments = attachments
+        self.blocks = blocks
+        self.unfurl_links = unfurl_links
+        self.unfurl_media = unfurl_media
+        self.timeout = timeout
+        self.proxy = proxy
+
+    @cached_property
+    def hook(self) -> SlackWebhookHook:
+        """Slack Incoming Webhook Hook."""
+        return SlackWebhookHook(
+            slack_webhook_conn_id=self.slack_webhook_conn_id, proxy=self.proxy, timeout=self.timeout
+        )
+
+    def notify(self, context):
+        """Send a message to a Slack Incoming Webhook."""
+        self.hook.send(
+            text=self.text,
+            blocks=self.blocks,
+            unfurl_links=self.unfurl_links,
+            unfurl_media=self.unfurl_media,
+            attachments=self.attachments,
+        )
+
+
+send_slack_webhook_notification = SlackWebhookNotifier

airflow/providers/slack/provider.yaml

@@ -84,3 +84,4 @@ connection-types:
 
 notifications:
   - airflow.providers.slack.notifications.slack.SlackNotifier
+  - airflow.providers.slack.notifications.slack_webhook.SlackWebhookNotifier

docs/apache-airflow-providers-slack/index.rst

@@ -43,7 +43,7 @@
 
     Connection Types <connections/index>
     Operators <operators/index>
-    Slack Notifications <notifications/slack_notifier_howto_guide>
+    Slack Notifications <notifications/index>
 
 .. toctree::
     :hidden:

docs/apache-airflow-providers-slack/notifications/index.rst

@@ -0,0 +1,30 @@
+ .. 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.
+
+
+
+Slack Notifications
+===================
+
+.. important:: This feature is only available in Airflow versions >= 2.6.0
+
+
+.. toctree::
+    :maxdepth: 1
+    :glob:
+
+    *

docs/apache-airflow-providers-slack/notifications/slackwebhook_notifier_howto_guide.rst

@@ -0,0 +1,59 @@
+ .. 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 Incoming Webhook notifications
+=====================================================
+
+Introduction
+------------
+Slack Incoming Webhook notifier (:class:`airflow.providers.slack.notifications.slack_webhook.SlackWebhookNotifier`)
+allows users to send messages to a slack channel thought `Incoming Webhook <https://api.slack.com/messaging/webhooks>`__
+using the various ``on_*_callbacks`` at both the DAG level and Task level
+
+You can also use a notifier with ``sla_miss_callback``.
+
+.. note::
+    When notifiers are used with `sla_miss_callback` the context will contain only values passed to the callback, refer :ref:`sla_miss_callback<concepts:sla_miss_callback>`.
+
+Example Code:
+-------------
+
+.. code-block:: python
+
+    from datetime import datetime, timezone
+    from airflow import DAG
+    from airflow.operators.bash import BashOperator
+    from airflow.providers.slack.notifications.slack_webhook import send_slack_webhook_notification
+
+    dag_failure_slack_webhook_notification = send_slack_webhook_notification(
+        slack_webhook_conn_id="slackwebhook", text="The dag {{ dag.dag_id }} failed"
+    )
+    task_failure_slack_webhook_notification = send_slack_webhook_notification(
+        slack_webhook_conn_id="slackwebhook",
+        text="The task {{ ti.task_id }} failed",
+    )
+
+    with DAG(
+        dag_id="mydag",
+        schedule="@once",
+        start_date=datetime(2023, 1, 1, tzinfo=timezone.utc),
+        on_failure_callback=[dag_failure_slack_webhook_notification],
+        catchup=False,
+    ):
+        BashOperator(
+            task_id="mytask", on_failure_callback=[task_failure_slack_webhook_notification], bash_command="fail"
+        )

tests/providers/slack/notifications/test_slack_webhook.py

@@ -0,0 +1,69 @@
+# 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
+
+from airflow.operators.empty import EmptyOperator
+from airflow.providers.slack.notifications.slack_webhook import (
+    SlackWebhookNotifier,
+    send_slack_webhook_notification,
+)
+
+
+class TestSlackNotifier:
+    def test_class_and_notifier_are_same(self):
+        assert send_slack_webhook_notification is SlackWebhookNotifier
+
+    @mock.patch("airflow.providers.slack.notifications.slack_webhook.SlackWebhookHook")
+    def test_slack_webhook_notifier(self, mock_slack_hook):
+        notifier = send_slack_webhook_notification(
+            text="foo-bar", blocks="spam-egg", attachments="baz-qux", unfurl_links=True, unfurl_media=False
+        )
+        notifier.notify({})
+        mock_slack_hook.return_value.send.assert_called_once_with(
+            text="foo-bar",
+            blocks="spam-egg",
+            unfurl_links=True,
+            unfurl_media=False,
+            attachments="baz-qux",
+        )
+
+    @mock.patch("airflow.providers.slack.notifications.slack_webhook.SlackWebhookHook")
+    def test_slack_webhook_templated(self, mock_slack_hook, dag_maker):
+        with dag_maker("test_send_slack_webhook_notification_templated") as dag:
+            EmptyOperator(task_id="task1")
+
+        notifier = send_slack_webhook_notification(
+            text="Who am I? {{ username }}",
+            blocks=[{"type": "header", "text": {"type": "plain_text", "text": "{{ dag.dag_id }}"}}],
+            attachments=[{"image_url": "{{ dag.dag_id }}.png"}],
+        )
+        notifier({"dag": dag, "username": "not-a-root"})
+        mock_slack_hook.return_value.send.assert_called_once_with(
+            text="Who am I? not-a-root",
+            blocks=[
+                {
+                    "type": "header",
+                    "text": {"type": "plain_text", "text": "test_send_slack_webhook_notification_templated"},
+                }
+            ],
+            attachments=[{"image_url": "test_send_slack_webhook_notification_templated.png"}],
+            unfurl_links=None,
+            unfurl_media=None,
+        )