Script to rerun failed github action workflows belonging to a PR with a specific title

cancel_workflows.py

#!/usr/bin/env python3
import subprocess
import json
import argparse
from typing import Iterable

try:
    from rich_argparse import RichHelpFormatter
except ImportError:
    RichHelpFormatter = argparse.HelpFormatter

parser = argparse.ArgumentParser(
    description='Cancel all workflows related to a PR.',
    formatter_class=RichHelpFormatter,
)
parser.add_argument(
    '--max_workflows',
    dest='max_workflows',
    action='store',
    help='The maximum number of workflows to look at (starting from the latest)',
    default=200,
)
parser.add_argument(
    '--title',
    dest='title',
    action='store',
    help='The title of the PR for which we want to rerun workflows (or part of the title) - or "master" for all pushes',
)
parser.add_argument('--not', dest='not_', action='append')
parser.add_argument('--repo')


def join(things: Iterable[str]):
    return f'{{{", ".join(set(things))}}}'


def main():
    args = parser.parse_args()

    nlimit = args.max_workflows

    repo = [
        '--repo',
        args.repo or (call('api', '/user')['login'] + '/duckdb'),
    ]

    results = call(
        'run',
        'list',
        '--json',
        'displayTitle,databaseId,status,conclusion,headSha,event,workflowName',
        f'--limit={nlimit}',
        *repo,
    )
    results = [run for run in results if run['status'] in ('queued', 'in_progress')]

    if args.title:
        results = [
            run for run in results if args.title.lower() in run['displayTitle'].lower()
        ]
        print('selected by title:', join(run['displayTitle'] for run in results))

    if args.not_:
        results = [
            run
            for run in results
            if not any(
                negate.lower() in run['workflowName'].lower() for negate in args.not_
            )
        ]
        print('selected by negation:', join(run["workflowName"] for run in results))

    result = results

    if not result:
        parser.error(
            f"No workflows found in the latest {nlimit} workflows that match.\nPerhaps try running with a higher "
            f"--max_workflows parameter?"
        )

    if input(f'Will kill {len(result)} workflows, continue? ') == 'yes':
        for run in result:
            subprocess.run(['gh', 'run', 'cancel', str(run['databaseId']), *repo])
    else:
        print('Aborting.')


def call(*params):
    return json.loads(subprocess.check_output(['gh'] + list(params)))


if __name__ == '__main__':
    main()

rerun_workflows.py

#!/usr/bin/env python3
import subprocess
import os
import pandas as pd
import argparse

from cancel_workflows import call

parser = argparse.ArgumentParser(description='Rerun failed workflows from a PR.')
parser.add_argument(
    '--max_workflows',
    dest='max_workflows',
    action='store',
    help='The maximum number of workflows to look at (starting from the latest)',
    default=200,
)
parser.add_argument(
    '--title',
    dest='title',
    action='store',
    help='The title of the PR for which we want to rerun workflows (or part of the title)',
    required=True,
)
parser.add_argument(
    '--rerun_cancelled',
    dest='rerun_cancelled',
    action='store',
    help='The title of the PR for which we want to rerun workflows (or part of the title)',
    default=False,
)
args = parser.parse_args()

nlimit = args.max_workflows
query = args.title


df = call(
    [
        'run',
        'list',
        '--json',
        'displayTitle,databaseId,status,conclusion,headSha',
        f'--limit={nlimit}',
    ]
)
result = [run['headSha'] for run in df if query.lower() in run['displayTitle'].lower()]
if not result:
    parser.error(
        f"No workflows found in the latest {nlimit} workflows that contain the text {query}.\nPerhaps try running with a higher --max_workflows parameter?"
    )

headSha = result[0][0]

result = duckdb.query(
    f"select databaseId from df where conclusion IN ('failure', 'cancelled') AND displayTitle LIKE '%{query}%' and headSha='{headSha}'"
).fetchall()
if not result:
    print(
        f"Found runs that match the text {query} but no failing or cancelled runs were found"
    )
for databaseId, *_ in result:
    os.system(f'gh run rerun {databaseId}')