gitlab-ci-and-mr.xbar.1m.py

#!/usr/local/bin/python

import json
import datetime
from urllib.request import urlopen


INSTANCES = [
    {
        # Your private key for accessing gitlab: User -> Settings -> Access tokens -> add personal access token with api scope
        'privateToken': 'glpat-xxxxxxxxxxxxxxxxxxxx',
        # Gitlab URL
        'url': 'https://gitlab.xxxxxxxxxx.com',
        # Define your server and projects (name: id)
        # To get id go to project -> Settings -> General -> General project settings
        'projects': {
            "xxxx/xx": 123,
        },
        'my_username': 'xxxxxxxxxxxx'
    },
]

pipelines = []

# Converts the gitlab status to emoji
def stateIcon(status):
    return {
        "created": "🌟",
        "pending": "💤",
        "running": "️♻️",
        "failed": "💥",
        "success": "✔️",
        "skipped": "🚧",
        "manual": "👋",
        "canceled": "🛑",
    }[status]

# Calls gitlab API endpoint with private_token
def api (instance, method):
    url = instance['url'] + "/api/v4/" + method
    param = 'private_token=' + instance['privateToken']
    # Detect if method has query string (we need to append private token)
    url = url + ('&' if "?" in url else '?') + param
    body = urlopen(url).read()
    return json.loads(body.decode('utf-8'))

# Project details
class Project:
    def __init__ (self, name, id):
        self.name = name
        self.id = id

# Pipeline job
class Job:
    def __init__ (self, json):
        self.name = json["stage"] + (": " + json["name"] if json["name"] != json["stage"] else "" )
        self.status = json["status"]
        self.duration = 0 if json["duration"] is None or self.status == 'running' else int(json["duration"])
        self.commit = json['commit']['title']
        self.commit_user_name = json['user']['username']

    # Jobs name with duration
    def displayName(self):
        return self.name + (' ' + str(self.duration) + 's' if self.duration > 0 else '')
    
class Pipeline:
    def __init__ (self, projectName, projectId, json, instance):
        self.project = Project(projectName, projectId)
        self.instance = instance
        self.id = json["id"]
        self.url = json['web_url']
        self.jobs = []
        self.runningJobs = []
        self.ref = str(json["ref"])
        self.commit = None
        self.commit_user_name = None

    # Display name with current running jobs
    def displayName(self):
        jobsString = '💤'

        # Get running jobs and append the name
        if len(self.runningJobs) > 0:
            strings = []
            for job in self.runningJobs:
                strings.append(job.displayName()) 

            jobsString = ', '.join(strings)

        return self.project.name + ' - ' + self.ref + ' (' + jobsString + ')'

    # Add jobs array json
    def addJobs(self, jobsArray):
        for jobJson in jobsArray:
            # Parse the job
            job = Job(jobJson)
            # Add the jobs array
            self.jobs.append(job)

            # Get the commit from the first job
            if self.commit is None:
                self.commit = job.commit
                self.commit_user_name = job.commit_user_name

            # Check if the job is running for running jobs array
            if job.status == 'running':
                self.runningJobs.append(job)

    @property
    def isMyPipeline(self):
        return self.commit_user_name == self.instance['my_username']


# Loop the projects and get thy jobs
daysAgo = (datetime.datetime.now()-datetime.timedelta(days=3)).strftime('%Y-%m-%dT%H:%M:%S.%fZ')
mrList = []
for instance in INSTANCES:
    for name, project in instance['projects'].items():
        refs = set()
        pipelinesResp = api(instance, "projects/"+str(project)+"/pipelines?username="+instance['my_username']+'&updated_after='+daysAgo)
        for pipelineJson in pipelinesResp:
            if pipelineJson['status'] not in ['running', 'failed', 'manual']:
                continue
            pipeline = Pipeline(name, project, pipelineJson, instance)

            # only latest pipeline for each ref
            if pipeline.ref in refs:
                continue
            refs.add(pipeline.ref)

            jobsArray = api(instance, "projects/"+str(project)+"/pipelines/"+str(pipeline.id)+"/jobs")
            if len(jobsArray) > 0:
                pipeline.addJobs(jobsArray)
                pipelines.append(pipeline)

    ## merge requests
    # https://docs.gitlab.com/ee/api/merge_requests.html#merge-requests-api
    state = 'opened'
    resp = api(instance, f"merge_requests?author_username={instance['my_username']}&state={state}")
    for mrJson in resp:
        color = '#E24329' if mrJson['has_conflicts'] else 'black'
        mr = f'{mrJson["references"]["full"]} {mrJson["title"]} | href={mrJson["web_url"]} color={color}'
        mrList.append(mr)

pipelineCount = len(pipelines)
if pipelineCount == 0:
    print("💤")
    exit


## Render the pipelines names (bitbar will loop)
for index, pipeline in enumerate(pipelines):
    print('🚀 ', end=' ')

    if pipelineCount > 1:
        print(str(index + 1) + '/' + str(pipelineCount) + ' ', end=' ')

    print(pipeline.displayName())


## Start menu
print("---")

for i, pipeline in enumerate(pipelines, 1):
    print(f'🚀 {i}/{len(pipelines)} ' + pipeline.project.name + ' - ' + pipeline.ref + ' - ' + pipeline.commit_user_name + '| color=black' + ' | href=' + pipeline.url)
    print('-- commit: ' + pipeline.commit + '| color=black')
    print('---')
    for job in pipeline.jobs:
        if job.status in ['created', 'skipped', 'canceled']:
            continue
        print(stateIcon(job.status) + " ", end=' ')

        style = ''
        if job.status == 'success':
            style = '| color=green'
        elif job.status == 'running':
            style = '| color=blue'

        print(job.displayName() + style)
        
print('---')
print('📝 Merge requests | color=black')
for mr in mrList:
    print(mr)