🦊 GitLab: A Python Script Displaying Latest Pipelines in a Group's Projects
Tags: gitlab, devops, python, productivity
Published: 2024-06-29 12:33:54

---
title: "\U0001F98A GitLab: A Python Script Displaying Latest Pipelines in a Group's Projects"
tags:
  - gitlab
  - devops
  - python
  - productivity
license: public-domain
cover_image: 'https://raw.githubusercontent.com/bcouetil/articles/main/images/gitlab/cicd/console-group-pipelineV2.jpg'
published: true
id: 1877995
date: '2024-06-29T12:33:54Z'
---

<!--
TODO
420 x 1000 =>
-->

- [Initial thoughts](#initial-thoughts)
- [1. Considered alternate solutions](#1-considered-alternate-solutions)
  - [GitLab's operations dashboard](#gitlabs-operations-dashboard)
  - [GitLab CI pipelines exporter](#gitlab-ci-pipelines-exporter)
  - [Glab ci view](#glab-ci-view)
  - [Tabs grid browser plugins](#tabs-grid-browser-plugins)
- [2. The Python script](#2-the-python-script)
  - [Pre-requisites](#pre-requisites)
  - [Source code](#source-code)
- [Further reading](#further-reading)

# Initial thoughts

As a GitLab user, you may be handling multiple projects at once, triggering pipelines. Wouldn't it be great if there was an easy way to monitor all these pipelines in real-time? Unfortunately, out-of-the-box solutions don't quite fit the bill.

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

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

![python-output](https://raw.githubusercontent.com/bcouetil/articles/main/images/gitlab/cicd/console-group-pipelines.gif)

# 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](https://docs.gitlab.com/ee/user/operations_dashboard/):

![operation-dashboard](https://raw.githubusercontent.com/bcouetil/articles/main/images/gitlab/cicd/operations-dashboard.png)

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](https://docs.gitlab.com/ee/ci/pipelines/pipeline_efficiency.html#pipeline-monitoring), the [GitLab CI Pipelines Exporter](https://github.com/mvisonneau/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. 

![prometheus-exporter](https://raw.githubusercontent.com/mvisonneau/gitlab-ci-pipelines-exporter/main/docs/images/grafana_dashboard_pipelines.jpg)

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

## Glab ci view

[GLab](https://docs.gitlab.com/ee/editor_extensions/gitlab_cli/) 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](https://gitlab.com/gitlab-org/cli/-/blob/main/docs/source/ci/view.md), `glab ci view` :

![glab-ci-view](https://raw.githubusercontent.com/bcouetil/articles/main/images/gitlab/cicd/console-ci-glab.png)

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
  - `pip install pytz`
- A token having access to all the projects in the group

## Source code

```python
#
# Display, in console, latest pipelines from each project in a given group
#
# 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 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 fetch_pipelines():
    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}")
        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://gitlab.com/api/v4/projects/{project['id']}/pipelines?per_page=1&sort=desc&order_by=id"
        response = requests.get(pipeline_url, headers=headers)
        if not response.json():
            no_pipelines_projects.append(project['name'])
            continue
        pipeline = response.json()[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
            case "created" | "waiting_for_resource" | "preparing" | "pending" | "canceled" | "skipped" | "manual":
                color = Color.GREY
            case "running":
                color = Color.BLUE
            case "failed":
                color = Color.RED
        print_or_gather(output,f"\n↓ {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://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"]
            
            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())

        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
```

![a humanoid fox from behind watching metrics dashboards, multiple computer monitors,manga style](https://raw.githubusercontent.com/bcouetil/articles/main/images/gitlab/dora/fox-dashboard-2.jpg)

_Illustrations generated locally by Pinokio using Stable Cascade plugin_

# Further reading

{% embed https://dev.to/zenika/gitlab-a-python-script-calculating-dora-metrics-258o %}

{% embed https://dev.to/zenika/gitlab-ci-the-majestic-single-server-runner-1b5b %}

{% embed https://dev.to/zenika/gitlab-ci-yaml-modifications-tackling-the-feedback-loop-problem-4ib1 %}

{% embed https://dev.to/zenika/gitlab-ci-optimization-15-tips-for-faster-pipelines-55al %}

{% embed https://dev.to/zenika/gitlab-ci-10-best-practices-to-avoid-widespread-anti-patterns-2mb5 %}

{% embed https://dev.to/zenika/gitlab-pages-preview-the-no-compromise-hack-to-serve-per-branch-pages-5599 %}

{% embed https://dev.to/zenika/chatgpt-if-you-please-make-me-a-gitlab-jobs-attributes-sorter-3co3 %}

{% embed https://dev.to/zenika/gitlab-runners-topologies-pros-and-cons-2pb1 %}

_This article was enhanced with the assistance of an AI language model to ensure clarity and accuracy in the content, as English is not my native language._