Cleanup Gitlab artifacts in 1 repository

cleanup_artifacts.py

import requests
import os
import sys

# --- Configuration ---
# Set your token as an environment variable: export GITLAB_TOKEN="glpat-..."
GITLAB_DOMAIN = os.environ.get("GITLAB_DOMAIN", "gitlab.com")
GITLAB_TOKEN = os.environ.get("GITLAB_TOKEN")
GITLAB_GRAPHQL_URL = f"https://{GITLAB_DOMAIN}/api/graphql"
PROJECT_PATH = os.environ.get("PROJECT_PATH")  # e.g., "mygroup/myproject"

if not PROJECT_PATH:
    print("Error: PROJECT_PATH environment variable is not set.")
    sys.exit(1)

# Safety toggle: Set to False to actually delete the artifacts
DRY_RUN = os.environ.get("DRY_RUN", "True").lower() in ("true", "1", "yes")
BATCH_SIZE = int(os.environ.get("BATCH_SIZE", 20)) # The mutation array size

if not GITLAB_TOKEN:
    print("Error: GITLAB_TOKEN environment variable is not set.")
    sys.exit(1)

HEADERS = {
    "Authorization": f"Bearer {GITLAB_TOKEN}",
    "Content-Type": "application/json"
}

# --- GraphQL Queries ---
QUERY_GET_ARTIFACTS = """
query getJobArtifacts($projectPath: ID!, $firstPageSize: Int, $nextPageCursor: String = "") {
  project(fullPath: $projectPath) {
    id
    jobs(withArtifacts: true, first: $firstPageSize, after: $nextPageCursor) {
      nodes {
        id
        name
        artifacts {
          nodes {
            id
            name
            size
          }
        }
      }
      pageInfo {
        hasNextPage
        endCursor
      }
    }
  }
}
"""

MUTATION_DELETE_ARTIFACTS = """
mutation bulkDestroyJobArtifacts($projectId: ProjectID!, $ids: [CiJobArtifactID!]!) {
  bulkDestroyJobArtifacts(input: {projectId: $projectId, ids: $ids}) {
    destroyedCount
    destroyedIds
    errors
  }
}
"""

def run_graphql(query, variables):
    """Executes a GraphQL query/mutation and returns the JSON response."""
    payload = {"query": query, "variables": variables}
    response = requests.post(GITLAB_GRAPHQL_URL, json=payload, headers=HEADERS)
    response.raise_for_status()
    result = response.json()
    if "errors" in result:
        print(f"GraphQL Error: {result['errors']}")
        sys.exit(1)
    return result

def main():
    print(f"Fetching artifacts for project: {PROJECT_PATH}...")
    
    has_next_page = True
    cursor = ""
    project_id = None
    all_artifact_ids = []
    total_size_bytes = 0

    # 1. Paginate and Collect Artifacts
    while has_next_page:
        variables = {
            "projectPath": PROJECT_PATH,
            "firstPageSize": 100,
            "nextPageCursor": cursor
        }
        
        data = run_graphql(QUERY_GET_ARTIFACTS, variables)["data"]["project"]
        
        if not data:
            print("Project not found. Check your permissions and path.")
            sys.exit(1)

        project_id = data["id"]
        jobs = data["jobs"]["nodes"]
        page_info = data["jobs"]["pageInfo"]

        for job in jobs:
            for artifact in job["artifacts"]["nodes"]:
                all_artifact_ids.append(artifact["id"])
                # Handle cases where size might be None
                if artifact.get("size"):
                    total_size_bytes += int(artifact["size"])

        has_next_page = page_info["hasNextPage"]
        cursor = page_info["endCursor"]

    total_size_mb = total_size_bytes / (1024 * 1024)
    print(f"Found {len(all_artifact_ids)} artifacts (Total Size: ~{total_size_mb:.2f} MB)")

    if not all_artifact_ids:
        print("No artifacts found to clean up.")
        return

    if DRY_RUN:
        print("\n[DRY RUN] The following actions would be taken:")
        print(f"- Target Project ID: {project_id}")
        print(f"- Deleting {len(all_artifact_ids)} artifacts in batches of {BATCH_SIZE}.")
        print("Set DRY_RUN = False in the script to execute deletion.")
        return

    # 2. Batch Delete Artifacts
    print("\nStarting deletion process...")
    for i in range(0, len(all_artifact_ids), BATCH_SIZE):
        batch = all_artifact_ids[i:i + BATCH_SIZE]
        
        variables = {
            "projectId": project_id,
            "ids": batch
        }
        
        result = run_graphql(MUTATION_DELETE_ARTIFACTS, variables)
        mutation_response = result["data"]["bulkDestroyJobArtifacts"]
        
        if mutation_response["errors"]:
            print(f"Error deleting batch {i//BATCH_SIZE + 1}: {mutation_response['errors']}")
        else:
            print(f"Successfully destroyed {mutation_response['destroyedCount']} artifacts. (Batch {i//BATCH_SIZE + 1})")

if __name__ == "__main__":
    main()

requirements.txt

requests