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

#gitlab #devops #python #productivity

Initial thoughts
1. Considered alternate solutions
2. The Python script
- Pre-requisites
- Source code
Further reading

Initial thoughts

As a GitLab user, you may be handling multiple projects at once, triggering pipelines. Wouldn't it be great to see all your pipelines at a glance instead of clicking through 15 project pages? Spoiler: GitLab doesn't offer that out of the box.

That's why we've developed a Python script that uses the power of GitLab API to display the latest pipeline runs for every projects in a group. As simple as :

python display-latest-pipelines.py --group-id 12345 --watch

1. Considered alternate solutions

Some alternate solutions have been explored before making a script from scratch.

GitLab's operations dashboard

For premium and ultimate GitLab users, there is the Operations Dashboard:

This is a nice start, but only the overall pipeline status is available, which is too light for pipeline-focused usages.

GitLab CI pipelines exporter

Mentioned in official documentation, the GitLab CI Pipelines Exporter for Prometheus fetches metrics from the API and pipeline events. It can check branches in projects automatically and get the pipeline status and duration. In combination with a Grafana dashboard, this helps build an actionable view for your operations team. Metric graphs can also be embedded into incidents making problem resolving easier. Additionally, it can also export metrics about jobs and environments.

Very interesting, but the scope is way larger than our usage and does not follow pipelines in progress.

Glab ci view

GLab is an open source GitLab CLI tool. It brings GitLab to your terminal: next to where you are already working with Git and your code, without switching between windows and browser tabs.

There is a particular command displaying a pipeline, glab ci view :

Multiple problems for our usage :

Limited to one project
The result is large and does not allow many pipelines on the same screen
You do not get "the latest" pipeline; you have to choose a branch

Tabs grid browser plugins

Tab grids browser plugins are easy to use, but does not update status in real time, you have to refresh the given tabs

2. The Python script

Here is the script, with some consideration :

GitLab host, token, group-id, excluded projects list, stages width and watch mode are configurable
For one shot mode (versus watch mode), projects pipelines are displayed one at a time to allow more real time display
The least possible vertical space is used as a specific goal; you could obtain nicer (but more verbose) output with some minor script adjustment

Pre-requisites

Some Python packages installed
- python -m pip install --upgrade --force-reinstall pip pytz
A token having access to all the projects in the group

If you're interested in similar GitLab API scripts, you might also find useful GitLab: A Python Script Calculating DORA Metrics which uses the same architecture to compute deployment frequency and lead time.

Source code

#
# Display, in console, latest pipelines from each project in a given group
#
# python -m pip install --upgrade --force-reinstall pip pytz
# python display-latest-pipelines.py --group-id=8784450 [--watch] [--token=$GITLAB_TOKEN] [--host=gitlab.com] [--exclude='TPs Benoit C,whatever'] [--stages-width=30]
#
import argparse
from datetime import datetime
from enum import Enum
import os
import pytz
import requests
import sys
import time
import unicodedata

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 = ""

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)')
parser.add_argument('--watch', action='store_true', help='Run indefinitely while refreshing output')
parser.add_argument('--stages-width', type=int, default=42, help='Width for stages display (default: 42)')
args = parser.parse_args()

if args.token is None:
    args.token = os.getenv('GITLAB_TOKEN', 'NONE')

headers = {"Private-Token": args.token}
projects_url = f"https://{args.host}/api/v4/groups/{args.group_id}/projects?include_subgroups=true&simple=true"

def print_or_gather(output, text):
    if args.watch:
        output.append(text)
    else:
        print(text)

def count_emoji(text):
    """Count the number of emojis in the input text."""
    custom_lengths = {
        "\U0001F3D7": 0,  # Construction sign 🏗️
        # Add more special characters as needed.
    }
    count = 0
    for char in text:
        if unicodedata.category(char).startswith('So'):
            if char in custom_lengths:
                count += custom_lengths[char]
            else:
                count += 1
    return count

def get_with_retry(url, headers, backoff_factor=2, parse_json=False):
    """Retry GET request indefinitely with exponential backoff on ChunkedEncodingError, ConnectionError, or JSONDecodeError if parse_json=True."""
    attempt = 0
    while True:
        try:
            response = requests.get(url, headers=headers)
            if parse_json:
                return response.json()
            else:
                return response
        except (requests.exceptions.ChunkedEncodingError, requests.exceptions.ConnectionError, requests.exceptions.JSONDecodeError) as e:
            attempt += 1
            wait_time = backoff_factor ** min(attempt - 1, 10)  # Cap exponent to avoid too long waits
            print(f"Request failed (attempt {attempt}): {e}. Retrying in {wait_time} seconds...")
            time.sleep(wait_time)

def fetch_pipelines():
    response = get_with_retry(projects_url, headers)
    if response.status_code != 200:
        print(f"\n{Color.RED.value}Failed to call GitLab instance: {response.json()}{Color.RESET.value}")
        return
    projects = response.json()

    pipeline_data = {}
    no_pipelines_projects = []
    excluded_projects = set(args.exclude.split(','))

    output = []

    for project in projects:
        if project["name"] in excluded_projects:
            continue
        pipeline_url = f"https://{args.host}/api/v4/projects/{project['id']}/pipelines?per_page=1&sort=desc&order_by=id"
        pipelines = get_with_retry(pipeline_url, headers, parse_json=True)
        if not pipelines:
            no_pipelines_projects.append(project['name'])
            continue
        pipeline = pipelines[0]

        updated_time = datetime.strptime(pipeline["updated_at"], "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=pytz.utc).astimezone(pytz.timezone('Europe/Paris'))

        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: # 2 hours in seconds
            updated_ago = f'{int(delta / 60)} minutes'
        elif delta < 172800: # 2 days in seconds
            updated_ago = f'{int(delta / 3600)} hours'
        else:
            updated_ago = f'{int(delta / 86400)} days'
        match pipeline["status"]:
            case "success":
                color = Color.GREEN
                emoji = "✅"
            case "created" | "waiting_for_resource" | "preparing" | "pending" | "manual":
                color = Color.GREY
                emoji = "⏸️"
            case "canceling" | "canceled" | "skipped":
                color = Color.GREY
                emoji = "✖️"
            case "running":
                color = Color.BLUE
                emoji = "⚙️"
            case "failed":
                color = Color.RED
                emoji = "❌"
            case _:
                print(f"Unknown pipeline status: {pipeline['status']}")
                sys.exit(1)
        print_or_gather(output,f"\n↓{emoji} {color.value}{project['name']} for {pipeline['ref']} : {pipeline['status']} (since {updated_at_human_readable}, {updated_ago} ago){Color.RESET.value}")
        job_data = {}
        jobs_url = f"https://{args.host}/api/v4/projects/{project['id']}/pipelines/{pipeline['id']}/jobs"
        jobs = get_with_retry(jobs_url, headers, parse_json=True)

        for job in list(reversed(jobs)):
            job_name = job["name"]
            stage = job["stage"]
            job_status = job["status"]

            match (job_status, pipeline["status"]):
                case ("success", _):
                    emoji = "🟢"
                    job_color = Color.GREEN
                case ("running", _):
                    emoji = "🔵"
                    job_color = Color.BLUE
                case ("pending" | "created", _):
                    emoji = "🔘"
                    job_color = Color.NO_CHANGE
                case ("skipped" | "canceled", _):
                    emoji = "🔘"
                    job_color = Color.GREY
                case ("warning", _):
                    emoji = "🟠"
                    job_color = Color.YELLOW
                case ("manual", _):
                    emoji = "▶️"
                    job_color = Color.NO_CHANGE
                case ("failed", "success"):
                    emoji = "🟠"
                    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 any stage for this pipeline
        max_jobs = max((len(jobs) for jobs in job_data.values()), default=0)

        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
        for stage, jobs in job_data.items():
            stage = "[ "+stage+" ]"
            lines[0] = f"{lines[0]}╔{stage.center(args.stages_width - 3 - count_emoji(stage), '═').upper()}╗ "
            for i, (job_name, job_status, job_color, emoji) in enumerate(jobs, start=1):
                # emojis in job names make this exercise a bit more difficult: ljust make them expand the size, so we compensate
                lines[i] = f"{lines[i]}{emoji} {job_color.value}{job_name[:args.stages_width - 6 - count_emoji(job_name)].ljust(args.stages_width - 3 - count_emoji(job_name))}{Color.RESET.value}"
            for j in range(len(jobs) + 1, max_jobs + 1):
                lines[j] = lines[j] + " ".ljust(args.stages_width)

        for line in lines:
            print_or_gather(output,line)

    if no_pipelines_projects:
        print_or_gather(output,f"\n\033[90mProjects without pipeline: {', '.join(no_pipelines_projects)}\033[0m")

    return "\n".join(output)

try:
    if args.watch:
            while True:
                output = fetch_pipelines()
                sys.stdout.write("\x1b[2J\x1b[H")  # Clear the screen
                sys.stdout.flush()
                print(output)
    else:
        fetch_pipelines()
except KeyboardInterrupt:
    pass