Merge pull request #21171 from nsoranzo/type_annot_job_handling

Add type annotations to job handling code

Author: nsoranzo

lib/galaxy/datatypes/registry.py

@@ -45,6 +45,7 @@
 if TYPE_CHECKING:
     from galaxy.datatypes.data import Data
     from galaxy.tool_util.toolbox.base import AbstractToolBox
+    from galaxy.tools import SetMetadataTool
 
 
 class ConfigurationError(Exception):
@@ -749,15 +750,15 @@ def reload_display_applications(self, display_application_ids=None):
                 failed.append(display_application_id)
         return (reloaded, failed)
 
-    def load_external_metadata_tool(self, toolbox):
+    def load_external_metadata_tool(self, toolbox: "AbstractToolBox") -> None:
         """Adds a tool which is used to set external metadata"""
         # We need to be able to add a job to the queue to set metadata. The queue will currently only accept jobs with an associated
         # tool.  We'll load a special tool to be used for Auto-Detecting metadata; this is less than ideal, but effective
         # Properly building a tool without relying on parsing an XML file is near difficult...so we bundle with Galaxy.
         set_meta_tool = toolbox.load_hidden_lib_tool(
             os.path.abspath(os.path.join(os.path.dirname(__file__), "set_metadata_tool.xml"))
         )
-        self.set_external_metadata_tool = set_meta_tool
+        self.set_external_metadata_tool = cast("SetMetadataTool", set_meta_tool)
         self.log.debug("Loaded external metadata tool: %s", self.set_external_metadata_tool.id)
 
     def set_default_values(self):

lib/galaxy/job_metrics/__init__.py

@@ -26,6 +26,7 @@
     NamedTuple,
     Optional,
     TYPE_CHECKING,
+    Union,
 )
 
 from galaxy import util
@@ -38,6 +39,7 @@
 
 if TYPE_CHECKING:
     from galaxy.job_metrics.instrumenters import InstrumentPlugin
+    from galaxy.util import Element
 
 log = logging.getLogger(__name__)
 
@@ -123,21 +125,23 @@ def raw_to_dictifiable(raw_metric: RawMetric) -> DictifiableMetric:
         metrics = map(raw_to_dictifiable, raw_metrics)
         return [m for m in metrics if m.safety.value >= allowed_safety.value]
 
-    def set_destination_conf_file(self, destination_id, conf_file):
+    def set_destination_conf_file(self, destination_id: str, conf_file: str) -> None:
         instrumenter = JobInstrumenter.from_file(self.plugin_classes, conf_file)
         self.set_destination_instrumenter(destination_id, instrumenter)
 
-    def set_destination_conf_element(self, destination_id, element):
+    def set_destination_conf_element(self, destination_id: str, element: "Element") -> None:
         plugin_source = plugin_config.PluginConfigSource("xml", element)
         instrumenter = JobInstrumenter(self.plugin_classes, plugin_source)
         self.set_destination_instrumenter(destination_id, instrumenter)
 
-    def set_destination_conf_dicts(self, destination_id, conf_dicts):
+    def set_destination_conf_dicts(self, destination_id: str, conf_dicts: List[Dict[str, Any]]) -> None:
         plugin_source = plugin_config.PluginConfigSource("dict", conf_dicts)
         instrumenter = JobInstrumenter(self.plugin_classes, plugin_source)
         self.set_destination_instrumenter(destination_id, instrumenter)
 
-    def set_destination_instrumenter(self, destination_id, job_instrumenter=None):
+    def set_destination_instrumenter(
+        self, destination_id: str, job_instrumenter: Union["JobInstrumenterI", None] = None
+    ) -> None:
         if job_instrumenter is None:
             job_instrumenter = NULL_JOB_INSTRUMENTER
         self.job_instrumenters[destination_id] = job_instrumenter