get_pipeline_jobs.py

#!/usr/bin/env python3
"""

Queries the jobs of a given pipeline, filtered by stage

"""

import argparse
import logging
import sys

import gitlab as gl

import common

__author__ = "Jonas Höppner"
__email__ = "jonas.hoeppner@garz-fricke.com"

verbose = 0


def get_pipeline_pipelinejobs(gitlab, project, pipeline, name=None, stage=None):
    # Accept either an gitlab object, or name or id to specify a project
    if isinstance(project, gl.v4.objects.projects.Project):
        gl_project = project
    else:
        gl_project = gitlab.projects.get(project)

    gl_pipeline = gl_project.pipelines.get(pipeline)
    jobs = gl_pipeline.jobs.list(all=True)
    if stage is not None:
        jobs = list(filter(lambda j: j.stage == stage, jobs))
    if name is not None:
        jobs = list(filter(lambda j: j.name == name, jobs))

    return jobs


def get_pipeline_jobs(gitlab, project, pipeline, name=None, stage=None):
    # Accept either an gitlab object, or name or id to specify a project
    if isinstance(project, gl.v4.objects.projects.Project):
        gl_project = project
    else:
        gl_project = gitlab.projects.get(project)

    pipeline_jobs = get_pipeline_pipelinejobs(gitlab, project, pipeline, name, stage)
    jobs = []
    # Project Jobs from Pipeline Job
    for job in pipeline_jobs:
        jobs.append(gl_project.jobs.get(job.id))
    return jobs


def get_pipeline_job_ids(gitlab, project, pipeline, name=None, stage=None):
    jobs = get_pipeline_pipelinejobs(gitlab, project, pipeline, name, stage)
    return [j.id for j in jobs]


def main(args):
    parser = argparse.ArgumentParser(description=__doc__, usage="%(prog)s [OPTIONS]")

    parser.add_argument(
        "--gitlab-url",
        help="""URL to the GitLab instance""",
        dest="gitlab_url",
        action="store",
        default=common.GITLAB_URL,
    )
    parser.add_argument(
        "--token",
        help="""GitLab REST API private access token""",
        dest="token",
        required=True,
    )
    parser.add_argument(
        "--project",
        action="store",
        dest="project",
        help="Specify the project by either by id or by path.",
        required=True,
    )
    parser.add_argument(
        "--pipeline",
        action="store",
        dest="pipeline",
        help="Specify the pipeline by id.",
    )
    parser.add_argument(
        "-s",
        "--stage",
        action="store",
        default=None,
        help="Filter the jobs by the given stage, if omnitted all jobs are returned.",
    )
    parser.add_argument(
        "-n",
        "--name",
        action="store",
        default=None,
        help="Filter the jobs by given name, if omnitted all jobs are returned.",
    )
    parser.add_argument(
        "-v",
        "--verbose",
        action="count",
        dest="verbose",
        default=0,
        help="Increase verbosity.",
    )

    options = parser.parse_args(args)
    if options.verbose:
        logging.basicConfig(level=logging.DEBUG)

    logging.debug(options)
    gitlab = gl.Gitlab(options.gitlab_url, private_token=options.token)
    jobs = get_pipeline_jobs(
        gitlab, options.project, options.pipeline, options.name, options.stage
    )

    for j in jobs:
        print(j.name, j.id)

    job_ids = get_pipeline_job_ids(
        gitlab, options.project, options.pipeline, options.name, options.stage
    )
    for j in job_ids:
        print(j)


if __name__ == "__main__":
    main(sys.argv[1:])