Manually cancel previous GitHub Action workflow runs in queue

cancel_github_workflows.py

#!/usr/bin/env python3
# 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.
"""
Manually cancel previous GitHub Action workflow runs in queue.

Example:
  # Set up
  export GITHUB_TOKEN=394ba3b48494ab8f930fbc93
  export GITHUB_REPOSITORY=apache/incubator-superset

  # cancel previous jobs for a PR
  ./cancel_github_workflows.py 1042

  # cancel previous jobs for a branch
  ./cancel_github_workflows.py my-branch

  # cancel all jobs
  ./cancel_github_workflows.py 1024 --include-last
"""
import os
from typing import Iterable, List, Optional, Union

import click
import requests
from click.exceptions import ClickException
from dateutil import parser
from typing_extensions import Literal

github_token = os.environ.get("GITHUB_TOKEN")
github_repo = os.environ.get("GITHUB_REPOSITORY", "apache/incubator-superset")


def request(method: Literal["GET", "POST", "DELETE", "PUT"], endpoint: str, **kwargs):
    resp = requests.request(
        method,
        f"https://api.github.com/{endpoint.lstrip('/')}",
        headers={"Authorization": f"Bearer {github_token}"},
        **kwargs,
    ).json()
    if "message" in resp:
        raise ClickException(f"{endpoint} >> {resp['message']} <<")
    return resp


def list_runs(repo: str, params=None):
    return request("GET", f"/repos/{repo}/actions/runs", params=params)


def cancel_run(repo: str, run_id: Union[str, int]):
    return request("POST", f"/repos/{repo}/actions/runs/{run_id}/cancel")


def get_pull_request(repo: str, pull_number: Union[str, int]):
    return request("GET", f"/repos/{repo}/pulls/{pull_number}")


def get_runs_by_branch(
    repo: str,
    branch: str,
    user: Optional[str] = None,
    statuses: Iterable[str] = ("queued", "in_progress"),
    events: Iterable[str] = ("pull_request", "push"),
):
    """Get workflow runs associated with the given branch"""
    return [
        item
        for event in events
        for status in statuses
        for item in list_runs(repo, {"event": event, "status": status})["workflow_runs"]
        if item["head_branch"] == branch
        and (user is None or (user == item["head_repository"]["owner"]["login"]))
    ]


def print_commit(commit):
    """Print out commit message for verification"""
    indented_message = "    \n".join(commit["message"].split("\n"))
    date_str = (
        parser.parse(commit["timestamp"])
        .astimezone(tz=None)
        .strftime("%a, %d %b %Y %H:%M:%S")
    )
    print(
        f"""HEAD {commit["id"]}
Author: {commit["author"]["name"]} <{commit["author"]["email"]}>
Date:   {date_str}

    {indented_message}
"""
    )


@click.command()
@click.option(
    "--repo",
    default=github_repo,
    help="The github repository name. For example, apache/incubator-superset.",
)
@click.option(
    "--event",
    type=click.Choice(["pull_request", "push", "issue"]),
    default=["pull_request", "push"],
    show_default=True,
    multiple=True,
)
@click.option(
    "--include-last/--skip-last",
    default=False,
    show_default=True,
    help="Whether to also cancel the lastest run.",
)
@click.option(
    "--include-running/--skip-running",
    default=True,
    show_default=True,
    help="Whether to also cancel running workflows.",
)
@click.argument("branch_or_pull")
def cancel_github_workflows(
    branch_or_pull: str,
    repo,
    event: List[str],
    include_last: bool,
    include_running: bool,
):
    """Cancel running or queued GitHub workflows by branch or pull request ID"""
    if not github_token:
        raise ClickException("Please provide GITHUB_TOKEN as an env variable")

    statuses = ("queued", "in_progress") if include_running else ("queued",)
    pr = None

    if branch_or_pull.isdigit():
        pr = get_pull_request(repo, pull_number=branch_or_pull)
        target_type = "pull request"
        title = f"#{pr['number']} - {pr['title']}"
    else:
        target_type = "branch"
        title = branch_or_pull

    print(
        f"\nCancel {'active' if include_running else 'previous'} "
        f"workflow runs for {target_type}\n\n    {title}\n"
    )

    if pr:
        # full branch name
        runs = get_runs_by_branch(
            repo,
            statuses=statuses,
            events=event,
            branch=pr["head"]["ref"],
            user=pr["user"]["login"],
        )
    else:
        user = None
        branch = branch_or_pull
        if ":" in branch:
            [user, branch] = branch.split(":", 2)
        runs = get_runs_by_branch(
            repo, statuses=statuses, events=event, branch=branch_or_pull, user=user
        )

    runs = sorted(runs, key=lambda x: x["created_at"])
    if not runs:
        print(f"No {' or '.join(statuses)} workflow runs found.\n")
        return

    if not include_last:
        # Only keep one item for each workflow
        seen = set()
        dups = []
        for item in reversed(runs):
            if item["workflow_id"] in seen:
                dups.append(item)
            else:
                seen.add(item["workflow_id"])
        if not dups:
            print(
                "Only the latest runs are in queue. "
                "Use --include-last to force cancelling them.\n"
            )
            return
        runs = dups[::-1]

    last_sha = None

    for entry in runs:
        head_commit = entry["head_commit"]
        if head_commit["id"] != last_sha:
            last_sha = head_commit["id"]
            print_commit(head_commit)
        try:
            print(f"[{entry['status']}] {entry['name']}", end="\r")
            cancel_run(repo, entry["id"])
            print(f"[Cancled] {entry['name']}     ")
        except ClickException as error:
            print(f"[Error: {error.message}] {entry['name']}    ")
    print("")


if __name__ == "__main__":
    # pylint: disable=no-value-for-parameter
    cancel_github_workflows()