mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-10-24 12:13:33 +08:00
Move more workflow logic out of controller into manager.
This logic for building up editor representation of the workflow. Introduce concept of a workflow to dict style - with 'export' and 'editor' as first cracks.
This commit is contained in:
John Chilton
@@ -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 = """<div class="toolForm tool-node-error"><div class="toolFormTitle form-row-error">Unrecognized Tool: %s</div><div class="toolFormBody"><div class="form-row">
|
||||
The tool id '%s' for this tool is unrecognized.<br/><br/>To save this workflow, you will need to delete this step or enable the tool.
|
||||
</div></div></div>""" % (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 )
|
||||
|
||||
@@ -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."
|
||||
|
||||
@@ -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 = """<div class="toolForm tool-node-error"><div class="toolFormTitle form-row-error">Unrecognized Tool: %s</div><div class="toolFormBody"><div class="form-row">
|
||||
The tool id '%s' for this tool is unrecognized.<br/><br/>To save this workflow, you will need to delete this step or enable the tool.
|
||||
</div></div></div>""" % (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 ):
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user