|
# WARNING: this script removes ALL deployments from a given repo |
|
# See the associated README.md for more information |
|
# usage: python3 nuke_gh_deployments.py lreading/slide-spec |
|
import argparse |
|
import os |
|
import sys |
|
import time |
|
|
|
import requests |
|
from dotenv import load_dotenv |
|
|
|
GH_PAT_ENV_KEY = "GITHUB_PAT" |
|
GH_API_BASE_URL = "https://api.github.com" |
|
|
|
|
|
def get_headers(): |
|
return { |
|
"Authorization": f"Bearer {os.getenv(GH_PAT_ENV_KEY)}", |
|
"Accept": "application/vnd.github+json", |
|
} |
|
|
|
|
|
def get_all_deployments(repo): |
|
deployments = [] |
|
page = 1 |
|
|
|
while True: |
|
url = f"{GH_API_BASE_URL}/repos/{repo}/deployments" |
|
params = {"per_page": 100, "page": page} |
|
|
|
r = requests.get(url, headers=get_headers(), params=params) |
|
if r.status_code != 200: |
|
print("Failed to list deployments:", r.text) |
|
sys.exit(1) |
|
|
|
batch = r.json() |
|
if not batch: |
|
break |
|
|
|
deployments.extend(batch) |
|
page += 1 |
|
|
|
return deployments |
|
|
|
|
|
def delete_statuses(deployment_id, repo): |
|
page = 1 |
|
|
|
while True: |
|
url = f"{GH_API_BASE_URL}/repos/{repo}/deployments/{deployment_id}/statuses" |
|
params = {"per_page": 100, "page": page} |
|
|
|
r = requests.get(url, headers=get_headers(), params=params) |
|
if r.status_code != 200: |
|
print(f"Failed to list statuses for {deployment_id}:", r.text) |
|
sys.exit(1) |
|
|
|
statuses = r.json() |
|
if not statuses: |
|
break |
|
|
|
for status in statuses: |
|
status_id = status["id"] |
|
del_url = f"{GH_API_BASE_URL}/repos/{repo}/deployments/{deployment_id}/statuses/{status_id}" |
|
dr = requests.delete(del_url, headers=get_headers()) |
|
if dr.status_code not in (204, 404): |
|
print(f"Failed deleting status {status_id}:", dr.text) |
|
sys.exit(1) |
|
|
|
page += 1 |
|
|
|
|
|
def mark_inactive_if_needed(deployment_id, repo): |
|
""" |
|
GitHub will not allow deleting an active deployment |
|
if multiple deployments exist. |
|
We force it inactive. |
|
""" |
|
url = f"{GH_API_BASE_URL}/repos/{repo}/deployments/{deployment_id}/statuses" |
|
|
|
payload = {"state": "inactive"} |
|
|
|
r = requests.post(url, headers=get_headers(), json=payload) |
|
if r.status_code not in (201, 200): |
|
print(f"Warning: Could not mark deployment {deployment_id} inactive:", r.text) |
|
|
|
|
|
def delete_deployment(deployment_id, repo): |
|
url = f"{GH_API_BASE_URL}/repos/{repo}/deployments/{deployment_id}" |
|
|
|
r = requests.delete(url, headers=get_headers()) |
|
if r.status_code == 204: |
|
print(f"Deleted deployment {deployment_id}") |
|
elif r.status_code == 422: |
|
print( |
|
f"Deployment {deployment_id} still active. Marking inactive and retrying..." |
|
) |
|
mark_inactive_if_needed(deployment_id, repo) |
|
time.sleep(1) |
|
r2 = requests.delete(url, headers=get_headers()) |
|
if r2.status_code == 204: |
|
print(f"Deleted deployment {deployment_id}") |
|
else: |
|
print("Failed deleting deployment:", r2.text) |
|
sys.exit(1) |
|
else: |
|
print("Failed deleting deployment:", r.text) |
|
sys.exit(1) |
|
|
|
|
|
def parse_and_validate_args(): |
|
parser = argparse.ArgumentParser( |
|
prog="GitHub Deployment Nuker", |
|
description="Deletes ALL deployments from a given repo", |
|
) |
|
parser.add_argument( |
|
"repo", |
|
help="The owner or org and the repo name, eg: lreading/slide-spec", |
|
) |
|
args = parser.parse_args() |
|
|
|
if len(args.repo) == 0: |
|
parser.print_help() |
|
exit(1) |
|
|
|
if "/" not in args.repo: |
|
parser.print_help() |
|
print() |
|
print(f"Invalid repo: {args.repo}") |
|
print("Repo must be formatted as <owner>/<name>") |
|
exit(1) |
|
|
|
return args |
|
|
|
|
|
def main(): |
|
load_dotenv() |
|
if not os.getenv(GH_PAT_ENV_KEY): |
|
print(f"{GH_PAT_ENV_KEY} is a required environment variable") |
|
print("You can use a .env file if you'd like") |
|
exit(1) |
|
|
|
args = parse_and_validate_args() |
|
|
|
deployments = get_all_deployments(args.repo) |
|
|
|
if not deployments: |
|
print("No deployments found.") |
|
return |
|
|
|
print(f"Found {len(deployments)} deployments.") |
|
|
|
for d in deployments: |
|
deployment_id = d["id"] |
|
print(f"Processing deployment {deployment_id}") |
|
|
|
delete_statuses(deployment_id, args.repo) |
|
delete_deployment(deployment_id, args.repo) |
|
|
|
print("All deployments removed.") |
|
|
|
|
|
if __name__ == "__main__": |
|
main() |