Pipelines and Jobs#
Project pipelines#
A pipeline is a group of jobs executed by GitLab CI.
Reference#
Examples#
List pipelines for a project:
pipelines = project.pipelines.list()
Get a pipeline for a project:
pipeline = project.pipelines.get(pipeline_id)
Get variables of a pipeline:
variables = pipeline.variables.list()
Create a pipeline for a particular reference with custom variables:
pipeline = project.pipelines.create({'ref': 'main', 'variables': [{'key': 'MY_VARIABLE', 'value': 'hello'}]})
Retry the failed builds for a pipeline:
pipeline.retry()
Cancel builds in a pipeline:
pipeline.cancel()
Delete a pipeline:
pipeline.delete()
Triggers#
Triggers provide a way to interact with the GitLab CI. Using a trigger a user or an application can run a new build/job for a specific commit.
Reference#
Examples#
List triggers:
triggers = project.triggers.list()
Get a trigger:
trigger = project.triggers.get(trigger_token)
Create a trigger:
trigger = project.triggers.create({'description': 'mytrigger'})
Remove a trigger:
project.triggers.delete(trigger_token)
# or
trigger.delete()
Full example with wait for finish:
def get_or_create_trigger(project):
trigger_decription = 'my_trigger_id'
for t in project.triggers.list():
if t.description == trigger_decription:
return t
return project.triggers.create({'description': trigger_decription})
trigger = get_or_create_trigger(project)
pipeline = project.trigger_pipeline('main', trigger.token, variables={"DEPLOY_ZONE": "us-west1"})
while pipeline.finished_at is None:
pipeline.refresh()
time.sleep(1)
You can trigger a pipeline using token authentication instead of user authentication. To do so create an anonymous Gitlab instance and use lazy objects to get the associated project:
gl = gitlab.Gitlab(URL) # no authentication
project = gl.projects.get(project_id, lazy=True) # no API call
project.trigger_pipeline('main', trigger_token)
Reference: https://docs.gitlab.com/ee/ci/triggers/#trigger-token
Pipeline schedules#
You can schedule pipeline runs using a cron-like syntax. Variables can be associated with the scheduled pipelines.
Reference#
v4 API
GitLab API: https://docs.gitlab.com/ce/api/pipeline_schedules.html
Examples#
List pipeline schedules:
scheds = project.pipelineschedules.list()
Get a single schedule:
sched = project.pipelineschedules.get(schedule_id)
Create a new schedule:
sched = project.pipelineschedules.create({
'ref': 'main',
'description': 'Daily test',
'cron': '0 1 * * *'})
Update a schedule:
sched.cron = '1 2 * * *'
sched.save()
Take ownership of a schedule:
sched.take_ownership()
Trigger a pipeline schedule immediately:
sched = projects.pipelineschedules.get(schedule_id)
sched.play()
Delete a schedule:
sched.delete()
List schedule variables:
# note: you need to use get() to retrieve the schedule variables. The
# attribute is not present in the response of a list() call
sched = projects.pipelineschedules.get(schedule_id)
vars = sched.attributes['variables']
Create a schedule variable:
var = sched.variables.create({'key': 'foo', 'value': 'bar'})
Edit a schedule variable:
var.value = 'new_value'
var.save()
Delete a schedule variable:
var.delete()
List all pipelines triggered by a pipeline schedule:
pipelines = sched.pipelines.list()
Jobs#
Jobs are associated to projects, pipelines and commits. They provide information on the jobs that have been run, and methods to manipulate them.
Reference#
Examples#
Jobs are usually automatically triggered, but you can explicitly trigger a new job:
project.trigger_build('main', trigger_token,
{'extra_var1': 'foo', 'extra_var2': 'bar'})
List jobs for the project:
jobs = project.jobs.list()
Get a single job:
project.jobs.get(job_id)
List the jobs of a pipeline:
project = gl.projects.get(project_id)
pipeline = project.pipelines.get(pipeline_id)
jobs = pipeline.jobs.list()
Note
Job methods (play, cancel, and so on) are not available on
ProjectPipelineJob
objects. To use these methods create a ProjectJob
object:
pipeline_job = pipeline.jobs.list()[0]
job = project.jobs.get(pipeline_job.id, lazy=True)
job.retry()
Get the artifacts of a job:
build_or_job.artifacts()
Get the artifacts of a job by its name from the latest successful pipeline of a branch or tag:
project.artifacts.download(ref_name='main', job='build')
Warning
Artifacts are entirely stored in memory in this example.
You can download artifacts as a stream. Provide a callable to handle the stream:
with open("archive.zip", "wb") as f:
build_or_job.artifacts(streamed=True, action=f.write)
You can also directly stream the output into a file, and unzip it afterwards:
zipfn = "___artifacts.zip"
with open(zipfn, "wb") as f:
build_or_job.artifacts(streamed=True, action=f.write)
subprocess.run(["unzip", "-bo", zipfn])
os.unlink(zipfn)
Or, you can also use the underlying response iterator directly:
artifact_bytes_iterator = build_or_job.artifacts(iterator=True)
This can be used with frameworks that expect an iterator (such as FastAPI/Starlette’s
StreamingResponse
) to forward a download from GitLab without having to download
the entire content server-side first:
@app.get("/download_artifact")
def download_artifact():
artifact_bytes_iterator = build_or_job.artifacts(iterator=True)
return StreamingResponse(artifact_bytes_iterator, media_type="application/zip")
Delete all artifacts of a project that can be deleted:
project.artifacts.delete()
Get a single artifact file:
build_or_job.artifact('path/to/file')
Get a single artifact file by branch and job:
project.artifacts.raw('branch', 'path/to/file', 'job')
Mark a job artifact as kept when expiration is set:
build_or_job.keep_artifacts()
Delete the artifacts of a job:
build_or_job.delete_artifacts()
Get a job log file / trace:
build_or_job.trace()
Warning
Traces are entirely stored in memory unless you use the streaming feature. See the artifacts example.
Cancel/retry a job:
build_or_job.cancel()
build_or_job.retry()
Play (trigger) a job:
build_or_job.play()
Erase a job (artifacts and trace):
build_or_job.erase()
Pipeline bridges#
Get a list of bridge jobs (including child pipelines) for a pipeline.
Reference#
Examples#
List bridges for the pipeline:
bridges = pipeline.bridges.list()
Pipeline test report#
Get a pipeline’s complete test report.
Reference#
Examples#
Get the test report for a pipeline:
test_report = pipeline.test_report.get()
Pipeline test report summary#
Get a pipeline’s test report summary.
Reference#
Examples#
Get the test report summary for a pipeline:
test_report_summary = pipeline.test_report_summary.get()