AWS Lamba - POSTs GitHub commit status messages based on CodePipeline events

event_pattern.json

{
  "source": [
    "aws.codepipeline"
  ],
  "detail-type": [
    "CodePipeline Pipeline Execution State Change"
  ],
  "detail": {
    "state": [
      "STARTED",
      "SUCCEEDED",
      "FAILED"
    ]
  }
}

POSTCodePipelineGitHubCommitStatus.py

import boto3
import json
import logging
import os
import re
import urllib3


http = urllib3.PoolManager()
logger = logging.getLogger()


GITHUB_API = "https://api.github.com"

STATES = {"STARTED": "pending", "SUCCEEDED": "success", "FAILED": "failure"}


def handler(event, context):

    # set logging level
    logger.setLevel(logging.INFO)

    # log the received event
    logger.info("received event:")
    logger.info(json.dumps(event))

    # set up CodePipeline client
    cp = boto3.client("codepipeline", region_name=event["region"])

    # fetch the token for github from param store
    ssm = boto3.client("ssm")
    GITHUB_TOKEN = ssm.get_parameter(
        Name=os.environ["GITHUB_TOKEN_PARAMETER"], WithDecryption=True,
    )["Parameter"]["Value"]

    # fetch information on this CodePipeline execution
    logger.info(
        f"retrieving details for {event['detail']['pipeline']} {event['detail']['execution-id']}"
    )
    execution = cp.get_pipeline_execution(
        pipelineName=event["detail"]["pipeline"],
        pipelineExecutionId=event["detail"]["execution-id"],
    )
    logger.info(json.dumps(execution))

    # check that the source artifact is a known type
    valid_source = False
    commit_id = None
    repository = None
    logger.info("validating that source artifact has required information")
    for a in execution["pipelineExecution"]["artifactRevisions"]:
        if "revisionSummary" in a:
            if a["revisionSummary"].startswith("{") and a["revisionSummary"].endswith(
                "}"
            ):
                rs = json.loads(a["revisionSummary"])
                if rs["ProviderType"] == "GitHub":
                    logger.info("this is a GitHub type source")
                    commit_id = a["revisionId"]
                    repository = re.match(
                        r"^.*FullRepositoryId=(?P<repo>[^&]+).*$", a["revisionUrl"]
                    ).group("repo")
                    valid_source = True
                    break
    if not valid_source:
        raise Exception("Unable to locate source artifact with a known type")
    logger.info(f"repository: {repository}; commit_id: {commit_id}")

    # generate link to codepipeline execution
    deeplink = (
        f"https://{event['region']}.console.aws.amazon.com/codesuite/codepipeline/"
        f"pipelines/{event['detail']['pipeline']}/executions/"
        f"{event['detail']['execution-id']}/timeline?region={event['region']}"
    )
    logger.info(f"generated deeplink: {deeplink}")

    # use repository information to post the status
    endpoint = f"{GITHUB_API}/repos/{repository}/statuses/{commit_id}"
    logger.info(f"POST url: {endpoint}")
    data = {
        "state": STATES[event["detail"]["state"]],
        "context": "continuous-integration/codepipeline",
        "target_url": deeplink,
    }
    encoded_data = json.dumps(data).encode("utf-8")
    logger.debug(f"POST body: {encoded_data}")
    r = http.request(
        "POST",
        endpoint,
        body=encoded_data,
        headers={
            "Accept": "application/vnd.github.v3+json",
            "Authorization": f"token {GITHUB_TOKEN}",
            "Content-Type": "application/json; charset=utf-8",
            "User-Agent": context.invoked_function_arn,
        },
    )
    logger.debug(f"{r.status} -- {r.data.decode('utf-8')}")
    if r.status != 201:
        raise Exception(f"expected 201 response from GitHub, got {r.status}")

    # all done
    logger.info("all done")
    return