diff --git a/lib/galaxy/model/dataset_collections/subcollections.py b/lib/galaxy/model/dataset_collections/subcollections.py index af6c2a397326..aa48c546f304 100644 --- a/lib/galaxy/model/dataset_collections/subcollections.py +++ b/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: diff --git a/lib/galaxy/tools/parameters/meta.py b/lib/galaxy/tools/parameters/meta.py index b74df54fa269..3427167812d5 100644 --- a/lib/galaxy/tools/parameters/meta.py +++ b/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,10 +358,12 @@ 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 @@ -359,7 +371,7 @@ def __expand_collection_parameter(trans, input_key, incoming_val, collections_to 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