Add Google Chat notification handler; Allow more than one handler to be used at a time; Fix MS teams notification test; Fix failing GitHub build; Prevent a job from failing when coverage step fails;

.github/workflows/python-package.yml

@@ -26,9 +26,10 @@ jobs:
           source venv/bin/activate
           pip install --upgrade pip
 
+      # --only-binary added to fix issue when building google-re2 from source (see note: https://airflow.apache.org/docs/apache-airflow/2.10.5/start.html#quick-start)
       - name: Check pre-commit status
         run: |
-          pip install .[tests]
+          pip install --only-binary=google-re2 .[tests]
           pip freeze
           pipdeptree
           pre-commit run --all-files
@@ -37,8 +38,10 @@ jobs:
         run: |
           tox -e py38
 
+      # TODO: Fix coverage report. For now, fails are ignored using "continue-on-error: true"
       - name: Report coverage
         uses: paambaati/codeclimate-action@v2.7.5
+        continue-on-error: true
         env:
           CC_TEST_REPORTER_ID: 2f513a15560b9848db21cca75606745b5283b68b5935e8c481c25eb0090a2a36
         with:

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## [Unreleased]
 
+- Add `Google Chat` notifications handler
+
 ## [0.35.0] - 2023-09-08
 
 ## [0.34.0] - 2023-08-10

dbt_airflow_factory/notifications/handler.py

@@ -17,6 +17,8 @@
 else:
     from airflow.hooks.base_hook import BaseHook
 
+from airflow.providers.http.hooks.http import HttpHook
+
 from dbt_airflow_factory.notifications.ms_teams_webhook_operator import (
     MSTeamsWebhookOperator,
 )
@@ -28,7 +30,7 @@ def failure_handler(context: Any) -> None:
             for handler_definition in handlers_config:
                 if handler_definition["type"] == "slack":
                     connection = BaseHook.get_connection(handler_definition["connection_id"])
-                    return SlackWebhookOperator(
+                    SlackWebhookOperator(
                         task_id="slack_failure_notification",
                         message=handler_definition["message_template"].format(
                             task=context.get("task_instance").task_id,
@@ -67,6 +69,35 @@ def failure_handler(context: Any) -> None:
                         theme_color="FF0000",
                         http_conn_id=handler_definition["connection_id"],
                     )
-                    return teams_notification.execute(context)
+                    teams_notification.execute(context)
+                elif handler_definition["type"] == "google_chat":
+                    import json
+                    import logging
+
+                    webserver_url = handler_definition["webserver_url"]
+                    webserver_url = (
+                        webserver_url[:-1] if webserver_url.endswith("/") else webserver_url
+                    )
+                    dag_id = context.get("task_instance").dag_id
+                    task_id = context.get("task_instance").task_id
+                    query = quote_plus(
+                        f"log?dag_id={dag_id}&task_id={task_id}&execution_date={context['ts']}",
+                        safe="=&?",
+                    )
+                    logs_url = f"{webserver_url}/{query}"
+
+                    message_text = handler_definition["message_template"].format(
+                        task_id=task_id,
+                        dag_id=dag_id,
+                        execution_date=context.get("execution_date"),
+                        log_url=logs_url,
+                    )
+                    message_body = {"text": message_text}
+
+                    HttpHook(http_conn_id=handler_definition["connection_id"]).run(
+                        data=json.dumps(message_body),
+                        headers={"Content-Type": "application/json; charset=UTF-8"},
+                    )
+                    logging.info("Webhook request sent to Google Chat")
 
         return failure_handler

docs/configuration.rst

@@ -50,8 +50,8 @@ airflow.yml file
      -
      - empty list
      - Each item of the list contains configuration of notifications handler in case Tasks or DAG fails. Each item is a dictionary with following fields
-       ``type`` (type of handler eg. *slack* or *teams*), ``webserver_url`` (Airflow Webserver URL), ``connection_id`` (id of the Airflow's connection) and ``message_template`` that will be sent.
-       More on how to configure the webhooks can be found here: `Slack <https://airflow.apache.org/docs/apache-airflow-providers-slack/stable/_api/airflow/providers/slack/operators/slack_webhook/index.html>`_ or `MS Teams <https://code.mendhak.com/Airflow-MS-Teams-Operator/#copy-hook-and-operator>`_
+       ``type`` (type of handler eg. *slack*, *teams* or *google_chat*), ``webserver_url`` (Airflow Webserver URL), ``connection_id`` (id of the Airflow's connection) and ``message_template`` that will be sent.
+       More on how to configure the webhooks can be found here: `Slack <https://airflow.apache.org/docs/apache-airflow-providers-slack/stable/_api/airflow/providers/slack/operators/slack_webhook/index.html>`_, `MS Teams <https://code.mendhak.com/Airflow-MS-Teams-Operator/#copy-hook-and-operator>`_ or `Google Chat <https://developers.google.com/workspace/chat/quickstart/webhooks>`_
    * - enable_project_dependencies
      - boolean
      -

docs/features.rst

@@ -62,14 +62,17 @@ Any other Airflow template variables will not work in ``airflow.yml``.
 
 Notifications
 +++++++++++++++++++
-It is possible to configure notifications in case of task failure. Currently, the only available channels are Slack and MS Teams.
+It is possible to configure notifications in case of task failure. Currently, the only available channels are Slack, MS Teams and Google Chat.
 
 .. image:: images/slack_notification.png
    :width: 800
 
 .. image:: images/msteams_notification.png
    :width: 800
 
+.. image:: images/google_chat_notification.png
+   :width: 800
+
 Source dependencies
 +++++++++++++++++++
 Dividing the DBT project into smaller pieces or using data produced by other teams usually needs to wait for

tests/config/notifications_google_chat/airflow.yml

@@ -0,0 +1,11 @@
+
+failure_handlers:
+  - type: google_chat
+    webserver_url: https://your.airflow-webserver.url
+    connection_id: google_chat_failure
+    message_template: |
+        🔴 *Task Failed*
+        *Task*: {task_id}
+        *Dag*: {dag_id}
+        *Execution Time*: {execution_date}
+        *Log Url*: {log_url}

tests/test_notifications.py

@@ -25,6 +25,7 @@
     (
         "notifications_slack",
         "notifications_teams",
+        "notifications_google_chat",
     ),
 )
 def test_notification_callback_creation(config_dir):
@@ -102,7 +103,7 @@ def test_notification_send_for_teams(mock_hook_run, mock_get_connection):
     # then
     request = mock_hook_run.call_args_list[0].kwargs
     webhook_post_data = json.loads(request["data"].replace("\n", "").replace(" ", ""))
-    assert mock_hook_run.called_once
+    mock_hook_run.assert_called_once()
     assert webhook_post_data == webhook_expected_payload
 
 
@@ -125,7 +126,41 @@ def create_teams_connection():
     return connection
 
 
+def create_google_chat_connection():
+    connection = Connection(
+        **{
+            "login": None,
+            "password": None,
+            "conn_type": "http",
+            "host": "google.com/webhook_endpoint",
+            "schema": "https",
+        }
+    )
+    return connection
+
+
 def create_context():
     task_instance = MagicMock()
     task_instance.configure_mock(**{"task_id": "task_id", "dag_id": "dag_id", "log_url": "log_url"})
     return {"task_instance": task_instance, "execution_date": "some date", "ts": "ts"}
+
+
+@patch(
+    "airflow.hooks.base.BaseHook.get_connection"
+    if IS_AIRFLOW_NEWER_THAN_2_4
+    else "airflow.hooks.base_hook.BaseHook.get_connection"
+)
+@patch("dbt_airflow_factory.notifications.handler.HttpHook")
+def test_notification_send_for_google_chat(mock_http_hook, mock_get_connection):
+    # given
+    notifications_config = AirflowDagFactory(
+        path.dirname(path.abspath(__file__)), "notifications_google_chat"
+    ).airflow_config["failure_handlers"]
+    factory = NotificationHandlersFactory()
+    context = create_context()
+    mock_get_connection.return_value = create_google_chat_connection()
+    # when
+    factory.create_failure_handler(notifications_config)(context)
+
+    # then
+    mock_http_hook.assert_called_once()

tox.ini

@@ -2,6 +2,9 @@
 envlist = py38
 
 [testenv]
+# install_command added to fix issue when building google-re2 from source (see note: https://airflow.apache.org/docs/apache-airflow/2.10.5/start.html#quick-start)
+install_command =
+    python -I -m pip install --upgrade pip --only-binary=google-re2 {opts} {packages}
 extras =
     tests
 commands=