@@ -0,0 +1,2756 @@
+import logging%0A%0Afrom pipelines.utils import (%0A    get_pipeline_run,%0A    stop_operation_runs_for_pipeline_run,%0A    skip_operation_runs_for_pipeline_run,%0A)%0Afrom polyaxon.celery_api import app as celery_app%0A%0Afrom pipelines import dags%0Afrom pipelines.constants import OperationStatuses%0Afrom polyaxon.config_settings import CeleryTasks, Intervals%0A%0Alogger = logging.getLogger('polyaxon.tasks.pipelines')%0A%0A%0A@celery_app.task(name=CeleryTasks.PIPELINES_START, bind=True, max_retries=None)%0Adef start_pipeline(self, pipeline_run_id):%0A    pipeline_run = get_pipeline_run(pipeline_run_id=pipeline_run_id)%0A    if not pipeline_run:%0A        logger.info('Pipeline %60%7B%7D%60 does not exist any more.'.format(pipeline_run_id))%0A%0A    pipeline_run.on_schedule()%0A    dag, op_runs = pipeline_run.dag%0A    independent_op_run_ids = dags.get_independent_nodes(dag=dag)%0A    op_runs_to_start = %5Bop_runs%5Bop_run_id%5D for op_run_id in independent_op_run_ids%0A                        if op_runs%5Bop_run_id%5D.last_status == OperationStatuses.CREATED%5D%0A    concurrency = pipeline_run.pipeline.concurrency%0A    concurrency = concurrency or len(op_runs_to_start)%0A    while op_runs_to_start:%0A        op_run = op_runs_to_start.pop()%0A        op_run.schedule_start()%0A        concurrency -= 1%0A        if concurrency == 0:%0A            break%0A%0A    if op_runs_to_start:%0A        # Schedule another task%0A        self.retry(countdown=Intervals.PIPELINES_SCHEDULER)%0A%0A%0A@celery_app.task(name=CeleryTasks.PIPELINES_STOP_OPERATIONS, bind=True, max_retries=None)%0Adef stop_pipeline_operation_runs(pipeline_run_id, message=None):%0A    pipeline_run = get_pipeline_run(pipeline_run_id=pipeline_run_id)%0A    if not pipeline_run:%0A        logger.info('Pipeline %60%7B%7D%60 does not exist any more.'.format(pipeline_run_id))%0A%0A    stop_operation_runs_for_pipeline_run(pipeline_run, message=message)%0A%0A%0A@celery_app.task(name=CeleryTasks.PIPELINES_SKIP_OPERATIONS, bind=True, max_retries=None)%0Adef skip_pipeline_operation_runs(pipeline_run_id, message=None):%0A    pipeline_run = get_pipeline_run(pipeline_run_id=pipeline_run_id)%0A    if not pipeline_run:%0A        logger.info('Pipeline %60%7B%7D%60 does not exist any more.'.format(pipeline_run_id))%0A%0A    # We stop all op runs first%0A    stop_operation_runs_for_pipeline_run(pipeline_run, message=message)%0A    # Then we marked them as skipped%0A    skip_operation_runs_for_pipeline_run(pipeline_run, message=message)%0A%0A%0A@celery_app.task(name=CeleryTasks.PIPELINES_CHECK_STATUS, ignore_result=True)%0Adef check_pipeline_run_status(pipeline_run_id, status, message=None):%0A    pipeline_run = get_pipeline_run(pipeline_run_id=pipeline_run_id)%0A    if not pipeline_run:%0A        logger.info('Pipeline %60%7B%7D%60 does not exist any more.'.format(pipeline_run_id))%0A%0A    pipeline_run.set_status(status=status, message=message)%0A

Add pipeline tasks