Merge pull request #10657 from jmchilton/workflow_step_type

Refactor logic for workflow_step.input_type into model.

lib/galaxy/model/__init__.py

@@ -4916,6 +4916,11 @@ class WorkflowStep(RepresentById):
 
     See :class:`galaxy.model.WorkflowStepInput` and :class:`galaxy.model.WorkflowStepConnection` for more information.
     """
+    STEP_TYPE_TO_INPUT_TYPE = {
+        "data_input": "dataset",
+        "data_collection_input": "dataset_collection",
+        "parameter_input": "parameter",
+    }
 
     def __init__(self):
         self.id = None
@@ -4936,6 +4941,11 @@ class WorkflowStep(RepresentById):
     def tool_uuid(self):
         return self.dynamic_tool and self.dynamic_tool.uuid
 
+    @property
+    def input_type(self):
+        assert self.type and self.type in self.STEP_TYPE_TO_INPUT_TYPE, "step.input_type can only be called on input step types"
+        return self.STEP_TYPE_TO_INPUT_TYPE[self.type]
+
     @property
     def input_default_value(self):
         tool_inputs = self.tool_inputs

lib/galaxy/workflow/modules.py

@@ -429,11 +429,6 @@ class SubWorkflowModule(WorkflowModule):
     def get_all_inputs(self, data_only=False, connectable_only=False):
         """ Get configure time data input descriptions. """
         # Filter subworkflow steps and get inputs
-        step_to_input_type = {
-            "data_input": "dataset",
-            "data_collection_input": "dataset_collection",
-            "parameter_input": "parameter",
-        }
         inputs = []
         if hasattr(self.subworkflow, 'input_steps'):
             for step in self.subworkflow.input_steps:
@@ -441,17 +436,16 @@ class SubWorkflowModule(WorkflowModule):
                 if not name:
                     step_module = module_factory.from_workflow_step(self.trans, step)
                     name = f"{step.order_index}:{step_module.get_name()}"
-                step_type = step.type
-                assert step_type in step_to_input_type
                 input = dict(
                     input_subworkflow_step_id=step.order_index,
                     name=name,
                     label=name,
                     multiple=False,
                     extensions=["data"],
-                    input_type=step_to_input_type[step_type],
+                    input_type=step.input_type,
                 )
-                if step.type == 'data_collection_input':
+                step_type = step.type
+                if step_type == 'data_collection_input':
                     input['collection_type'] = step.tool_inputs.get('collection_type') if step.tool_inputs else None
                 if step_type == 'parameter_input':
                     input['type'] = step.tool_inputs['parameter_type']