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)

lib/galaxy/tools/parameters/basic.py

@@ -53,7 +53,6 @@
     string_as_bool,
     string_as_bool_or_none,
     unicodify,
-    XML,
 )
 from galaxy.util.dictifiable import UsesDictVisibleKeys
 from galaxy.util.expressions import ExpressionContext
@@ -1619,54 +1618,24 @@ class DrillDownSelectToolParameter(SelectToolParameter):
     """
 
     def __init__(self, tool, input_source, context=None):
-        def recurse_option_elems(cur_options, option_elems):
-            for option_elem in option_elems:
-                selected = string_as_bool(option_elem.get("selected", False))
-                cur_options.append(
-                    {
-                        "name": option_elem.get("name"),
-                        "value": option_elem.get("value"),
-                        "options": [],
-                        "selected": selected,
-                    }
-                )
-                recurse_option_elems(cur_options[-1]["options"], option_elem.findall("option"))
-
         input_source = ensure_input_source(input_source)
         ToolParameter.__init__(self, tool, input_source)
-        # TODO: abstract XML out of here - so non-XML InputSources can
-        # specify DrillDown parameters.
-        elem = input_source.elem()
-        self.multiple = string_as_bool(elem.get("multiple", False))
-        self.display = elem.get("display", None)
-        self.hierarchy = elem.get("hierarchy", "exact")  # exact or recurse
-        self.separator = elem.get("separator", ",")
-        if from_file := elem.get("from_file", None):
-            if not os.path.isabs(from_file):
-                from_file = os.path.join(tool.app.config.tool_data_path, from_file)
-            elem = XML(f"<root>{open(from_file).read()}</root>")
-        self.dynamic_options = elem.get("dynamic_options", None)
-        if self.dynamic_options:
-            self.is_dynamic = True
-        self.options = []
-        self.filtered: Dict[str, Any] = {}
-        if elem.find("filter"):
+        self.multiple = input_source.get_bool("multiple", False)
+        self.display = input_source.get("display", None)
+        self.hierarchy = input_source.get("hierarchy", "exact")  # exact or recurse
+        self.separator = input_source.get("separator", ",")
+        tool_data_path = tool.app.config.tool_data_path
+        drill_down_dynamic_options = input_source.parse_drill_down_dynamic_options(tool_data_path)
+        if drill_down_dynamic_options is not None:
             self.is_dynamic = True
-            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 self.filtered:
-                        self.filtered[filter.get("data_ref")] = {}
-                    if filter.get("meta_key") not in self.filtered[filter.get("data_ref")]:
-                        self.filtered[filter.get("data_ref")][filter.get("meta_key")] = {}
-                    if filter.get("value") not in self.filtered[filter.get("data_ref")][filter.get("meta_key")]:
-                        self.filtered[filter.get("data_ref")][filter.get("meta_key")][filter.get("value")] = []
-                    recurse_option_elems(
-                        self.filtered[filter.get("data_ref")][filter.get("meta_key")][filter.get("value")],
-                        filter.find("options").findall("option"),
-                    )
-        elif not self.dynamic_options:
-            recurse_option_elems(self.options, elem.find("options").findall("option"))
+            self.dynamic_options = drill_down_dynamic_options.code_block
+            self.filtered = drill_down_dynamic_options.filters
+            self.options = []
+        else:
+            self.is_dynamic = False
+            self.dynamic_options = None
+            self.filtered = {}
+            self.options = input_source.parse_drill_down_static_options(tool_data_path)
 
     def _get_options_from_code(self, trans=None, other_values=None):
         assert self.dynamic_options, Exception("dynamic_options was not specifed")

test/functional/tools/parameters/gx_drill_down_exact.xml

@@ -0,0 +1,35 @@
+<tool id="gx_drill_down_exact" name="gx_drill_down_exact" version="1.0.0">
+    <macros>
+        <import>macros.xml</import>
+    </macros>
+    <command><![CDATA[
+echo "parameter: $parameter" > '$output'
+    ]]></command>
+    <inputs>
+        <!-- default drill_down, i.e. hierarchy="exact"
+            - select exacty the chosen values
+            - "inner" option nodes (non-leaves) are selectable
+        -->
+        <param name="parameter" type="drill_down" hierarchy="exact">
+            <expand macro="drill_down_static_options" />
+        </param>
+    </inputs>
+    <expand macro="simple_text_output" />
+    <tests>
+        <!-- select options from different levels of tree -->
+        <test>
+            <param name="parameter" value="a"/>
+            <expand macro="assert_output">
+                <has_line line="parameter: a"/>
+            </expand>
+        </test>
+        <test>
+            <param name="parameter" value="ab"/>
+            <expand macro="assert_output">
+                <has_line line="parameter: ab"/>
+            </expand>
+        </test>
+    </tests>
+    <help>
+    </help>
+</tool>