Richer typing / models / parsing for drill down parameters.

lib/galaxy/tool_util/parameters/factory.py

@@ -30,6 +30,7 @@
     CwlUnionParameterModel,
     DataCollectionParameterModel,
     DataParameterModel,
+    DrillDownParameterModel,
     FloatParameterModel,
     HiddenParameterModel,
     IntegerParameterModel,
@@ -145,6 +146,19 @@ 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,
+            )
         else:
             raise Exception(f"Unknown Galaxy parameter type {param_type}")
     elif input_type == "conditional":

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,
@@ -498,6 +499,61 @@ def request_requires_value(self) -> bool:
         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["drill_down"] = "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:
+        # TODO: retest this logic in the API for drill downs
+        return self.multiple and not self.optional
+
+
 DiscriminatorType = Union[bool, str]
 
 
@@ -829,6 +885,7 @@ def request_requires_value(self) -> bool:
     DataCollectionParameterModel,
     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,20 @@ 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 +695,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

lib/galaxy/tool_util/parser/xml.py

@@ -30,12 +30,16 @@
     Element,
     ElementTree,
     string_as_bool,
+    XML,
     xml_text,
     xml_to_string,
 )
 from .interface import (
     AssertionList,
     Citation,
+    DrillDownDynamicFilters,
+    DrillDownDynamicOptions,
+    DrillDownOptionsDict,
     DynamicOptions,
     InputSource,
     PageSource,
@@ -1321,6 +1325,64 @@ def parse_static_options(self) -> List[Tuple[str, str, bool]]:
             deduplicated_static_options[value] = (text, value, selected)
         return list(deduplicated_static_options.values())
 
+    def parse_drill_down_dynamic_options(
+        self, tool_data_path: Optional[str] = None
+    ) -> Optional[DrillDownDynamicOptions]:
+        from_file = self.input_elem.get("from_file", None)
+        if from_file:
+            if not os.path.isabs(from_file):
+                assert tool_data_path, "This tool cannot be parsed outside of a Galaxy context"
+                from_file = os.path.join(tool_data_path, from_file)
+            elem = XML(f"<root>{open(from_file).read()}</root>")
+        else:
+            elem = self.input_elem
+
+        dynamic_options_raw = elem.get("dynamic_options", None)
+        dynamic_options: Optional[str] = str(dynamic_options_raw) if dynamic_options_raw else None
+        filters: Optional[DrillDownDynamicFilters] = None
+        if elem.find("filter"):
+            _filters: DrillDownDynamicFilters = {}
+            for filter in elem.findall("filter"):
+                # currently only filtering by metadata key matching input file is allowed
+                if filter.get("type") == "data_meta":
+                    if filter.get("data_ref") not in filters:
+                        filters[filter.get("data_ref")] = {}
+                    if filter.get("meta_key") not in filters[filter.get("data_ref")]:
+                        filters[filter.get("data_ref")][filter.get("meta_key")] = {}
+                    if filter.get("value") not in filters[filter.get("data_ref")][filter.get("meta_key")]:
+                        filters[filter.get("data_ref")][filter.get("meta_key")][filter.get("value")] = []
+                    _recurse_drill_down_elems(
+                        filters[filter.get("data_ref")][filter.get("meta_key")][filter.get("value")],
+                        filter.find("options").findall("option"),
+                    )
+            filters = _filters
+        if filters is None and dynamic_options is None:
+            return None
+        else:
+            return XmlDrillDownDynamicOptions(
+                code_block=dynamic_options,
+                filters=filters,
+            )
+
+    def parse_drill_down_static_options(self, tool_data_path: Optional[str] = None) -> Optional[List[DrillDownOptionsDict]]:
+        from_file = self.input_elem.get("from_file", None)
+        if from_file:
+            if not os.path.isabs(from_file):
+                assert tool_data_path, "This tool cannot be parsed outside of a Galaxy context"
+                from_file = os.path.join(tool_data_path, from_file)
+            elem = XML(f"<root>{open(from_file).read()}</root>")
+        else:
+            elem = self.input_elem
+
+        dynamic_options_elem = elem.get("dynamic_options", None)
+        filter_elem = elem.get("filter", None)
+        if dynamic_options_elem is not None and filter_elem is not None:
+            return None
+
+        root_options: List[DrillDownOptionsDict] = []
+        _recurse_drill_down_elems(root_options, elem.find("options").findall("option"))
+        return root_options
+
     def parse_optional(self, default=None):
         """Return boolean indicating whether parameter is optional."""
         elem = self.input_elem
@@ -1450,3 +1512,33 @@ def parse_citation_elem(citation_elem: Element) -> Optional[Citation]:
         type=citation_type,
         content=content,
     )
+
+
+class XmlDrillDownDynamicOptions(DrillDownDynamicOptions):
+
+    def __init__(self, code_block: Optional[str], filters: Optional[DrillDownDynamicFilters]):
+        self._code_block = code_block
+        self._filters = filters
+
+    def from_code_block(self) -> Optional[str]:
+        """Get a code block to do an eval on."""
+        return self._code_block
+
+    def from_filters(self) -> Optional[DrillDownDynamicFilters]:
+        return self._filters
+
+
+def _recurse_drill_down_elems(options: List[DrillDownOptionsDict], option_elems: List[Element]):
+    for option_elem in option_elems:
+        selected = string_as_bool(option_elem.get("selected", False))
+        nested_options = []
+        current_option: DrillDownOptionsDict = DrillDownOptionsDict(
+            {
+                "name": option_elem.get("name"),
+                "value": option_elem.get("value"),
+                "options": nested_options,
+                "selected": selected,
+            }
+        )
+        _recurse_drill_down_elems(nested_options, option_elem.findall("option"))
+        options.append(current_option)