gbdx_cloud_harness/workflow.py
def _build_worklfow_json(self):
        """
        Build a workflow definition from the cloud_harness task.
        """
        wf_json = {'tasks': [], 'name': 'cloud-harness_%s' % str(uuid.uuid4())}

        task_def = json.loads(self.task_template.json())

        d = {
            "name": task_def['name'],
            "outputs": [],
            "inputs": [],
            "taskType": task_def['taskType']
        }

        # Add input ports
        for port in self.task_template.input_ports:
            port_value = port.value

            if port_value is False:
                port_value = 'false'
            if port_value is True:
                port_value = 'true'

            d['inputs'].append({
                "name": port._name,
                "value": port_value
            })

        # Add output ports
        for port in self.task_template.output_ports:
            d['outputs'].append({
                "name": port._name
            })

        # Add task to workflow
        wf_json['tasks'].append(d)

        # Add port to be saved
        for port in self.task_template.output_ports:
            # Add save data locations
            if hasattr(port, 'stageToS3') and port.stageToS3:
                save_location = '{customer_storage}/{run_name}/{port}'.format(
                    customer_storage=self.storage.location,
                    run_name=self.task_template.run_name,
                    port=port.name
                )
                new_task = dict(**self.STAGE_TO_S3)
                new_task['inputs'] = [
                    {'name': 'data', 'source': '%s:%s' % (task_def['name'], port._name)},
                    {'name': 'destination', 'value': save_location}
                ]
                wf_json['tasks'].append(new_task)

        return wf_json
def _build_worklfow_json(self):
        """
        Build a workflow definition from the cloud_harness task.
        """
        wf_json = {'tasks': [], 'name': 'cloud-harness_%s' % str(uuid.uuid4())}

        task_def = json.loads(self.task_template.json())

        d = {
            "name": task_def['name'],
            "outputs": [],
            "inputs": [],
            "taskType": task_def['taskType']
        }

        # Add input ports
        for port in self.task_template.input_ports:
            port_value = port.value

            if port_value is False:
                port_value = 'false'
            if port_value is True:
                port_value = 'true'

            d['inputs'].append({
                "name": port._name,
                "value": port_value
            })

        # Add output ports
        for port in self.task_template.output_ports:
            d['outputs'].append({
                "name": port._name
            })

        # Add task to workflow
        wf_json['tasks'].append(d)

        # Add port to be saved
        for port in self.task_template.output_ports:
            # Add save data locations
            if hasattr(port, 'stageToS3') and port.stageToS3:
                save_location = '{customer_storage}/{run_name}/{port}'.format(
                    customer_storage=self.storage.location,
                    run_name=self.task_template.run_name,
                    port=port.name
                )
                new_task = dict(**self.STAGE_TO_S3)
                new_task['inputs'] = [
                    {'name': 'data', 'source': '%s:%s' % (task_def['name'], port._name)},
                    {'name': 'destination', 'value': save_location}
                ]
                wf_json['tasks'].append(new_task)

        return wf_json
Build a workflow definition from the cloud_harness task.