Automatisierung mit Azure Functions

def get_webhook():
    data = clickup.get_webhooks()
    for webhook in data["webhooks"]:
        if "endpoint" in webhook and webhook["endpoint"] == "myEndpointURL":
            return webhook
    return None

def create_webhook():
    # this webhook should be triggered when a task is created
    values = {
        "endpoint": "myEndpointURL",
        "events": [
            "taskCreated"
        ]   
    }
    
    # clickup is a wrapper around the API calls 
    resp = clickup.post_webhook(json.dumps(values))
    if resp.status_code == 200:
        data = json.loads(resp.text)
        if "webhook" in data:
            return data["webhook"]
    return None

def cleanup_webhooks():
    # clickup is a wrapper around the API calls
    data = clickup.get_webhooks()
    for webhook in data["webhooks"]:
        if "endpoint" in webhook and webhook["enpoint"] == "myEndpointURL":
            clickup.delete_webhook(webhook["id"])

token = os.getenv("TOKEN", "")
urlaub_list_id = os.getenv("URLAUB_LIST_ID", "")
bitperfect_space_id = os.getenv("BITPERFECT_SPACE_ID", "")

clickup = ClickupAPI(token, bitperfect_space_id)
task_webhook = get_webhook()
if not task_webhook:
    task_webhook = create_webhook()
    if not task_webhook:
        cleanup_webhooks()
        raise EnvironmentError("could not establish webhook")

def verify_signature(msg, sec, sig):
    message = bytes(msg, 'utf-8')
    secret = bytes(sec, 'utf-8')
    hash = hmac.new(secret, message, hashlib.sha256)
    return sig == hash.hexdigest()

def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info("Function is beeing executed")
    signature = req.headers.get("X-Signature")
    body = req.get_json()
    if signature:
        if not verify_signature(json.dumps(body, separators=(',', ':'), ensure_ascii=False), task_webhook["secret"], signature):
            logging.warning(f"Request with Invalid Signature: {req}")
            # to prevent external services to try accessing the endpoint again we send status_code 200
            return func.HttpResponse("Signature Invalid", status_code=200)
        else:
            request_handler = AbscenseRequestHandler(body, clickup, list_id)
            if request_handler.handle():
                logging.info("Request handled successfully")
                return func.HttpResponse("ok", status_code=200)
            else:
                logging.warning(f"Error while handling request: {body}")
                # this is status_code 200 because otherwise we would receive the same request again
                return func.HttpResponse("unable to handle request", status_code=200)
    else:
        logging.warning("Request without Signature received")
        return func.HttpResponse("Unauthorized", status_code=401)

class AbscenseRequestHandler:
    def __init__(self, request_body, clickup, urlaub_list_id):
        self.request_body = request_body
        self.clickup: ClickupAPI = clickup
        self.urlaub_list_id = urlaub_list_id
        self.urlaub_keywords = ["urlaub", "holiday"]
        self.illness_keywords = ["krank", "krankenstand", "illness"]
        self.ef_keywords = ["ef", "erlaubte fehlstunden", "fehlstunden", "erlaubt", "geburt", "umzug", "tod", "beerdigung"]

    def handle(self):
        if self.is_in_abscence_list():
            task = self.clickup.get_task(self.request_body["task_id"])
            if not self.set_tag(task):
                return False
            if not self.add_assignee(task):
                return False
            return True
        else:
            return False

    def set_tag(self, task):
        if self.is_in_list(self.urlaub_keywords, task["name"].lower()):
            return self.clickup.add_tag(task["id"], "urlaub")
        elif self.is_in_list(self.illness_keywords, task["name"].lower()):
            return self.clickup.add_tag(task["id"], "krankenstand")
        elif self.is_in_list(self.ef_keywords, task["name"].lower()):
            return self.clickup.add_tag(task["id"], "ef")
        else:
            return True

    @staticmethod
    def is_in_list(keywords, name):
        for keyword in keywords:
            if keyword in name:
                return True
        return False

    def add_assignee(self, task):
        if not task["assignees"]:
            return self.clickup.update_tasks(task["id"], json.dumps({"assignees": {"add": [task["creator"]["id"]]}}))
        return True

    def is_in_abscence_list(self):
        if "history_items" in self.request_body:
            for entry in self.request_body["history_items"]:
                return "parent_id" in entry and entry["parent_id"] == self.urlaub_list_id
        return False