Source code for airflow.providers.amazon.aws.hooks.chime
## 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__importannotationsimportjsonimportrefromfunctoolsimportcached_propertyfromtypingimportAnyfromairflow.exceptionsimportAirflowExceptionfromairflow.providers.http.hooks.httpimportHttpHook
[docs]classChimeWebhookHook(HttpHook):""" Interact with Amazon Chime Webhooks to create notifications. .. warning:: This hook is only designed to work with web hooks and not chat bots. :param chime_conn_id: :ref:`Amazon Chime Connection ID <howto/connection:chime>` with Endpoint as `https://hooks.chime.aws` and the webhook token in the form of ``{webhook.id}?token{webhook.token}`` """
def_get_webhook_endpoint(self,conn_id:str)->str:""" Given a Chime conn_id return the default webhook endpoint. :param conn_id: The provided connection ID. :return: Endpoint(str) for chime webhook. """conn=self.get_connection(conn_id)token=conn.get_password()iftokenisNone:raiseAirflowException("Webhook token field is missing and is required.")url=conn.schema+"://"+conn.hostendpoint=url+token# Check to make sure the endpoint matches what Chime expectsifnotre.fullmatch(r"[a-zA-Z0-9_-]+\?token=[a-zA-Z0-9_-]+",token):raiseAirflowException("Expected Chime webhook token in the form of '{webhook.id}?token={webhook.token}'.")returnendpointdef_build_chime_payload(self,message:str)->str:""" Build payload for Chime and ensures messages do not exceed max length allowed. :param message: The message you want to send to your Chime room. (max 4096 characters) """payload:dict[str,Any]={}# We need to make sure that the message does not exceed the max length for Chimeiflen(message)>4096:raiseAirflowException("Chime message must be 4096 characters or less.")payload["Content"]=messagereturnjson.dumps(payload)
[docs]defsend_message(self,message:str)->None:""" Execute calling the Chime webhook endpoint. :param message: The message you want to send to your Chime room.(max 4096 characters) """chime_payload=self._build_chime_payload(message)self.run(endpoint=self.webhook_endpoint,data=chime_payload,headers={"Content-type":"application/json"})
@classmethod
[docs]defget_ui_field_behaviour(cls)->dict[str,Any]:"""Return custom field behaviour to only get what is needed for Chime webhooks to function."""return{"hidden_fields":["login","port","extra"],"relabeling":{"host":"Chime Webhook Endpoint","password":"Chime Webhook token",},"placeholders":{"schema":"https","host":"hooks.chime.aws/incomingwebhook/","password":"T00000000?token=XXXXXXXXXXXXXXXXXXXXXXXX",},}