Parameter Model Improvements

lib/galaxy/tool_util/parameters/factory.py

@@ -1,8 +1,10 @@
 from typing import (
     Any,
+    cast,
     Dict,
     List,
     Optional,
+    Union,
 )
 
 from galaxy.tool_util.parser.cwl import CwlInputSource
@@ -27,7 +29,9 @@
     CwlStringParameterModel,
     CwlUnionParameterModel,
     DataCollectionParameterModel,
+    DataColumnParameterModel,
     DataParameterModel,
+    DrillDownParameterModel,
     FloatParameterModel,
     HiddenParameterModel,
     IntegerParameterModel,
@@ -129,14 +133,9 @@ def _from_input_source_galaxy(input_source: InputSource) -> ToolParameterT:
         elif param_type == "select":
             # Function... example in devteam cummeRbund.
             optional = input_source.parse_optional()
-            dynamic_options = input_source.get("dynamic_options", None)
             dynamic_options_config = input_source.parse_dynamic_options()
-            if dynamic_options_config:
-                dynamic_options_elem = dynamic_options.elem()
-            else:
-                dynamic_options_elem = None
+            is_static = dynamic_options_config is None
             multiple = input_source.get_bool("multiple", False)
-            is_static = dynamic_options is None and dynamic_options_elem is None
             options: Optional[List[LabelValue]] = None
             if is_static:
                 options = []
@@ -148,15 +147,40 @@ def _from_input_source_galaxy(input_source: InputSource) -> ToolParameterT:
                 options=options,
                 multiple=multiple,
             )
+        elif param_type == "drill_down":
+            multiple = input_source.get_bool("multiple", False)
+            hierarchy = input_source.get("hierarchy", "exact")
+            dynamic_options = input_source.parse_drill_down_dynamic_options()
+            static_options = None
+            if dynamic_options is None:
+                static_options = input_source.parse_drill_down_static_options()
+            return DrillDownParameterModel(
+                name=input_source.parse_name(),
+                multiple=multiple,
+                hierarchy=hierarchy,
+                options=static_options,
+            )
+        elif param_type == "data_column":
+            return DataColumnParameterModel(
+                name=input_source.parse_name(),
+            )
         else:
             raise Exception(f"Unknown Galaxy parameter type {param_type}")
     elif input_type == "conditional":
         test_param_input_source = input_source.parse_test_input_source()
-        test_parameter = _from_input_source_galaxy(test_param_input_source)
+        test_parameter = cast(
+            Union[BooleanParameterModel, SelectParameterModel], _from_input_source_galaxy(test_param_input_source)
+        )
         whens = []
         default_value = object()
         if isinstance(test_parameter, BooleanParameterModel):
             default_value = test_parameter.value
+        elif isinstance(test_parameter, SelectParameterModel):
+            select_parameter = cast(SelectParameterModel, test_parameter)
+            select_default_value = select_parameter.default_value
+            if select_default_value is not None:
+                default_value = select_default_value
+
         # TODO: handle select parameter model...
         for value, case_inputs_sources in input_source.parse_when_input_sources():
             if isinstance(test_parameter, BooleanParameterModel):

lib/galaxy/tool_util/parameters/models.py

@@ -37,6 +37,7 @@
 )
 
 from galaxy.exceptions import RequestParameterInvalidException
+from galaxy.tool_util.parser.interface import DrillDownOptionsDict
 from ._types import (
     cast_as_type,
     is_optional,
@@ -478,9 +479,112 @@ def pydantic_template(self, state_representation: StateRepresentationT) -> Dynam
     def has_selected_static_option(self):
         return self.options is not None and any(o.selected for o in self.options)
 
+    @property
+    def default_value(self) -> Optional[str]:
+        if self.options:
+            for option in self.options:
+                if option.selected:
+                    return option.value
+            # single value pick up first value
+            if not self.optional:
+                return self.options[0].value
+
+        return None
+
+    @property
+    def request_requires_value(self) -> bool:
+        # API will allow an empty value and just grab the first static option
+        # see API Tests -> test_tools.py -> test_select_first_by_default
+        # so only require a value in the multiple case if optional is False
+        return self.multiple and not self.optional
+
+
+DrillDownHierarchyT = Literal["recurse", "exact"]
+
+
+def drill_down_possible_values(options: List[DrillDownOptionsDict], multiple: bool) -> List[str]:
+    possible_values = []
+
+    def add_value(option: str, is_leaf: bool):
+        if not multiple and not is_leaf:
+            return
+        possible_values.append(option)
+
+    def walk_selection(option: DrillDownOptionsDict):
+        child_options = option["options"]
+        is_leaf = not child_options
+        add_value(option["value"], is_leaf)
+        if not is_leaf:
+            for child_option in child_options:
+                walk_selection(child_option)
+
+    for option in options:
+        walk_selection(option)
+
+    return possible_values
+
+
+class DrillDownParameterModel(BaseGalaxyToolParameterModelDefinition):
+    parameter_type: Literal["gx_drill_down"] = "gx_drill_down"
+    options: Optional[List[DrillDownOptionsDict]] = None
+    multiple: bool
+    hierarchy: DrillDownHierarchyT
+
+    @property
+    def py_type(self) -> Type:
+        if self.options is not None:
+            literal_options: List[Type] = [
+                cast_as_type(Literal[o]) for o in drill_down_possible_values(self.options, self.multiple)
+            ]
+            py_type = union_type(literal_options)
+        else:
+            py_type = StrictStr
+
+        if self.multiple:
+            py_type = list_type(py_type)
+
+        return py_type
+
+    def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation:
+        return dynamic_model_information_from_py_type(self, self.py_type)
+
     @property
     def request_requires_value(self) -> bool:
-        return not self.optional and not self.has_selected_static_option
+        options = self.options
+        if options:
+            # if any of these are selected, they seem to serve as defaults - check out test_tools -> test_drill_down_first_by_default
+            return not any_drill_down_options_selected(options)
+        else:
+            # I'm not sure how to handle dynamic options... they might or might not be required?
+            # do we need to default to assuming they're not required?
+            return False
+
+
+def any_drill_down_options_selected(options: List[DrillDownOptionsDict]) -> bool:
+    for option in options:
+        selected = option.get("selected")
+        if selected:
+            return True
+        child_options = option.get("options", [])
+        if any_drill_down_options_selected(child_options):
+            return True
+
+    return False
+
+
+class DataColumnParameterModel(BaseGalaxyToolParameterModelDefinition):
+    parameter_type: Literal["gx_data_column"] = "gx_data_column"
+
+    @property
+    def py_type(self) -> Type:
+        return StrictInt
+
+    def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation:
+        return dynamic_model_information_from_py_type(self, self.py_type)
+
+    @property
+    def request_requires_value(self) -> bool:
+        return False
 
 
 DiscriminatorType = Union[bool, str]
@@ -586,18 +690,31 @@ def pydantic_template(self, state_representation: StateRepresentationT) -> Dynam
             self.parameters, f"Repeat_{self.name}", state_representation
         )
 
+        initialize_repeat: Any
+        if self.request_requires_value:
+            initialize_repeat = ...
+        else:
+            initialize_repeat = None
+
         class RepeatType(RootModel):
-            root: List[instance_class] = Field(..., min_length=self.min, max_length=self.max)  # type: ignore[valid-type]
+            root: List[instance_class] = Field(initialize_repeat, min_length=self.min, max_length=self.max)  # type: ignore[valid-type]
 
         return DynamicModelInformation(
             self.name,
-            (RepeatType, ...),
+            (RepeatType, initialize_repeat),
             {},
         )
 
     @property
     def request_requires_value(self) -> bool:
-        return True  # TODO:
+        if self.min is None or self.min == 0:
+            return False
+        # so we know we need at least one value, but maybe none of the parameters in the list
+        # are required
+        for parameter in self.parameters:
+            if parameter.request_requires_value:
+                return True
+        return False
 
 
 class SectionParameterModel(BaseGalaxyToolParameterModelDefinition):
@@ -799,8 +916,10 @@ def request_requires_value(self) -> bool:
     SelectParameterModel,
     DataParameterModel,
     DataCollectionParameterModel,
+    DataColumnParameterModel,
     DirectoryUriParameterModel,
     RulesParameterModel,
+    DrillDownParameterModel,
     ColorParameterModel,
     ConditionalParameterModel,
     RepeatParameterModel,

lib/galaxy/tool_util/parser/interface.py

@@ -424,6 +424,20 @@ def get_index_file_name(self) -> Optional[str]:
         """If dynamic options are loaded from an index file, return the name."""
 
 
+DrillDownDynamicFilters = Dict[str, Dict[str, dict]]  # {input key: {metadata_key: metadata values}}
+
+
+class DrillDownDynamicOptions(metaclass=ABCMeta):
+
+    @abstractmethod
+    def from_code_block(self) -> Optional[str]:
+        """Get a code block to do an eval on."""
+
+    @abstractmethod
+    def from_filters(self) -> Optional[DrillDownDynamicFilters]:
+        """Get filters to apply to target datasets."""
+
+
 class InputSource(metaclass=ABCMeta):
     default_optional = False
 
@@ -491,12 +505,22 @@ def parse_dynamic_options(self) -> Optional[DynamicOptions]:
         """
         return None
 
+    def parse_drill_down_dynamic_options(
+        self, tool_data_path: Optional[str] = None
+    ) -> Optional["DrillDownDynamicOptions"]:
+        return None
+
     def parse_static_options(self) -> List[Tuple[str, str, bool]]:
         """Return list of static options if this is a select type without
         defining a dynamic options.
         """
         return []
 
+    def parse_drill_down_static_options(
+        self, tool_data_path: Optional[str] = None
+    ) -> Optional[List["DrillDownOptionsDict"]]:
+        return None
+
     def parse_conversion_tuples(self):
         """Return list of (name, extension) to describe explicit conversions."""
         return []
@@ -673,3 +697,10 @@ def from_dict(as_dict):
 
     def to_dict(self):
         return dict(name=self.name, attributes=self.attrib, element_tests=self.element_tests)
+
+
+class DrillDownOptionsDict(TypedDict):
+    name: Optional[str]
+    value: str
+    options: List["DrillDownOptionsDict"]
+    selected: bool