| 116 | |
| 117 | |
| 118 | def enqueue_pipeline(pipeline_id): |
| 119 | pipeline_schedule = models.PipelineSchedule.objects.get(pipeline_id=pipeline_id) |
| 120 | queue = queues.get(pipeline_schedule.get_run_priority_display()) |
| 121 | |
| 122 | if pipeline_schedule.status in [ |
| 123 | models.PipelineRun.Status.RUNNING, |
| 124 | models.PipelineRun.Status.QUEUED, |
| 125 | ]: |
| 126 | logger.warning( |
| 127 | ( |
| 128 | f"Cannot enqueue a new execution for {pipeline_id} " |
| 129 | "until the previous one has finished." |
| 130 | ) |
| 131 | ) |
| 132 | return |
| 133 | |
| 134 | run = models.PipelineRun.objects.create( |
| 135 | pipeline=pipeline_schedule, |
| 136 | ) |
| 137 | job = queue.enqueue( |
| 138 | execute_pipeline, |
| 139 | pipeline_id, |
| 140 | run.run_id, |
| 141 | job_id=str(run.run_id), |
| 142 | on_failure=set_run_failure, |
| 143 | job_timeout=f"{pipeline_schedule.execution_timeout}h", |
| 144 | ) |
| 145 | |
| 146 | |
| 147 | def dequeue_job(job_id): |