Merge pull request #12899 from common-workflow-lab/type_updates2

Type updates from @mr-c - part 2

lib/galaxy/managers/collections.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Any, Dict, List, Union
 
 from sqlalchemy.orm import joinedload, Query
 
@@ -149,10 +150,16 @@ def create(self, trans, parent, name, collection_type, element_identifiers=None,
 
     def _create_instance_for_collection(self, trans, parent, name, dataset_collection, implicit_output_name=None, implicit_inputs=None, tags=None, set_hid=True, flush=True):
         if isinstance(parent, model.History):
-            dataset_collection_instance = model.HistoryDatasetCollectionAssociation(
+            dataset_collection_instance: Union[
+                model.HistoryDatasetCollectionAssociation,
+                model.LibraryDatasetCollectionAssociation,
+            ] = model.HistoryDatasetCollectionAssociation(
                 collection=dataset_collection,
                 name=name,
             )
+            assert isinstance(
+                dataset_collection_instance, model.HistoryDatasetCollectionAssociation
+            )
             if implicit_inputs:
                 for input_name, input_collection in implicit_inputs:
                     dataset_collection_instance.add_implicit_input_collection(input_name, input_collection)
@@ -549,7 +556,7 @@ def apply_rules(self, hdca, rule_set, handle_dataset):
     def _build_elements_from_rule_data(self, collection_type_description, rule_set, data, sources, handle_dataset):
         identifier_columns = rule_set.identifier_columns
         mapping_as_dict = rule_set.mapping_as_dict
-        elements = {}
+        elements: Dict[str, Any] = {}
         for data_index, row_data in enumerate(data):
             # For each row, find place in depth for this element.
             collection_type_at_depth = collection_type_description
@@ -592,18 +599,22 @@ def _build_elements_from_rule_data(self, collection_type_description, rule_set,
                         found = True
 
                     if not found:
-                        sub_collection = {}
+                        # Create a new collection whose elements are defined in the next loop
+                        sub_collection: Dict[str, Any] = {}
                         sub_collection["src"] = "new_collection"
                         sub_collection["collection_type"] = collection_type_at_depth.collection_type
                         sub_collection["elements"] = {}
+                        # Update elements with new collection
                         elements_at_depth[identifier] = sub_collection
+                        # Subsequent loop fills elements of newly created collection
                         elements_at_depth = sub_collection["elements"]
 
         return elements
 
     def __init_rule_data(self, elements, collection_type_description, parent_identifiers=None):
         parent_identifiers = parent_identifiers or []
-        data, sources = [], []
+        data: List[List[str]] = []
+        sources: List[Dict[str, str]] = []
         for element in elements:
             element_object = element.element_object
             identifiers = parent_identifiers + [element.element_identifier]

lib/galaxy/managers/collections_util.py

@@ -111,7 +111,7 @@ def dictify_dataset_collection_instance(dataset_collection_instance, parent, sec
         encoded_history_id = security.encode_id(parent.id)
         dict_value['url'] = web.url_for('history_content_typed', history_id=encoded_history_id, id=encoded_id, type="dataset_collection")
     elif isinstance(parent, model.LibraryFolder):
-        encoded_library_id = security.encode_id(parent.library.id)
+        encoded_library_id = security.encode_id(parent.library_root.id)
         encoded_folder_id = security.encode_id(parent.id)
         # TODO: Work in progress - this end-point is not right yet...
         dict_value['url'] = web.url_for('library_content', library_id=encoded_library_id, id=encoded_id, folder_id=encoded_folder_id)
@@ -163,10 +163,9 @@ def dictify_element_reference(element, rank_fuzzy_counts=None, recursive=True, s
             object_details["hda_ldda"] = 'hda'
             object_details["history_id"] = element_object.history_id
 
+        dictified["object"] = object_details
     else:
-        object_details = None
-
-    dictified["object"] = object_details
+        dictified["object"] = None
     return dictified
 
 

lib/galaxy/managers/tools.py

@@ -1,4 +1,5 @@
 import logging
+from typing import TYPE_CHECKING
 from uuid import uuid4
 
 from sqlalchemy import sql
@@ -11,6 +12,9 @@
 
 log = logging.getLogger(__name__)
 
+if TYPE_CHECKING:
+    from galaxy.managers.base import OrmFilterParsersType
+
 
 class DynamicToolManager(ModelManager):
     """ Manages dynamic tools stored in Galaxy's database.
@@ -103,6 +107,7 @@ def deactivate(self, dynamic_tool):
 
 
 class ToolFilterMixin:
+    orm_filter_parsers: "OrmFilterParsersType"
 
     def create_tool_filter(self, attr, op, val):