🦊 GitLab: A Python Script Displaying Latest Pipelines in Group’s Projects

watch --no-title --color python display-latest-pipelines.py --group-id 12345

#
### Display, in console, latest pipelines from each project in a given group
#
## one shot:
# python display-latest-pipelines.py --exclude='TPs Benoit C,whatever' --group-id=87844506
#
## continuous update:
# watch --no-title --color "python display-latest-pipelines.py --exclude='TPs Benoit C,whatever' --group-id=87844506"
#
import argparse
import requests
from datetime import datetime
import pytz
import os
from enum import Enum class Color(Enum): GREEN = "\033[92m" GREY = "\033[90m" CYAN = "\033[96m" RED = "\033[91m" YELLOW = "\033[93m" BLUE = "\033[94m" RESET = "\033[0m" NO_CHANGE = "" # Parse command-line arguments for access token, group ID, and GitLab hostname
parser = argparse.ArgumentParser(description='Retrieve GitLab pipeline data for projects in a group')
parser.add_argument('--host', type=str, default="gitlab.com", help='Hostname of the GitLab instance')
parser.add_argument('--token', type=str, default=None, help='GitLab API access token (default: $GITLAB_TOKEN (exported) environment variable)')
parser.add_argument('--group-id', type=int, help='ID of the group to retrieve projects from')
parser.add_argument('--exclude', type=str, default="", help='Comma-separated list of project names to exclude (default: none)')
args = parser.parse_args() # Set maximum length for job names
stages_width = 36
# Check if token is set via environment variable and use it as default if not provided
if args.token is None: args.token = os.getenv('GITLAB_TOKEN', 'NONE') # Set headers and URLs for API requests
headers = {"Private-Token": args.token}
projects_url = f"https://{args.host}/api/v4/groups/{args.group_id}/projects?include_subgroups=true&simple=true" response = requests.get(projects_url, headers=headers)
if response.status_code != 200: print(f"\n{Color.RED.value}Failed to call GitLab instance: {response.json()}{Color.RESET.value}") exit()
projects = response.json() pipeline_data = {}
no_pipelines_projects = []
excluded_projects = set(args.exclude.split(',')) for project in projects: if project["name"] in excluded_projects: continue # Retrieve the most recent pipeline for the current project
 pipeline_url = f"https://gitlab.com/api/v4/projects/{project['id']}/pipelines?per_page=1&sort=desc&order_by=updated_at" response = requests.get(pipeline_url, headers=headers) if not response.json(): # check if there are any pipelines for this project
 no_pipelines_projects.append(project['name']) continue pipeline = response.json()[0] # Convert updated_at to a datetime object and convert it to Paris timezone
 updated_time = datetime.strptime(pipeline["updated_at"], "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=pytz.utc).astimezone(pytz.timezone('Europe/Paris')) # Format the datetime object as a human-readable string
 updated_at_human_readable = updated_time.strftime("%d %b %Y at %H:%M:%S") time_diff = datetime.now(pytz.utc) - updated_time delta = time_diff.total_seconds() if delta < 120: updated_ago = f'{int(delta)} seconds' elif delta < 7200: updated_ago = f'{int(delta / 60)} minutes' elif delta < 172800: # 2 days in seconds (60*60*24*2)
 updated_ago = f'{int(delta / 3600)} hours' else: updated_ago = f'{int(delta / 86400)} days' # 1 day in seconds (60*60*24)
 # Determine color based on pipeline status
 match pipeline["status"]: case "success": color = Color.GREEN case "created" | "waiting_for_resource" | "preparing" | "pending" | "canceled" | "skipped" | "manual": color = Color.GREY case "running": color = Color.CYAN case "failed": color = Color.RED print(f"\n↓ {color.value}{project['name']} for {pipeline['ref']} : {pipeline['status']} (since {updated_at_human_readable}, {updated_ago} ago){Color.RESET.value}") job_data = {} # Retrieve data about all jobs for the current pipeline
 jobs_url = f"https://gitlab.com/api/v4/projects/{project['id']}/pipelines/{pipeline['id']}/jobs" response = requests.get(jobs_url, headers=headers) jobs = response.json() for job in list(reversed(jobs)): job_name = job["name"] stage = job["stage"] job_status = job["status"] # Determine emoji based on job status and pipeline status
 match (job_status, pipeline["status"]): case ("success", _): emoji = "🍀" # 🟢✅ do not work within a "watch" on macos
 job_color = Color.GREEN case ("running", _): emoji = "🔵" job_color = Color.CYAN case ("pending" | "created", _): emoji = "🔘" job_color = Color.NO_CHANGE case ("skipped", _): emoji = "🔘" job_color = Color.GREY case ("warning", _): emoji = "🍊" # 🟠 does not work within a "watch" on macos
 job_color = Color.YELLOW case ("manual", _): emoji = "▶️ " job_color = Color.NO_CHANGE case ("failed", "success"): emoji = "🍊" # 🟠 does not work within a "watch" on macos
 job_color = Color.YELLOW case ("failed", _): emoji = "🔴" job_color = Color.RED case (_, _): print(job_status) if stage not in job_data: job_data[stage] = [] job_data[stage].append((job_name, job_status, job_color, emoji)) # Sort jobs within each stage alphabetically by job name
 for stage in job_data: job_data[stage].sort(key=lambda x: x[0]) # Find the maximum number of jobs in a stage for this pipeline
 max_jobs = max(len(jobs) for jobs in job_data.values()) lines = [" "] * (max_jobs + 1) lines[0] = "" # stages start with a border character instead of a space
 # Print out the job data for each stage, padding to make all stages content the same length (emojis make this exercise a bit more difficult)
 for stage, jobs in job_data.items(): stage = "[ "+stage+" ]" lines[0] = f"{lines[0]}╔{stage.center(stages_width-3, '═').upper()}╗ " for i, (job_name, job_status, job_color, emoji) in enumerate(jobs, start=1): lines[i] = f"{lines[i]}{emoji} {job_color.value}{job_name[:stages_width-5].ljust(stages_width-3)}{Color.RESET.value}" for j in range(len(jobs) + 1, max_jobs + 1): lines[j] = lines[j] + " ".ljust(stages_width) for line in lines: print(line) if no_pipelines_projects: print(f"\n\033[90mProjects without pipeline:", ', '.join(no_pipelines_projects))