Skip to content

Instantly share code, notes, and snippets.

@ktmud
Last active January 9, 2023 19:59
Show Gist options
  • Save ktmud/a0a9673254524d01189a8a7000829357 to your computer and use it in GitHub Desktop.
Save ktmud/a0a9673254524d01189a8a7000829357 to your computer and use it in GitHub Desktop.
Manually cancel previous GitHub Action workflow runs in queue
#!/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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment