Cleanup typing - rebase into paired/unpaired...

lib/galaxy/tool_util/parser/interface.py

@@ -596,7 +596,7 @@ class TestCollectionDefElementInternal(TypedDict):
 class XmlTestCollectionDefDict(TypedDict):
     model_class: Literal["TestCollectionDef"]
     attributes: TestCollectionAttributeDict
-    collection_type: CollectionType
+    collection_type: Optional[CollectionType]
     fields: Optional[List[FieldDict]]
     elements: List[TestCollectionDefElementDict]
     name: str

lib/galaxy/tools/parameters/basic.py

@@ -2020,15 +2020,23 @@ def do_validate(v):
                 raise ValueError(f"At most {self.max} datasets are required for {self.name}")
 
 
-def src_id_to_item(
-    sa_session: "Session", value: typing.MutableMapping[str, Any], security: "IdEncodingHelper"
-) -> Union[
+ItemFromSrcAny = Union[
     DatasetCollectionElement,
     HistoryDatasetAssociation,
     HistoryDatasetCollectionAssociation,
     LibraryDatasetDatasetAssociation,
     CollectionAdapter,
-]:
+]
+ItemFromSrcCollection = Union[
+    DatasetCollectionElement,
+    HistoryDatasetCollectionAssociation,
+    CollectionAdapter,
+]
+
+
+def src_id_to_item(
+    sa_session: "Session", value: typing.MutableMapping[str, Any], security: "IdEncodingHelper"
+) -> ItemFromSrcAny:
     adapter_model = None
     if value["src"] == "CollectionAdapter":
         adapter_model = validate_collection_adapter_src_dict(value)
@@ -2064,6 +2072,15 @@ def src_id_to_item(
     return item
 
 
+def src_id_to_item_collection(
+    sa_session: "Session", value: typing.MutableMapping[str, Any], security: "IdEncodingHelper"
+) -> ItemFromSrcCollection:
+    rval = src_id_to_item(sa_session, value, security)
+    if isinstance(rval, (LibraryDatasetDatasetAssociation, HistoryDatasetAssociation)):
+        raise ValueError("Expected to find collection, but got single dataset wrapper")
+    return cast(ItemFromSrcCollection, rval)
+
+
 class DataToolParameter(BaseDataToolParameter):
     # TODO, Nate: Make sure the following unit tests appropriately test the dataset security
     # components.  Add as many additional tests as necessary.
@@ -2491,7 +2508,7 @@ def from_json(self, value, trans, other_values=None):
         session = trans.sa_session
 
         other_values = other_values or {}
-        rval: Optional[Union[DatasetCollectionElement, HistoryDatasetCollectionAssociation, CollectionAdapter]] = None
+        rval: Optional[ItemFromSrcCollection] = None
         if trans.workflow_building_mode is workflow_building_modes.ENABLED:
             return None
         if not value and not self.optional and not self.default_object:

lib/galaxy/tools/wrappers.py

@@ -782,7 +782,7 @@ def has_single_item(self) -> bool:
         return self.__input_supplied and len(self.__element_instance_list) == 1
 
     @property
-    def single_item(self) -> "DatasetCollectionElementWrapper":
+    def single_item(self) -> Optional["DatasetCollectionElementWrapper"]:
         return self[0]
 
     @property