diff --git a/lib/galaxy/managers/workflows.py b/lib/galaxy/managers/workflows.py index cf48239699f..a0faebde8ad 100644 --- a/lib/galaxy/managers/workflows.py +++ b/lib/galaxy/managers/workflows.py @@ -14,6 +14,7 @@ from galaxy.workflow.steps import attach_ordered_steps from galaxy.workflow.modules import module_factory, is_tool_module_type, ToolModule from galaxy.tools.parameters.basic import DataToolParameter, DataCollectionToolParameter from galaxy.tools.parameters import visit_input_values +from galaxy.web import url_for class WorkflowsManager( object ): @@ -314,8 +315,143 @@ class WorkflowContentsManager(UsesAnnotations): errors.append( "This workflow contains cycles" ) return workflow, errors - def workflow_to_dict(self, trans, stored, style="export" ): + def workflow_to_dict( self, trans, stored, style="export" ): + """ Export the workflow contents to a dictionary ready for JSON-ification and to be + sent out via API for instance. There are two styles of export allowed 'editor' and 'export'. + The Galaxy team will do it best to preserve the backward compatibility of the 'export' stye - + but the 'editor' style is subject to rapid and unannounced changes. """ + if style == "editor": + return self._workflow_to_dict_editor( trans, stored ) + else: + return self._workflow_to_dict_export( trans, stored ) + + def _workflow_to_dict_editor(self, trans, stored): + """ + """ + workflow = stored.latest_workflow + # Pack workflow data into a dictionary and return + data = {} + data['name'] = workflow.name + data['steps'] = {} + data['upgrade_messages'] = {} + # For each step, rebuild the form and encode the state + for step in workflow.steps: + # Load from database representation + module = module_factory.from_workflow_step( trans, step ) + if not module: + step_annotation = self.get_item_annotation_obj( trans.sa_session, trans.user, step ) + annotation_str = "" + if step_annotation: + annotation_str = step_annotation.annotation + invalid_tool_form_html = """
Unrecognized Tool: %s
+ The tool id '%s' for this tool is unrecognized.

To save this workflow, you will need to delete this step or enable the tool. +
""" % (step.tool_id, step.tool_id) + step_dict = { + 'id': step.order_index, + 'type': 'invalid', + 'tool_id': step.tool_id, + 'name': 'Unrecognized Tool: %s' % step.tool_id, + 'tool_state': None, + 'tooltip': None, + 'tool_errors': ["Unrecognized Tool Id: %s" % step.tool_id], + 'data_inputs': [], + 'data_outputs': [], + 'form_html': invalid_tool_form_html, + 'annotation': annotation_str, + 'input_connections': {}, + 'post_job_actions': {}, + 'workflow_outputs': [] + } + # Position + step_dict['position'] = step.position + # Add to return value + data['steps'][step.order_index] = step_dict + continue + # Fix any missing parameters + upgrade_message = module.check_and_update_state() + if upgrade_message: + # FIXME: Frontend should be able to handle workflow messages + # as a dictionary not just the values + data['upgrade_messages'][step.order_index] = upgrade_message.values() + # Get user annotation. + step_annotation = self.get_item_annotation_obj( trans.sa_session, trans.user, step ) + annotation_str = "" + if step_annotation: + annotation_str = step_annotation.annotation + # Pack attributes into plain dictionary + step_dict = { + 'id': step.order_index, + 'type': module.type, + 'tool_id': module.get_tool_id(), + 'name': module.get_name(), + 'tool_state': module.get_state(), + 'tooltip': module.get_tooltip( static_path=url_for( '/static' ) ), + 'tool_errors': module.get_errors(), + 'data_inputs': module.get_data_inputs(), + 'data_outputs': module.get_data_outputs(), + 'form_html': module.get_config_form(), + 'annotation': annotation_str, + 'post_job_actions': {}, + 'workflow_outputs': [] + } + # Connections + input_connections = step.input_connections + input_connections_type = {} + multiple_input = {} # Boolean value indicating if this can be mutliple + if step.type is None or step.type == 'tool': + # Determine full (prefixed) names of valid input datasets + data_input_names = {} + + def callback( input, value, prefixed_name, prefixed_label ): + if isinstance( input, DataToolParameter ) or isinstance( input, DataCollectionToolParameter ): + data_input_names[ prefixed_name ] = True + multiple_input[ prefixed_name ] = input.multiple + if isinstance( input, DataToolParameter ): + input_connections_type[ input.name ] = "dataset" + if isinstance( input, DataCollectionToolParameter ): + input_connections_type[ input.name ] = "dataset_collection" + visit_input_values( module.tool.inputs, module.state.inputs, callback ) + # Filter + # FIXME: this removes connection without displaying a message currently! + input_connections = [ conn for conn in input_connections if conn.input_name in data_input_names ] + # post_job_actions + pja_dict = {} + for pja in step.post_job_actions: + pja_dict[pja.action_type + pja.output_name] = dict( + action_type=pja.action_type, + output_name=pja.output_name, + action_arguments=pja.action_arguments + ) + step_dict['post_job_actions'] = pja_dict + #workflow outputs + outputs = [] + for output in step.workflow_outputs: + outputs.append(output.output_name) + step_dict['workflow_outputs'] = outputs + # Encode input connections as dictionary + input_conn_dict = {} + for conn in input_connections: + input_type = "dataset" + if conn.input_name in input_connections_type: + input_type = input_connections_type[ conn.input_name ] + conn_dict = dict( id=conn.output_step.order_index, output_name=conn.output_name, input_type=input_type ) + if conn.input_name in multiple_input: + if conn.input_name in input_conn_dict: + input_conn_dict[ conn.input_name ].append( conn_dict ) + else: + input_conn_dict[ conn.input_name ] = [ conn_dict ] + else: + input_conn_dict[ conn.input_name ] = conn_dict + step_dict['input_connections'] = input_conn_dict + # Position + step_dict['position'] = step.position + # Add to return value + data['steps'][step.order_index] = step_dict + return data + + def _workflow_to_dict_export( self, trans, stored ): + """ Export the workflow contents to a dictionary ready for JSON-ification and export. """ workflow = stored.latest_workflow workflow_annotation = self.get_item_annotation_obj( trans.sa_session, trans.user, stored ) diff --git a/lib/galaxy/webapps/galaxy/api/workflows.py b/lib/galaxy/webapps/galaxy/api/workflows.py index e2e8d420609..91bc8aa9819 100644 --- a/lib/galaxy/webapps/galaxy/api/workflows.py +++ b/lib/galaxy/webapps/galaxy/api/workflows.py @@ -28,6 +28,7 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesHis super( BaseAPIController, self ).__init__( app ) self.history_manager = histories.HistoryManager() self.workflow_manager = workflows.WorkflowsManager( app ) + self.workflow_contents_manager = workflows.WorkflowContentsManager() @expose_api def index(self, trans, **kwd): @@ -221,7 +222,8 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesHis """ stored_workflow = self.__get_stored_accessible_workflow( trans, workflow_id ) - ret_dict = self._workflow_to_dict( trans, stored_workflow ) + style = kwd.get("style", "export") + ret_dict = self.workflow_contents_manager.workflow_to_dict( trans, stored_workflow, style=style ) if not ret_dict: # This workflow has a tool that's missing from the distribution message = "Workflow cannot be exported due to missing tools." diff --git a/lib/galaxy/webapps/galaxy/controllers/workflow.py b/lib/galaxy/webapps/galaxy/controllers/workflow.py index ec9950a3a2d..11baa0ef945 100644 --- a/lib/galaxy/webapps/galaxy/controllers/workflow.py +++ b/lib/galaxy/webapps/galaxy/controllers/workflow.py @@ -669,126 +669,8 @@ class WorkflowController( BaseUIController, SharableMixin, UsesStoredWorkflowMix """ trans.workflow_building_mode = True stored = self.get_stored_workflow( trans, id, check_ownership=True, check_accessible=False ) - workflow = stored.latest_workflow - # Pack workflow data into a dictionary and return - data = {} - data['name'] = workflow.name - data['steps'] = {} - data['upgrade_messages'] = {} - # For each step, rebuild the form and encode the state - for step in workflow.steps: - # Load from database representation - module = module_factory.from_workflow_step( trans, step ) - if not module: - step_annotation = self.get_item_annotation_obj( trans.sa_session, trans.user, step ) - annotation_str = "" - if step_annotation: - annotation_str = step_annotation.annotation - invalid_tool_form_html = """
Unrecognized Tool: %s
- The tool id '%s' for this tool is unrecognized.

To save this workflow, you will need to delete this step or enable the tool. -
""" % (step.tool_id, step.tool_id) - step_dict = { - 'id': step.order_index, - 'type': 'invalid', - 'tool_id': step.tool_id, - 'name': 'Unrecognized Tool: %s' % step.tool_id, - 'tool_state': None, - 'tooltip': None, - 'tool_errors': ["Unrecognized Tool Id: %s" % step.tool_id], - 'data_inputs': [], - 'data_outputs': [], - 'form_html': invalid_tool_form_html, - 'annotation': annotation_str, - 'input_connections': {}, - 'post_job_actions': {}, - 'workflow_outputs': [] - } - # Position - step_dict['position'] = step.position - # Add to return value - data['steps'][step.order_index] = step_dict - continue - # Fix any missing parameters - upgrade_message = module.check_and_update_state() - if upgrade_message: - # FIXME: Frontend should be able to handle workflow messages - # as a dictionary not just the values - data['upgrade_messages'][step.order_index] = upgrade_message.values() - # Get user annotation. - step_annotation = self.get_item_annotation_obj( trans.sa_session, trans.user, step ) - annotation_str = "" - if step_annotation: - annotation_str = step_annotation.annotation - # Pack attributes into plain dictionary - step_dict = { - 'id': step.order_index, - 'type': module.type, - 'tool_id': module.get_tool_id(), - 'name': module.get_name(), - 'tool_state': module.get_state(), - 'tooltip': module.get_tooltip( static_path=url_for( '/static' ) ), - 'tool_errors': module.get_errors(), - 'data_inputs': module.get_data_inputs(), - 'data_outputs': module.get_data_outputs(), - 'form_html': module.get_config_form(), - 'annotation': annotation_str, - 'post_job_actions': {}, - 'workflow_outputs': [] - } - # Connections - input_connections = step.input_connections - input_connections_type = {} - multiple_input = {} # Boolean value indicating if this can be mutliple - if step.type is None or step.type == 'tool': - # Determine full (prefixed) names of valid input datasets - data_input_names = {} - - def callback( input, value, prefixed_name, prefixed_label ): - if isinstance( input, DataToolParameter ) or isinstance( input, DataCollectionToolParameter ): - data_input_names[ prefixed_name ] = True - multiple_input[ prefixed_name ] = input.multiple - if isinstance( input, DataToolParameter ): - input_connections_type[ input.name ] = "dataset" - if isinstance( input, DataCollectionToolParameter ): - input_connections_type[ input.name ] = "dataset_collection" - visit_input_values( module.tool.inputs, module.state.inputs, callback ) - # Filter - # FIXME: this removes connection without displaying a message currently! - input_connections = [ conn for conn in input_connections if conn.input_name in data_input_names ] - # post_job_actions - pja_dict = {} - for pja in step.post_job_actions: - pja_dict[pja.action_type + pja.output_name] = dict( - action_type=pja.action_type, - output_name=pja.output_name, - action_arguments=pja.action_arguments - ) - step_dict['post_job_actions'] = pja_dict - #workflow outputs - outputs = [] - for output in step.workflow_outputs: - outputs.append(output.output_name) - step_dict['workflow_outputs'] = outputs - # Encode input connections as dictionary - input_conn_dict = {} - for conn in input_connections: - input_type = "dataset" - if conn.input_name in input_connections_type: - input_type = input_connections_type[ conn.input_name ] - conn_dict = dict( id=conn.output_step.order_index, output_name=conn.output_name, input_type=input_type ) - if conn.input_name in multiple_input: - if conn.input_name in input_conn_dict: - input_conn_dict[ conn.input_name ].append( conn_dict ) - else: - input_conn_dict[ conn.input_name ] = [ conn_dict ] - else: - input_conn_dict[ conn.input_name ] = conn_dict - step_dict['input_connections'] = input_conn_dict - # Position - step_dict['position'] = step.position - # Add to return value - data['steps'][step.order_index] = step_dict - return data + workflow_contents_manager = workflows.WorkflowContentsManager() + return workflow_contents_manager.workflow_to_dict( trans, stored, style="editor" ) @web.json def save_workflow( self, trans, id, workflow_data ): diff --git a/test/api/test_workflows.py b/test/api/test_workflows.py index 3c054ed44b8..f2cb8c41bf9 100644 --- a/test/api/test_workflows.py +++ b/test/api/test_workflows.py @@ -324,6 +324,46 @@ class WorkflowsApiTestCase( BaseWorkflowsApiTestCase ): first_input = downloaded_workflow[ "steps" ][ "0" ][ "inputs" ][ 0 ] assert first_input[ "name" ] == "WorkflowInput1" assert first_input[ "description" ] == "input1 description" + self._assert_has_keys( downloaded_workflow, "a_galaxy_workflow", "format-version", "annotation", "uuid", "steps" ) + for step in downloaded_workflow["steps"].values(): + self._assert_has_keys( + step, + 'id', + 'type', + 'tool_id', + 'tool_version', + 'name', + 'tool_state', + 'tool_errors', + 'annotation', + 'inputs', + 'user_outputs', + 'outputs' + ) + if step['type'] == "tool": + self._assert_has_keys( step, "post_job_actions" ) + + def test_export_editor( self ): + uploaded_workflow_id = self.workflow_populator.simple_workflow( "test_for_export" ) + downloaded_workflow = self._download_workflow( uploaded_workflow_id, style="editor" ) + self._assert_has_keys( downloaded_workflow, "name", "steps", "upgrade_messages" ) + for step in downloaded_workflow["steps"].values(): + self._assert_has_keys( + step, + 'id', + 'type', + 'tool_id', + 'name', + 'tool_state', + 'tooltip', + 'tool_errors', + 'data_inputs', + 'data_outputs', + 'form_html', + 'annotation', + 'post_job_actions', + 'workflow_outputs', + ) def test_import_export_with_runtime_inputs( self ): workflow = self.workflow_populator.load_workflow_from_resource( name="test_workflow_with_runtime_input" ) @@ -771,8 +811,11 @@ class WorkflowsApiTestCase( BaseWorkflowsApiTestCase ): ) return self._post( route, import_data ) - def _download_workflow(self, workflow_id): - download_response = self._get( "workflows/%s/download" % workflow_id ) + def _download_workflow(self, workflow_id, style=None): + params = {} + if style: + params = {"style": style} + download_response = self._get( "workflows/%s/download" % workflow_id, params ) self._assert_status_code_is( download_response, 200 ) downloaded_workflow = download_response.json() return downloaded_workflow