Strong typing around HDAs manager.

lib/galaxy/managers/base.py

@@ -42,6 +42,7 @@ from typing import (
 
 import sqlalchemy
 from sqlalchemy.orm.scoping import scoped_session
+from typing_extensions import Protocol
 
 from galaxy import exceptions
 from galaxy import model
@@ -727,6 +728,12 @@ class ModelSerializer(HasAModelManager):
         return self.views[view][:]
 
 
+class Deserializer(Protocol):
+
+    def __call__(self, item: Any, key: Any, val: Any, **kwargs) -> Any:
+        ...
+
+
 class ModelDeserializer(HasAModelManager):
     """
     An object that converts an incoming serialized dict into values that can be
@@ -741,7 +748,7 @@ class ModelDeserializer(HasAModelManager):
         super().__init__(app, **kwargs)
         self.app = app
 
-        self.deserializers: Dict[str, Callable] = {}
+        self.deserializers: Dict[str, Deserializer] = {}
         self.deserializable_keyset: Set[str] = set()
         self.add_deserializers()
         # a sub object that can validate incoming values

lib/galaxy/managers/hdas.py

@@ -7,6 +7,7 @@ history.
 import gettext
 import logging
 import os
+from typing import Any, Dict, List
 
 from sqlalchemy.orm.session import object_session
 
@@ -22,7 +23,7 @@ from galaxy.managers import (
     taggable,
     users,
 )
-from galaxy.structured_app import MinimalManagerApp
+from galaxy.structured_app import MinimalManagerApp, StructuredApp
 
 log = logging.getLogger(__name__)
 
@@ -58,7 +59,7 @@ class HDAManager(datasets.DatasetAssociationManager,
     def get_owned_ids(self, object_ids, history=None):
         """Get owned IDs.
         """
-        filters = [self.model_class.id.in_(object_ids), self.model_class.history_id == history.id]
+        filters = [self.model_class.table.c.id.in_(object_ids), self.model_class.table.c.history_id == history.id]
         return self.list(filters=filters)
 
     # .... security and permissions
@@ -260,8 +261,9 @@ class HDASerializer(  # datasets._UnflattenedMetadataDatasetAssociationSerialize
         taggable.TaggableSerializerMixin,
         annotatable.AnnotatableSerializerMixin):
     model_manager_class = HDAManager
+    app: StructuredApp
 
-    def __init__(self, app: MinimalManagerApp):
+    def __init__(self, app: StructuredApp):
         super().__init__(app)
         self.hda_manager = self.manager
 
@@ -446,7 +448,7 @@ class HDASerializer(  # datasets._UnflattenedMetadataDatasetAssociationSerialize
         """
         Return dictionary containing new-style display app urls.
         """
-        display_apps = []
+        display_apps: List[Dict[str, Any]] = []
         for display_app in hda.get_display_applications(trans).values():
 
             app_links = []
@@ -465,7 +467,7 @@ class HDASerializer(  # datasets._UnflattenedMetadataDatasetAssociationSerialize
         """
         Return dictionary containing old-style display app urls.
         """
-        display_apps = []
+        display_apps: List[Dict[str, Any]] = []
         if not self.app.config.enable_old_display_applications:
             return display_apps
 
@@ -541,9 +543,9 @@ class HDADeserializer(datasets.DatasetAssociationDeserializer,
         self.deserializers.update({
             'visible': self.deserialize_bool,
             # remapped
-            'genome_build': lambda i, k, v, **c: self.deserialize_genome_build(i, 'dbkey', v),
-            'misc_info': lambda i, k, v, **c: self.deserialize_basestring(i, 'info', v,
-                                                                          convert_none_to_empty=True),
+            'genome_build': lambda item, key, val, **c: self.deserialize_genome_build(item, 'dbkey', val),
+            'misc_info': lambda item, key, val, **c: self.deserialize_basestring(item, 'info', val,
+                                                                                 convert_none_to_empty=True),
         })
         self.deserializable_keyset.update(self.deserializers.keys())
 

lib/galaxy/managers/hdcas.py

@@ -17,7 +17,7 @@ from galaxy.managers import (
     taggable
 )
 from galaxy.managers.collections_util import get_hda_and_element_identifiers
-from galaxy.structured_app import MinimalManagerApp
+from galaxy.structured_app import StructuredApp
 from galaxy.util.zipstream import ZipstreamWrapper
 
 
@@ -95,7 +95,7 @@ class DCESerializer(base.ModelSerializer):
     Serializer for DatasetCollectionElements.
     """
 
-    def __init__(self, app: MinimalManagerApp):
+    def __init__(self, app: StructuredApp):
         super().__init__(app)
         self.hda_serializer = hdas.HDASerializer(app)
         self.dc_serializer = DCSerializer(app, dce_serializer=self)
@@ -129,7 +129,7 @@ class DCSerializer(base.ModelSerializer):
     Serializer for DatasetCollections.
     """
 
-    def __init__(self, app: MinimalManagerApp, dce_serializer=None):
+    def __init__(self, app: StructuredApp, dce_serializer=None):
         super().__init__(app)
         self.dce_serializer = dce_serializer or DCESerializer(app)
 
@@ -167,8 +167,9 @@ class DCASerializer(base.ModelSerializer):
     """
     Base (abstract) Serializer class for HDCAs and LDCAs.
     """
+    app: StructuredApp
 
-    def __init__(self, app: MinimalManagerApp, dce_serializer=None):
+    def __init__(self, app: StructuredApp, dce_serializer=None):
         super().__init__(app)
         self.dce_serializer = dce_serializer or DCESerializer(app)
 
@@ -223,7 +224,7 @@ class HDCASerializer(
     Serializer for HistoryDatasetCollectionAssociations.
     """
 
-    def __init__(self, app: MinimalManagerApp):
+    def __init__(self, app: StructuredApp):
         super().__init__(app)
         self.hdca_manager = HDCAManager(app)
 

lib/galaxy/structured_app.py

@@ -123,3 +123,4 @@ class StructuredApp(MinimalManagerApp):
     job_manager: Any  # galaxy.jobs.manager.JobManager
     user_manager: Any
     api_keys_manager: Any  # 'galaxy.managers.api_keys.ApiKeyManager'
+    visualizations_registry: Any  # 'galaxy.visualization.plugins.registry.VisualizationsRegistry'

setup.cfg

@@ -593,8 +593,6 @@ check_untyped_defs = False
 check_untyped_defs = False
 [mypy-galaxy.tools.error_reports]
 check_untyped_defs = False
-[mypy-galaxy.managers.hdas]
-check_untyped_defs = False
 [mypy-galaxy.jobs.dynamic_tool_destination]
 check_untyped_defs = False
 [mypy-galaxy.jobs.runners.state_handler_factory]