Add ChimeWebhookHook

airflow/providers/amazon/aws/hooks/chime.py

@@ -0,0 +1,104 @@
+#
+# 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.
+
+"""This module contains a web hook for Chime."""
+from __future__ import annotations
+
+import json
+import re
+from typing import Any
+
+from airflow.exceptions import AirflowException
+from airflow.providers.http.hooks.http import HttpHook
+
+
+class ChimeWebhookHook(HttpHook):
+    """Interact with Chime Web Hooks to create notifications.
+
+    .. warning:: This hook is only designed to work with Web Hooks and not Chatbots.
+
+    :param http_conn_id: Http connection ID with host as "https://hooks.chime.aws" and
+                         default webhook endpoint in the extra field in the form of
+                         {"webhook_endpoint": "incomingwebhooks/{webhook.id}?token{webhook.token}"}
+    :param webhook_endpoint: Chime webhook endpoint in the form of
+                             "incomingwebhooks/{webhook.id}?token={webhook.token}"
+    :param message: The message you want to send to your Chime room.
+                    (max 4096 characters)
+    """
+
+    conn_name_attr = "http_conn_id"
+    default_conn_name = "chime_default"
+    conn_type = "chime"
+    hook_name = "Chime Web Hook"
+
+    def __init__(
+        self,
+        http_conn_id: str | None = None,
+        webhook_endpoint: str | None = None,
+        *args: Any,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(*args, **kwargs)
+        self.webhook_endpoint = self._get_webhook_endpoint(http_conn_id, webhook_endpoint)
+
+    def _get_webhook_endpoint(self, http_conn_id: str | None, webhook_endpoint: str | None) -> str:
+        """
+        Given a Chime http_conn_id return the default webhook endpoint or override if
+        webhook_endpoint is manually provided.
+
+        :param http_conn_id: The provided connection ID
+        :param webhook_endpoint: The manually provided webhook endpoint
+        :return: Webhook Endpoint(str) to use with Chime
+        """
+        if webhook_endpoint:
+            endpoint = webhook_endpoint
+        elif http_conn_id:
+            conn = self.get_connection(http_conn_id)
+            extra = conn.extra_dejson
+            endpoint = extra.get("webhook_endpoint", "")
+        else:
+            raise AirflowException(
+                "Missing one of http_conn_id or webhook_endpoint arguments which are required."
+            )
+
+        # Check to make sure the endpoint matches what Chime expects
+        if not re.match("^incomingwebhooks/[a-zA-Z0-9_-]+\?token=[a-zA-Z0-9_-]+$", endpoint):
+            raise AirflowException(
+                "Expected Chime webhook endpoint in the form of "
+                '"incomingwebhooks/{webhook.id}?token={webhook.token}".'
+            )
+
+        return endpoint
+
+    def _build_chime_payload(self, message: str) -> str:
+        """Builds payload for Chime and ensures messages do not exceed max length allowed."""
+        payload: dict[str, Any] = {}
+        # We need to make sure that the message does not exceed the max length for Chime
+        if len(message) <= 4096:
+            payload["Content"] = message
+        else:
+            raise AirflowException("Chime message must be 4096 characters or less.")
+
+        return json.dumps(payload)
+
+    def send_message(self, message: str) -> None:
+        """Execute calling the Chime webhook endpoint."""
+        chime_payload = self._build_chime_payload(message)
+        self.run(
+            endpoint=self.webhook_endpoint, data=chime_payload, headers={"Content-type": "application/json"}
+        )

airflow/providers/amazon/provider.yaml

@@ -406,6 +406,9 @@ hooks:
   - integration-name: Amazon Athena
     python-modules:
       - airflow.providers.amazon.aws.hooks.athena
+  - integration-name: Amazon Chime
+    python-modules:
+      - airflow.providers.amazon.aws.hooks.chime
   - integration-name: Amazon DynamoDB
     python-modules:
       - airflow.providers.amazon.aws.hooks.dynamodb
@@ -624,6 +627,8 @@ connection-types:
     connection-type: emr
   - hook-class-name: airflow.providers.amazon.aws.hooks.redshift_sql.RedshiftSQLHook
     connection-type: redshift
+  - hook-class-name: airflow.providers.amazon.aws.hooks.chime.ChimeWebhookHook
+    connection-type: chime
 
 secrets-backends:
   - airflow.providers.amazon.aws.secrets.secrets_manager.SecretsManagerBackend

docs/apache-airflow-providers-amazon/connections/chime.rst

@@ -0,0 +1,74 @@
+.. 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.
+
+.. _howto/connection:chime:
+
+Amazon Chime Connection
+==========================
+
+The Chime connection works with calling Chime webhooks to send messages to a chime room.
+
+Authenticating to Amazon Chime
+---------------------------------
+When a webhook is created in a Chime room a token will be included in the url for authentication.
+
+
+Default Connection IDs
+----------------------
+
+The default connection ID is ``chime_default``.
+
+Configuring the Connection
+--------------------------
+
+Login (optional)
+    Chime does not require a login for a webhook, this field can be left blank.
+
+Password (optional)
+    The token for authentication should be included in extras. No passwords are used for Chime webhooks.
+
+Host (optional)
+    Specify the entire url or the base of the url for the service.
+
+Port (optional)
+    Specify a port number if applicable.
+
+Schema (optional)
+    Specify the service type etc: http/https.
+
+Extras (optional)
+    Specify webhook_endpoint here which will start with ``incomingwebhooks/``
+If you are configuring the connection via a URI, ensure that all components of the URI are URL-encoded.
+
+Examples
+--------
+
+**Connection**
+
+* **Login**:
+* **Password**:
+* **Host**: hooks.chime.aws
+* **Port**:
+* **Schema**: https
+
+* **Extras**:
+
+.. code-block:: json
+
+    {
+      "webhook_endpoint": "incomingwebhooks/abceasd-3423-a1237-ffff-000cccccccc?token=somechimetoken"
+    }

generated/provider_dependencies.json

@@ -40,6 +40,7 @@
       "exasol",
       "ftp",
       "google",
+      "http",
       "imap",
       "mongo",
       "salesforce",

tests/providers/amazon/aws/hooks/test_chime.py

@@ -0,0 +1,106 @@
+#
+# 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
+
+import pytest
+
+from airflow.exceptions import AirflowException
+from airflow.models import Connection
+from airflow.providers.amazon.aws.hooks.chime import ChimeWebhookHook
+from airflow.utils import db
+
+
+class TestChimeWebhookHook:
+
+    _config = {
+        "http_conn_id": "default-chime-webhook",
+        "webhook_endpoint": "incomingwebhooks/abcd-1134?token=somechimetoken_111",
+        "message": "your message here",
+    }
+
+    expected_payload_dict = {
+        "Content": _config["message"],
+    }
+
+    expected_payload = json.dumps(expected_payload_dict)
+
+    def setup_method(self):
+        db.merge_conn(
+            Connection(
+                conn_id="default-chime-webhook",
+                conn_type="chime",
+                host="https://hooks.chime.aws",
+                extra='{"webhook_endpoint": "incomingwebhooks/abcd-1134?token=somechimetoken_111"}',
+            )
+        )
+
+    def test_get_webhook_endpoint_manual_token(self):
+        # Given
+        provided_endpoint = "incomingwebhooks/abcd-1134?token=somechimetoken_111"
+        hook = ChimeWebhookHook(webhook_endpoint=provided_endpoint)
+
+        # When
+        webhook_endpoint = hook._get_webhook_endpoint(None, provided_endpoint)
+
+        # Then
+        assert webhook_endpoint == provided_endpoint
+
+    def test_get_webhook_endpoint_invalid_url(self):
+        # Given
+        provided_endpoint = "https://hooks.chime.aws/some-invalid-webhook-url"
+
+        # When/Then
+        expected_message = "Expected Chime webhook endpoint in the form of"
+        with pytest.raises(AirflowException, match=expected_message):
+            ChimeWebhookHook(webhook_endpoint=provided_endpoint)
+
+    def test_get_webhook_endpoint_conn_id(self):
+        # Given
+        conn_id = "default-chime-webhook"
+        hook = ChimeWebhookHook(http_conn_id=conn_id)
+        expected_webhook_endpoint = "incomingwebhooks/abcd-1134?token=somechimetoken_111"
+
+        # When
+        webhook_endpoint = hook._get_webhook_endpoint(conn_id, None)
+
+        # Then
+        assert webhook_endpoint == expected_webhook_endpoint
+
+    def test_build_chime_payload(self):
+        # Given
+        hook = ChimeWebhookHook(**self._config)
+
+        # When
+        payload = hook._build_chime_payload()
+
+        # Then
+        assert self.expected_payload == payload
+
+    def test_build_chime_payload_message_length(self):
+        # Given
+        config = self._config.copy()
+        # create message over the character limit
+        config["message"] = "c" * 4097
+        hook = ChimeWebhookHook(**config)
+
+        # When/Then
+        expected_message = "Chime message must be 4096 characters or less."
+        with pytest.raises(AirflowException, match=expected_message):
+            hook._build_chime_payload()