Typing of tool request expansion stuff.

lib/galaxy/model/dataset_collections/subcollections.py

@@ -1,17 +1,33 @@
+from typing import (
+    List,
+    TYPE_CHECKING,
+)
+
 from galaxy import exceptions
 
+if TYPE_CHECKING:
+    from galaxy.model import (
+        DatasetCollection,
+        DatasetCollectionElement,
+        HistoryDatasetCollectionAssociation,
+    )
+
 
-def split_dataset_collection_instance(dataset_collection_instance, collection_type):
+def split_dataset_collection_instance(
+    dataset_collection_instance: "HistoryDatasetCollectionAssociation", collection_type: str
+) -> List["DatasetCollectionElement"]:
     """Split up collection into collection."""
     return _split_dataset_collection(dataset_collection_instance.collection, collection_type)
 
 
-def _split_dataset_collection(dataset_collection, collection_type):
+def _split_dataset_collection(
+    dataset_collection: DatasetCollection, collection_type: str
+) -> List["DatasetCollectionElement"]:
     this_collection_type = dataset_collection.collection_type
     if not this_collection_type.endswith(collection_type) or this_collection_type == collection_type:
         raise exceptions.MessageException("Cannot split collection in desired fashion.")
 
-    split_elements = []
+    split_elements: List["DatasetCollectionElement"] = []
     for element in dataset_collection.elements:
         child_collection = element.child_collection
         if child_collection is None:

lib/galaxy/tools/parameters/meta.py

@@ -8,13 +8,18 @@
     List,
     Optional,
     Tuple,
+    Union,
 )
 
 from galaxy import (
     exceptions,
     util,
 )
-from galaxy.model import HistoryDatasetCollectionAssociation
+from galaxy.model import (
+    DatasetCollectionElement,
+    DatasetInstance,
+    HistoryDatasetCollectionAssociation,
+)
 from galaxy.model.dataset_collections import (
     matching,
     subcollections,
@@ -327,7 +332,12 @@ def visitor(input, value, prefix, prefixed_name, prefixed_label, error, **kwargs
     return (single_inputs_nested, matched_multi_inputs, multiplied_multi_inputs)
 
 
-def __expand_collection_parameter(trans, input_key, incoming_val, collections_to_match, linked=False):
+CollectionExpansionListT = Union[List[DatasetCollectionElement], List[DatasetInstance]]
+
+
+def __expand_collection_parameter(
+    trans, input_key: str, incoming_val, collections_to_match: "matching.CollectionsToMatch", linked=False
+) -> CollectionExpansionListT:
     # If subcollectin multirun of data_collection param - value will
     # be "hdca_id|subcollection_type" else it will just be hdca_id
     if "|" in incoming_val:
@@ -348,18 +358,20 @@ def __expand_collection_parameter(trans, input_key, incoming_val, collections_to
         raise exceptions.ToolInputsNotReadyException("An input collection is not populated.")
     collections_to_match.add(input_key, hdc, subcollection_type=subcollection_type, linked=linked)
     if subcollection_type is not None:
-        subcollection_elements = subcollections.split_dataset_collection_instance(hdc, subcollection_type)
+        subcollection_elements: List[DatasetCollectionElement] = subcollections.split_dataset_collection_instance(
+            hdc, subcollection_type
+        )
         return subcollection_elements
     else:
-        hdas = []
+        hdas: List[DatasetInstance] = []
         for element in hdc.collection.dataset_elements:
             hda = element.dataset_instance
             hda.element_identifier = element.element_identifier
             hdas.append(hda)
         return hdas
 
 
-def __collection_multirun_parameter(value):
+def __collection_multirun_parameter(value: Dict[str, Any]) -> bool:
     is_batch = value.get("batch", False)
     if not is_batch:
         return False