Simple app (GCP function) to add link to PR with Kanbanize card number to the card

github_kanbanize.py

import hashlib
import hmac
import logging
import os
import re
import sys
from typing import Dict

import flask
import requests
from flask import Flask, request

log = logging.getLogger(__name__)
stdout = logging.StreamHandler(sys.stdout)
stdout.setLevel(logging.DEBUG)
log.addHandler(stdout)
log.setLevel(logging.DEBUG)
log.propagate = False

try:
    KANBANIZE_API_TOKEN = os.environ["KANBANIZE_API_TOKEN"]
except KeyError:
    log.critical("ERROR: Set KANBANIZE_API_TOKEN env var")
    exit(1)

try:
    KANBANIZE_DOMAIN_PREFIX = os.environ["KANBANIZE_DOMAIN_PREFIX"]
except KeyError:
    log.critical("ERROR: Set KANBANIZE_DOMAIN_PREFIX env var")
    exit(1)

KANBANIZE_PR_FIELD = "Github PR"
GITHUB_WEBHOOK_SECRET = os.environ.get("GITHUB_WEBHOOK_SECRET", None)

app = Flask(__name__)
kanbanize_task_re = re.compile(r"KB(\d+)", re.IGNORECASE)


# Defines a handler on the /webhook url. Used in local development
@app.route("/webhook", methods=["POST"])
def local_dev():
    return webhook_handler(request)


def process_pr(d: Dict):
    log.info(f"Got PR event for PR {d['number']}.")
    if (
        d["action"] == "opened"
        or (
            d["action"] == "edited"
            and d["changes"]["title"]["from"] != d["pull_request"]["title"]
        )
    ):
        title = d["pull_request"]["title"]
        match = kanbanize_task_re.search(title)
        if match:
            kanbanize_card_id = match.group(1)
            log.info(f"Found card id {kanbanize_card_id} in {title}.")
            add_pr_url_to_kanbanize(kanbanize_card_id, d["pull_request"]["html_url"])
            log.info(f"Set url for PR {d['number']} on task {kanbanize_card_id}.")
        else:
            log.info(f"No taskid found in PR title: \"{d['pull_request']['title']}\".")
    else:
        log.info("Uninteresting PR event, skipped.")


def add_pr_url_to_kanbanize(card_id, url):
    response = requests.post(
        (
            f"http://{KANBANIZE_DOMAIN_PREFIX}.kanbanize.com/index.php"
            "/api/kanbanize/edit_custom_fields"
        ),
        headers={"apikey": KANBANIZE_API_TOKEN},
        json={
            "cardid": card_id, "fields": [{"name": KANBANIZE_PR_FIELD, "value": url}]
        },
    )
    if not response.ok:
        log.error(
            f"Unexpected {response.status_code} response from kanbanize: "
            f"{response.content}"
        )


def request_signature_valid(req: flask.Request, secret):
    digest = hmac.new(bytes(secret), request.data, hashlib.sha1).hexdigest()
    sig_parts = req.headers.get("X-Hub-Signature", "").split("=", 1)
    return (
        len(sig_parts) == 2
        and sig_parts[0] == "sha1"
        and hmac.compare_digest(sig_parts[1], digest)
    )


def webhook_handler(req: flask.Request):
    """
    Handle the actual request.

    Accept a request object instead of using a global :roll:. This function is
    the one used by Google Cloud Functions.
    """
    event_type = req.headers.get("X-Github-Event")
    if not event_type:
        return "Missing event header", 400

    if event_type != "pull_request":
        return "All events besides 'pull_request' are ignored", 200

    if GITHUB_WEBHOOK_SECRET and request_signature_valid(req, GITHUB_WEBHOOK_SECRET):
        return "Invalid request digest", 401

    data = req.get_json()
    process_pr(data)

    return "", 204