opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Introduce yet another workflow to_dict style - 'instance'.

Browse Source 
This is the style used by workflow 'show' - this consolidates all step handling to_dict style-logic from controller/base.py, controller/workflow.py, and api/workflows.py into the new WorkflowContentsManager - with API test cases for everything.
This commit is contained in:
John Chilton
2014-12-15 22:20:11 -05:00
parent 58ce775d14
commit bac14a9f44
2 changed files with 49 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/managers/workflows.py
+47 -3
View File
@@ -317,12 +317,17 @@ class WorkflowContentsManager(UsesAnnotations):
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.
sent out via API for instance. There are three styles of export allowed 'export', 'instance', and
'editor'. The Galaxy team will do it best to preserve the backward compatibility of the
'export' stye - this is the export method meant to be portable across Galaxy instances and over
time. The 'editor' style is subject to rapid and unannounced changes. The 'instance' export
option describes the workflow in a context more tied to the current Galaxy instance and includes
fields like 'url' and 'url' and actual unencoded step ids instead of 'order_index'.
"""
if style == "editor":
return self._workflow_to_dict_editor( trans, stored )
elif style == "instance":
return self._workflow_to_dict_instance( trans, stored )
else:
return self._workflow_to_dict_export( trans, stored )
@@ -552,6 +557,45 @@ class WorkflowContentsManager(UsesAnnotations):
data['steps'][step.order_index] = step_dict
return data
def _workflow_to_dict_instance(self, trans, stored):
item = stored.to_dict( view='element', value_mapper={ 'id': trans.security.encode_id } )
workflow = stored.latest_workflow
item['url'] = url_for('workflow', id=item['id'])
item['owner'] = stored.user.username
inputs = {}
for step in workflow.steps:
step_type = step.type
if step_type in ['data_input', 'data_collection_input']:
if step.tool_inputs and "name" in step.tool_inputs:
label = step.tool_inputs['name']
elif step_type == "data_input":
label = "Input Dataset"
elif step_type == "data_collection_input":
label = "Input Dataset Collection"
else:
raise ValueError("Invalid step_type %s" % step_type)
inputs[step.id] = {'label': label, 'value': ""}
else:
pass
# Eventually, allow regular tool parameters to be inserted and modified at runtime.
# p = step.get_required_parameters()
item['inputs'] = inputs
item['annotation'] = self.get_item_annotation_str( trans.sa_session, stored.user, stored )
steps = {}
for step in workflow.steps:
steps[step.id] = {'id': step.id,
'type': step.type,
'tool_id': step.tool_id,
'tool_version': step.tool_version,
'annotation': self.get_item_annotation_str( trans.sa_session, stored.user, step ),
'tool_inputs': step.tool_inputs,
'input_steps': {}}
for conn in step.input_connections:
steps[step.id]['input_steps'][conn.input_name] = {'source_step': conn.output_step_id,
'step_output': conn.output_name}
item['steps'] = steps
return item
class MissingToolsException(object):
lib/galaxy/webapps/galaxy/api/workflows.py
+2 -41
View File
@@ -76,8 +76,7 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesHis
if trans.sa_session.query(trans.app.model.StoredWorkflowUserShareAssociation).filter_by(user=trans.user, stored_workflow=stored_workflow).count() == 0:
message = "Workflow is neither importable, nor owned by or shared with current user"
raise exceptions.ItemAccessibilityException( message )
latest_workflow = stored_workflow.latest_workflow
return self.__encode_workflow( trans, stored_workflow, latest_workflow )
return self.workflow_contents_manager.workflow_to_dict( trans, stored_workflow, style="instance" )
@expose_api
def create(self, trans, payload, **kwd):
@@ -303,7 +302,7 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesHis
else:
message = "Updating workflow requires dictionary containing 'workflow' attribute with new JSON description."
raise exceptions.RequestParameterInvalidException( message )
return self.__encode_workflow( trans, stored_workflow, workflow )
return self.workflow_contents_manager.workflow_to_dict( trans, stored_workflow, style="instance" )
def __api_import_new_workflow( self, trans, payload, **kwd ):
data = payload['workflow']
@@ -510,44 +509,6 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesHis
)
return self.__encode_invocation_step( trans, invocation_step )
def __encode_workflow( self, trans, stored_workflow, workflow ):
item = stored_workflow.to_dict( view='element', value_mapper={ 'id': trans.security.encode_id } )
item['url'] = url_for('workflow', id=item['id'])
item['owner'] = stored_workflow.user.username
inputs = {}
for step in workflow.steps:
step_type = step.type
if step_type in ['data_input', 'data_collection_input']:
if step.tool_inputs and "name" in step.tool_inputs:
label = step.tool_inputs['name']
elif step_type == "data_input":
label = "Input Dataset"
elif step_type == "data_collection_input":
label = "Input Dataset Collection"
else:
raise ValueError("Invalid step_type %s" % step_type)
inputs[step.id] = {'label': label, 'value': ""}
else:
pass
# Eventually, allow regular tool parameters to be inserted and modified at runtime.
# p = step.get_required_parameters()
item['inputs'] = inputs
item['annotation'] = self.get_item_annotation_str( trans.sa_session, stored_workflow.user, stored_workflow )
steps = {}
for step in workflow.steps:
steps[step.id] = {'id': step.id,
'type': step.type,
'tool_id': step.tool_id,
'tool_version': step.tool_version,
'annotation': self.get_item_annotation_str( trans.sa_session, stored_workflow.user, step ),
'tool_inputs': step.tool_inputs,
'input_steps': {}}
for conn in step.input_connections:
steps[step.id]['input_steps'][conn.input_name] = {'source_step': conn.output_step_id,
'step_output': conn.output_name}
item['steps'] = steps
return item
def __encode_invocation_step( self, trans, invocation_step ):
return self.encode_all_ids(
trans,