GITLABにアクセスしてPipelineの一覧を取得、操作するアプリを作成

方針

・環境変数を設定
・pythonプログラムを書く

環境変数を設定

環境変数の設定ファイルはこちら、以前のGITLABの設定より

.envrc

export  GITLAB_USERNAME="root"
export  GITLAB_PASSWORD="bKORzwipG166LWNQmj5uQMHxfVR18Q7fVDwxP9F8Ha4="

pythonプログラムを書く

gitlab_api.py

import os
import requests
import argparse
from urllib3 import disable_warnings
from prettytable import PrettyTable

disable_warnings()

GITLAB_URL = "https://localhost"
HEADERS = {"Content-Type":"application/json","Accept":"application/json"}

class GitLab_Auth():
    def __init__(self):
        payload = {
            "username":os.environ["GITLAB_USERNAME"],
            "password":os.environ["GITLAB_PASSWORD"],
            "grant_type":"password"
        }
        
        auth_response = requests.post(
            url=f"{GITLAB_URL}/oauth/token",json=payload,headers=HEADERS,verify=False
        )
        
        auth_response.raise_for_status()
        self.token = auth_response.json()["access_token"]
        
def gitlab(ctx):
    if "GITLAB_USERNAME" not in os.environ or "GITLAB_PASSWORD" not in os.environ:
        print("ERROR:You must specify GITLAB_USERNAME and GITLAB_PASSWORD as environment variables")
        exit(1)
    ctx.obj = GitLab_Auth()
    
def projects(gitlab_obj,all):
    table = PrettyTable()
    headers = HEADERS.copy()
    headers["Authorization"] = f"Bearer {gitlab_obj.token}"
    
    try:
        project_list = requests.get(
            url = f"{GITLAB_URL}/api/v4/projects?owned=True",headers=headers, verify=False
        )
    except Exception as e:
        print(f"ERROR: Unable to get Gitlab Project: {e}")
        exit(1)
        
    project_list_json = project_list.json()
    table.field_names = ["ID","Name","Group","URL","Pipeline Status"]
    table.align = "l"
    
    for proj in project_list_json:
        add_row = True
        row = [
            proj["id"],
            proj["name"],
            proj["name_with_namespace".split("/")[0].strip()],
            proj["web_url"]
        ]
        
        try:
            pipeline_status_check = requests.get(
                url = f"{GITLAB_URL}/api/v4/projects/{proj['id']}/pipelines",headers=headers,verify=False
            )
            pipeline_status_check.raise_for_status()
        except Exception as e:
            print(f"WARNING: Unable to get pipeline list for project {proj['id']}: {e}")
            continue
        
        pipeline_list = pipeline_status_check.json()
        
        if len(pipeline_list) > 0 :
            row.append(pipeline_list[0]["status"])
        else:
            row.append("N/A")
            
        if not all:
            if len(pipeline_list) == 0:
                add_row = False
        
        if add_row:
            table.add_row(row)
            
    print(table)

def list_pipelines(gitlab_obj,all,project_id):
    table = PrettyTable()
    headers = HEADERS.copy()
    headers["Authorization"] = f"Bearer {gitlab_obj.token}"
    
    try:
        pipeline_list = requests.get(
            url = f"{GITLAB_URL}/api/v4/projects/{project_id}/pipelines",
            headers= headers, verify=False
        )
    except Exception as e:
        print(f"ERROR: Failed to load pipelines for {project_id}: {e}")
        exit(1)
    
    table.field_names = ["ID","Branch","Started At","Finished At","URL","Status"]
    table.align = "l"
    
    pipeline_list_json = pipeline_list.json()
    
    for pipeline in pipeline_list_json:
        try:
            response = requests.get(
                url = f"{GITLAB_URL}/api/v4/projects/{project_id}/pipelines/{pipeline['id']}",
                headers=headers,verify=False
            )
            response.raise_for_status()
        except Exception as e:
            print(f"Warning: Unable to get pipeline detail for {pipeline['id']}: {e}")
            continue
        
        pipeline_detail = response.json()
        pipeline_low = [
                pipeline_detail['id'],
                pipeline_detail['ref'],
                pipeline_detail['started_at'],
                pipeline_detail['finished_at'],
                pipeline_detail['web_url'],
                pipeline_detail['status']
            ]
        table.add_row(pipeline_low)

        
        if not all:
            break
    print(table)
        
def retry_pipeline(gitlab_obj,project_id,pipeline_id):
    headers = HEADERS.copy()
    headers["Authorization"] = f"Bearer {gitlab_obj.token}"
    
    try:
        retry_pipeline_query = requests.post(
            url = f"{GITLAB_URL}/api/v4/projects/{project_id}/pipelines/{pipeline_id}/retry",
            headers=headers, verify=False
        )
        retry_pipeline_query.raise_for_status()
    except Exception as e:
        print(f"ERROR: Failed to retry the pipeline {pipeline_id} for project {pipeline_id} : {e}")
        exit(1)
        
    print(f"INFO: Pipeline {pipeline_id} in the project {project_id} has been successfully retried")

def main():
    parser = argparse.ArgumentParser(description="Gitlab Project and Pipeline Management Tool")
    subparsers = parser.add_subparsers(dest = "command")
    
    projects_parser = subparsers.add_parser("projects",help="Displays a list of projects in tabular format")
    projects_parser.add_argument('--all',action='store_true',help="Display all projects with or without pipeline")
    
    pipelines_parser = subparsers.add_parser("pipelines", help= "List/Retry a pipeine from a given project")
    pipelines_subparsers = pipelines_parser.add_subparsers(dest="pipeline_command")
    
    list_pipelines_parser = pipelines_subparsers.add_parser('list',help="List pipelines for a given project")
    list_pipelines_parser.add_argument('--all',action='store_true',help="Show all pipelines")
    list_pipelines_parser.add_argument('--project_id',type=int, required=True, help='PROJECT ID')
    
    retry_pipelines_parser = pipelines_subparsers.add_parser('retry',help="Retry pipelines for a given project")
    retry_pipelines_parser.add_argument('--project_id',type=int,required=True,help="PROJECT ID")
    retry_pipelines_parser.add_argument('--pipeline_id',type=int,required=True,help='PIPELINE_ID')
    
    args = parser.parse_args() 
    
    gitlab_obj = GitLab_Auth()
    if args.command == "projects":
        projects(gitlab_obj,args.all)
    elif args.command == "pipelines":
        if args.pipeline_command == "list":
            list_pipelines(gitlab_obj,args.all,args.project_id)
        elif args.pipeline_command == "retry":
            retry_pipeline(gitlab_obj,args.project_id,args.pipeline_id)
    
if __name__ =="__main__":
    main()

実行

$pip3 install prettytable
$source .envrc
$ python gitlab_api.py --help
usage: gitlab_api.py [-h] {projects,pipelines} ...

Gitlab Project and Pipeline Management Tool

positional arguments:
  {projects,pipelines}
    projects            Displays a list of projects in tabular format
    pipelines           List/Retry a pipeine from a given project

options:
  -h, --help            show this help message and exit
$ python gitlab_api.py projects
+----+-------------------+-----------------------------------+------------------------------------------+-----------------+
| ID | Name              | Group                             | URL                                      | Pipeline Status |
+----+-------------------+-----------------------------------+------------------------------------------+-----------------+
| 2  | MDP_OSPF_RESTCONF | Administrator / MDP_OSPF_RESTCONF | https://127.0.0.1/root/MDP_OSPF_RESTCONF | failed          |
+----+-------------------+-----------------------------------+------------------------------------------+-----------------+
$ python gitlab_api.py pipelines list --project_id 2 
+----+--------+--------------------------+--------------------------+---------------------------------------------------------+--------+
| ID | Branch | Started At               | Finished At              | URL                                                     | Status |
+----+--------+--------------------------+--------------------------+---------------------------------------------------------+--------+
| 16 | master | 2024-11-26T22:52:13.349Z | 2024-11-26T23:06:24.623Z | https://127.0.0.1/root/MDP_OSPF_RESTCONF/-/pipelines/16 | failed |
+----+--------+--------------------------+--------------------------+---------------------------------------------------------+--------+

$ python gitlab_api.py pipelines retry --project_id 2 --pipeline_id 16

$ python gitlab_api.py pipelines list --project_id 2 
+----+--------+--------------------------+--------------------------+---------------------------------------------------------+---------+
| ID | Branch | Started At               | Finished At              | URL                                                     | Status  |
+----+--------+--------------------------+--------------------------+---------------------------------------------------------+---------+
| 16 | master | 2024-11-26T22:52:13.349Z | 2024-11-26T23:06:24.623Z | https://127.0.0.1/root/MDP_OSPF_RESTCONF/-/pipelines/16 | running |
+----+--------+--------------------------+--------------------------+---------------------------------------------------------+---------+

課題
argparseによってコマンドアプリが簡単に作れるが使い方がいまいちよく分からない。練習が必要。