diff --git a/deployment/piplines/clean_pipelines.py b/deployment/piplines/clean_pipelines.py new file mode 100644 index 0000000000000000000000000000000000000000..76854df6b2999567ba738683616d4ec170740ec2 --- /dev/null +++ b/deployment/piplines/clean_pipelines.py @@ -0,0 +1,57 @@ +""" +Simple script to clean up old pipelines from the GitLab project. +""" + +from datetime import UTC, datetime, timedelta + +import requests +from environ import environ +from rich.progress import Progress + +env = environ.FileAwareEnv( + API_TOKEN=(str, None), + THRESHOLD_ALL=(int, 52), + THRESHOLD_FAILED=(int, 26), + CI_API_V4_URL=(str, None), + CI_PROJECT_ID=(int, None), +) + + +TOKEN = env('API_TOKEN') +PER_PAGE = 100 +DELETE_ALL = datetime.now(tz=UTC) - timedelta(weeks=env('THRESHOLD_ALL')) +DELETE_FAILED = datetime.now(tz=UTC) - timedelta(weeks=env('THRESHOLD_FAILED')) +BASE_URL = f'{env('CI_API_V4_URL')}/projects/{env("CI_PROJECT_ID")}' + +if TOKEN is None: + raise ValueError('API_TOKEN is required') + +while True: + response = requests.get( + f'{BASE_URL}/pipelines?per_page={PER_PAGE}&updated_before={DELETE_ALL.strftime("%Y-%m-%dT%H:%M:%SZ")}&sort=asc&order_by=updated_at&page=1', + headers={'PRIVATE-TOKEN': TOKEN}, + ) + pipelines = response.json() + if not pipelines: + break + with Progress() as progress: + task1 = progress.add_task('[red]Deleting...', total=len(pipelines)) + for pipeline in pipelines: + delete_response = requests.delete(f'{BASE_URL}/pipelines/{pipeline["id"]}', headers={'PRIVATE-TOKEN': TOKEN}) + assert delete_response.status_code == 204 + progress.update(task1, advance=1) + +while True: + response = requests.get( + f'{BASE_URL}/pipelines?per_page={PER_PAGE}&updated_before={DELETE_FAILED.strftime("%Y-%m-%dT%H:%M:%SZ")}&status=failed&sort=asc&order_by=updated_at&page=1', + headers={'PRIVATE-TOKEN': TOKEN}, + ) + pipelines = response.json() + if not pipelines: + break + with Progress() as progress: + task1 = progress.add_task('[red]Deleting...', total=len(pipelines)) + for pipeline in pipelines: + delete_response = requests.delete(f'{BASE_URL}/pipelines/{pipeline["id"]}', headers={'PRIVATE-TOKEN': TOKEN}) + assert delete_response.status_code == 204 + progress.update(task1, advance=1)