Scripts for downloading a Github actions workflow artifact

gh-get-artifacts-dev.py

import json
import zipfile
from typing import Union, Optional

import requests

GITHUB_OAUTH_TOKEN = ''
OWNER = ''
REPO = ''


def gh_request(rest_url, params: Optional[dict] = None):
    headers = {'Authorization': 'token {}'.format(GITHUB_OAUTH_TOKEN), 'Accept': 'application/vnd.github.v3+json'}
    return requests.get('https://api.github.com{}'.format(rest_url), headers=headers, params=params)


def gh_get_last_workflow_run(workflow_id: Union[str, int], branch: str) -> dict:
    # Only get the last successful run at the specified branch
    params = {
        'status': 'success',
        'per_page': 1,
        'branch': branch,
    }
    response = gh_request('/repos/{}/{}/actions/workflows/{}/runs'.format(OWNER, REPO, workflow_id, branch),
                          params=params)
    if response.status_code != 200:
        raise Exception('TODO')
    # print(json.dumps(response.json(), indent=4))
    return response.json()['workflow_runs'][0]


def gh_get_workflow_run_artifact(run_id: int, name: str) -> Optional[dict]:
    response = gh_request('/repos/{}/{}/actions/runs/{}/artifacts'.format(OWNER, REPO, run_id))
    if response.status_code != 200:
        raise Exception('TODO')
    # print(json.dumps(response.json(), indent=4))

    artifacts = response.json()['artifacts']
    for a in artifacts:
        if a['name'] == name:
            if a['expired']:
                print('WARN: Artifact has expired: {}'.format(a['expires_at']))
                return None
            return a
    return None


def gh_download_artifact(artifact_id: int, out_filename: str) -> str:
    response = gh_request('/repos/{}/{}/actions/artifacts/{}/{}'.format(OWNER, REPO, artifact_id, 'zip'))
    if response.status_code != 200:
        raise Exception('TODO')

    zip_filename = '{}.zip'.format(out_filename)
    with open(zip_filename, 'wb') as f:
        f.write(response.content)

    return zip_filename


if __name__ == '__main__':
    print('Getting latest run id')
    last_workflow_run = gh_get_last_workflow_run('ccpp.yml', 'dev')

    sha256 = last_workflow_run['head_sha']
    run_id = last_workflow_run['id']
    print('SHA: {}, run id: {}'.format(sha256, run_id))

    binaries_name = 'binaries-{}'.format(sha256)
    print('Getting "{}" artifact id'.format(binaries_name))
    artifact = gh_get_workflow_run_artifact(run_id, binaries_name)
    artifact_id = artifact['id']
    print('Artifact id: {}'.format(artifact_id))

    print('Downloading binaries')
    zip_filename = gh_download_artifact(artifact_id, binaries_name)

    with zipfile.ZipFile(zip_filename, 'r') as zip_ref:
        zip_ref.extractall('binaries')

gh-get-artifacts-dev.sh

#!/bin/bash -e

TOKEN="$GITHUB_OAUTH_TOKEN" # https://github.com/settings/tokens
OWNER="$1"
REPO="$2"

if [ -z "$TOKEN" ]; then
  >&2 echo -e "\033[33mWARNING: \$TOKEN is empty\033[0m"
fi

function gh_curl() {
  API_GITHUB="https://api.github.com"
  curl ${TOKEN:+-H "Authorization: token $TOKEN"} \
       -H "Accept: application/vnd.github.v3+json" \
       -sL $API_GITHUB$@
}

function gh_get_last_workflow_run_id() {
  workflow_id="$1"
  branch="$2"

  run_id=$(gh_curl "/repos/$OWNER/$REPO/actions/workflows/$workflow_id/runs?branch=$branch&status=success&per_page=1" | jq ".workflow_runs[0].id")
  if [ "$run_id" = "null" ]; then
    >&2 echo -e "\033[31mERROR: Could not find run id for workflow \"$workflow_id\" on branch \"$branch\"\033[0m"
    exit 1
  fi

  echo $run_id
}

function gh_get_workflow_run_artifact_id() {
  run_id="$1"
  name="$2"

  artifact=$(gh_curl "/repos/$OWNER/$REPO/actions/runs/$run_id/artifacts" | jq ".artifacts | map(select(.name | contains(\"$name\")))[0]")
  expired=$(echo $artifact | jq ".expired")
  if [ "$expired" = "true" ]; then
    >&2 echo -e "\033[31mERROR: Artifact with name \"$name\" has expired\033[0m"
    exit 1
  fi

  artifact_id=$(echo $artifact | jq ".id")
  if [ "$artifact_id" = "null" ]; then
    >&2 echo -e "\033[31mERROR: Could not find artifact with name \"$name\" for run id \"$run_id\"\033[0m"
    exit 1
  fi

  echo $artifact_id
}

function gh_download_artifact() {
  artifact_id="$1"
  file="$2"
  archive_format="zip"

  # Download the artifact and store it as a zip file
  gh_curl "/repos/$OWNER/$REPO/actions/artifacts/$artifact_id/$archive_format" -o "$file".zip
}

echo "Getting latest run id"
run_id=$(gh_get_last_workflow_run_id "ccpp.yml" "dev")
echo "Run id:" $run_id

echo "Getting binaries artifact id"
artifact_id=$(gh_get_workflow_run_artifact_id "$run_id" "binaries")
echo "Artifact id:" $artifact_id

echo "Downloading files"
gh_download_artifact "$artifact_id" "binaries"

# Remove any old directory and unzip the latest binaries into a directory with the same name
rm -rf binaries
unzip binaries.zip -d binaries