Add type annotations

lib/galaxy/managers/workflows.py

@@ -46,7 +46,10 @@
     joinedload,
     subqueryload,
 )
-from typing_extensions import Annotated
+from typing_extensions import (
+    Annotated,
+    TypeAlias,
+)
 
 from galaxy import (
     exceptions,
@@ -1827,7 +1830,7 @@ def __module_from_dict(
             self.add_item_annotation(sa_session, trans.get_user(), step, annotation)
 
         # Stick this in the step temporarily
-        DictConnection = Dict[str, Union[int, str]]
+        DictConnection: TypeAlias = Dict[str, Union[int, str]]
         temp_input_connections: Dict[str, Union[List[DictConnection], DictConnection]] = step_dict.get(
             "input_connections", {}
         )

lib/galaxy/tool_util/cwl/parser.py

@@ -15,6 +15,7 @@
 from typing import (
     Dict,
     List,
+    Optional,
     overload,
     Union,
 )
@@ -531,7 +532,7 @@ class WorkflowProxy:
     def __init__(self, workflow, workflow_path=None):
         self._workflow = workflow
         self._workflow_path = workflow_path
-        self._step_proxies = None
+        self._step_proxies: Optional[List[Union[SubworkflowStepProxy, ToolStepProxy]]] = None
 
     @property
     def cwl_id(self):
@@ -562,7 +563,7 @@ def get_outputs_for_label(self, label):
 
     def tool_reference_proxies(self):
         """Fetch tool source definitions for all referenced tools."""
-        references = []
+        references: List[ToolProxy] = []
         for step in self.step_proxies():
             references.extend(step.tool_reference_proxies())
         return references

lib/galaxy/tools/__init__.py

@@ -458,10 +458,10 @@ class ToolBox(AbstractToolBox):
     dependency management, etc.
     """
 
-    def __init__(self, config_filenames, tool_root_dir, app, save_integrated_tool_panel=True):
+    def __init__(self, config_filenames, tool_root_dir, app, save_integrated_tool_panel: bool = True):
         self._reload_count = 0
         self.tool_location_fetcher = ToolLocationFetcher()
-        self.cache_regions = {}
+        self.cache_regions: Dict[str, ToolDocumentCache] = {}
         # This is here to deal with the old default value, which doesn't make
         # sense in an "installed Galaxy" world.
         # FIXME: ./
@@ -517,7 +517,7 @@ def load_builtin_converters(self):
             tool.hidden = False
             section.elems.append_tool(tool)
 
-    def persist_cache(self, register_postfork=False):
+    def persist_cache(self, register_postfork: bool = False):
         """
         Persists any modified tool cache files to disk.
 
@@ -561,13 +561,13 @@ def tools_by_id(self):
         # Deprecated method, TODO - eliminate calls to this in test/.
         return self._tools_by_id
 
-    def get_cache_region(self, tool_cache_data_dir):
+    def get_cache_region(self, tool_cache_data_dir: Optional[str]):
         if self.app.config.enable_tool_document_cache and tool_cache_data_dir:
             if tool_cache_data_dir not in self.cache_regions:
                 self.cache_regions[tool_cache_data_dir] = ToolDocumentCache(cache_dir=tool_cache_data_dir)
             return self.cache_regions[tool_cache_data_dir]
 
-    def create_tool(self, config_file: str, tool_cache_data_dir=None, **kwds):
+    def create_tool(self, config_file: str, tool_cache_data_dir: Optional[str] = None, **kwds):
         cache = self.get_cache_region(tool_cache_data_dir)
         if config_file.endswith(".xml") and cache and not cache.disabled:
             tool_document = cache.get(config_file)

lib/galaxy/tools/execute.py

@@ -16,6 +16,7 @@
     NamedTuple,
     Optional,
     Tuple,
+    TypeAlias,
     Union,
 )
 
@@ -52,7 +53,7 @@
 
 
 CompletedJobsT = Dict[int, Optional[model.Job]]
-JobCallbackT = Callable
+JobCallbackT: TypeAlias = Callable
 WorkflowResourceParametersT = Dict[str, Any]
 DatasetCollectionElementsSliceT = Dict[str, model.DatasetCollectionElement]
 DEFAULT_USE_CACHED_JOB = False