def launch_pipeline()

in pipe-cli/src/api/pipeline.py [0:0]

• 58 lines of code
• 24 McCabe index (conditional complexity)

    def launch_pipeline(cls, pipeline_id, version, parameters,
                        instance_disk=None, instance_type=None,
                        docker_image=None, cmd_template=None,
                        timeout=None, config_name=None, instance_count=None,
                        price_type=None, region_id=None, parent_node=None, non_pause=None, friendly_url=None,
                        status_notifications=False,
                        status_notifications_status=None, status_notifications_recipient=None,
                        status_notifications_subject=None, status_notifications_body=None,
                        run_as_user=None):
        api = cls.instance()
        params = {}
        for parameter in parameters:
            if parameter.value:
                parameter_value = cls.parse_parameter_value(parameter.value)
                params[parameter.name] = {'value': parameter_value[0], 'type': parameter_value[1] if parameter_value[2] else parameter.parameter_type}
        payload = {'pipelineId': pipeline_id, 'version': version, 'params': params}
        if instance_disk is not None:
            payload['hddSize'] = instance_disk
        if instance_type is not None:
            payload['instanceType'] = instance_type
        if docker_image is not None:
            payload['dockerImage'] = docker_image
        if cmd_template is not None:
            payload['cmdTemplate'] = cmd_template
        if timeout is not None:
            payload['timeout'] = timeout
        if config_name is not None:
            payload['configurationName'] = config_name
        if instance_count is not None and instance_count > 0:
            payload['nodeCount'] = instance_count
        if price_type:
            payload['isSpot'] = price_type == PriceType.SPOT
        if region_id is not None:
            payload['cloudRegionId'] = region_id
        if parent_node is not None:
            cls.__add_parent_node_params(payload, parent_node)
        if non_pause is not None:
            payload['nonPause'] = non_pause
        if friendly_url:
            payload['prettyUrl'] = friendly_url
        if run_as_user:
            payload['runAs'] = run_as_user
        if status_notifications:
            if status_notifications_body:
                with open(status_notifications_body, 'r') as f:
                    status_notifications_body_content = f.read()
            else:
                status_notifications_body_content = None
            payload['notifications'] = [{
                'type': 'PIPELINE_RUN_STATUS',
                'triggerStatuses': status_notifications_status or None,
                'recipients': status_notifications_recipient or None,
                'subject': status_notifications_subject or None,
                'body': status_notifications_body_content or None
            }]
        data = json.dumps(payload)
        response_data = api.call('run', data)
        return PipelineRunModel.load(response_data['payload'])