Type annotations and fixes

- Use `galaxy.util.path.StrPath` everywhere

lib/galaxy/app.py

@@ -34,6 +34,7 @@
     GalaxyTaskBeforeStartUserRateLimitPostgres,
     GalaxyTaskBeforeStartUserRateLimitStandard,
 )
+from galaxy.config import GalaxyAppConfiguration
 from galaxy.config_watchers import ConfigWatchers
 from galaxy.datatypes.registry import Registry
 from galaxy.files import (
@@ -206,7 +207,7 @@ def shutdown(self):
 
 
 class SentryClientMixin:
-    config: config.GalaxyAppConfiguration
+    config: GalaxyAppConfiguration
     application_stack: ApplicationStack
 
     def configure_sentry_client(self):
@@ -263,7 +264,7 @@ class MinimalGalaxyApplication(BasicSharedApp, HaltableContainer, SentryClientMi
     """Encapsulates the state of a minimal Galaxy application"""
 
     model: GalaxyModelMapping
-    config: config.GalaxyAppConfiguration
+    config: GalaxyAppConfiguration
     tool_cache: ToolCache
     job_config: jobs.JobConfiguration
     toolbox_search: ToolBoxSearch
@@ -287,7 +288,7 @@ def __init__(self, fsmon=False, **kwargs) -> None:
         self.name = "galaxy"
         self.is_webapp = False
         # Read config file and check for errors
-        self.config = self._register_singleton(config.GalaxyAppConfiguration, config.GalaxyAppConfiguration(**kwargs))
+        self.config = self._register_singleton(GalaxyAppConfiguration, GalaxyAppConfiguration(**kwargs))
         self.config.check()
         config_file = kwargs.get("global_conf", {}).get("__file__", None)
         if config_file:

lib/galaxy/app_unittest_utils/tools_support.py

@@ -20,6 +20,7 @@
 from galaxy.tool_util.parser import get_tool_source
 from galaxy.tools import create_tool_from_source
 from galaxy.util.bunch import Bunch
+from galaxy.util.path import StrPath
 
 datatypes_registry = galaxy.datatypes.registry.Registry()
 datatypes_registry.load_datatypes()
@@ -83,10 +84,10 @@ def _init_tool(
         tool_id="test_tool",
         extra_file_contents=None,
         extra_file_path=None,
-        tool_path=None,
+        tool_path: Optional[StrPath] = None,
     ):
         if tool_path is None:
-            self.tool_file = os.path.join(self.test_directory, filename)
+            self.tool_file: StrPath = os.path.join(self.test_directory, filename)
             contents_template = string.Template(tool_contents)
             tool_contents = contents_template.safe_substitute(dict(version=version, profile=profile, tool_id=tool_id))
             self.__write_tool(tool_contents)
@@ -96,7 +97,7 @@ def _init_tool(
             self.tool_file = tool_path
         return self.__setup_tool()
 
-    def _init_tool_for_path(self, tool_file):
+    def _init_tool_for_path(self, tool_file: StrPath):
         self.tool_file = tool_file
         return self.__setup_tool()
 

lib/galaxy/celery/tasks.py

@@ -75,10 +75,8 @@ def setup_data_table_manager(app):
 
 
 @lru_cache
-def cached_create_tool_from_representation(app, raw_tool_source):
-    return create_tool_from_representation(
-        app=app, raw_tool_source=raw_tool_source, tool_dir="", tool_source_class="XmlToolSource"
-    )
+def cached_create_tool_from_representation(app: MinimalManagerApp, raw_tool_source: str):
+    return create_tool_from_representation(app=app, raw_tool_source=raw_tool_source, tool_source_class="XmlToolSource")
 
 
 @galaxy_task(action="recalculate a user's disk usage")
@@ -305,7 +303,7 @@ def _fetch_data(setup_return):
     working_directory = Path(tool_job_working_directory) / "working"
     datatypes_registry = DatatypesRegistry()
     datatypes_registry.load_datatypes(
-        galaxy_directory,
+        galaxy_directory(),
         config=Path(tool_job_working_directory) / "metadata" / "registry.xml",
         use_build_sites=False,
         use_converters=False,

lib/galaxy/datatypes/data.py

@@ -207,6 +207,7 @@ class Data(metaclass=DataMeta):
     edam_data = "data_0006"
     edam_format = "format_1915"
     file_ext = "data"
+    is_subclass = False
     # Data is not chunkable by default.
     CHUNKABLE = False
 

lib/galaxy/datatypes/sniff.py

@@ -17,8 +17,10 @@
     Callable,
     Dict,
     IO,
+    Iterable,
     NamedTuple,
     Optional,
+    TYPE_CHECKING,
     Union,
 )
 
@@ -44,6 +46,8 @@
     pass
 import magic  # isort:skip
 
+if TYPE_CHECKING:
+    from .data import Data
 
 log = logging.getLogger(__name__)
 
@@ -689,7 +693,7 @@ def _get_file_prefix(filename_or_file_prefix: Union[str, FilePrefix], auto_decom
     return filename_or_file_prefix
 
 
-def run_sniffers_raw(file_prefix: FilePrefix, sniff_order):
+def run_sniffers_raw(file_prefix: FilePrefix, sniff_order: Iterable["Data"]):
     """Run through sniffers specified by sniff_order, return None of None match."""
     fname = file_prefix.filename
     file_ext = None
@@ -718,15 +722,16 @@ def run_sniffers_raw(file_prefix: FilePrefix, sniff_order):
                 continue
         try:
             if hasattr(datatype, "sniff_prefix"):
-                if file_prefix.compressed_format and getattr(datatype, "compressed_format", None):
+                datatype_compressed_format = getattr(datatype, "compressed_format", None)
+                if file_prefix.compressed_format and datatype_compressed_format:
                     # Compare the compressed format detected
                     # to the expected.
-                    if file_prefix.compressed_format != datatype.compressed_format:
+                    if file_prefix.compressed_format != datatype_compressed_format:
                         continue
                 if datatype.sniff_prefix(file_prefix):
                     file_ext = datatype.file_ext
                     break
-            elif datatype.sniff(fname):
+            elif hasattr(datatype, "sniff") and datatype.sniff(fname):
                 file_ext = datatype.file_ext
                 break
         except Exception:

lib/galaxy/files/sources/util.py

@@ -1,10 +1,8 @@
 import time
-from os import PathLike
 from typing import (
     List,
     Optional,
     Tuple,
-    Union,
 )
 
 from galaxy import exceptions
@@ -20,8 +18,7 @@
     requests,
 )
 from galaxy.util.config_parsers import IpAllowedListEntryT
-
-TargetPathT = Union[str, PathLike]
+from galaxy.util.path import StrPath
 
 
 def _not_implemented(drs_uri: str, desc: str) -> NotImplementedError:
@@ -79,7 +76,7 @@ def _get_access_info(obj_url: str, access_method: dict, headers=None) -> Tuple[s
 
 def fetch_drs_to_file(
     drs_uri: str,
-    target_path: TargetPathT,
+    target_path: StrPath,
     user_context: Optional[FileSourcesUserContext],
     force_http=False,
     retry_options: Optional[RetryOptions] = None,

lib/galaxy/jobs/runners/__init__.py

@@ -10,12 +10,18 @@
 import threading
 import time
 import traceback
-import typing
 import uuid
 from queue import (
     Empty,
     Queue,
 )
+from typing import (
+    Any,
+    Dict,
+    Optional,
+    TYPE_CHECKING,
+    Union,
+)
 
 from sqlalchemy import select
 from sqlalchemy.orm import object_session
@@ -58,7 +64,7 @@
 from galaxy.util.monitors import Monitors
 from .state_handler_factory import build_state_handlers
 
-if typing.TYPE_CHECKING:
+if TYPE_CHECKING:
     from galaxy.app import GalaxyManagerApplication
     from galaxy.jobs import (
         JobDestination,
@@ -183,7 +189,7 @@ def run_next(self):
                         # Prevent fail_job cycle in the work_queue
                         self.work_queue.put((self.fail_job, job_state))
 
-    def _ensure_db_session(self, arg: typing.Union["JobWrapper", "JobState"]) -> None:
+    def _ensure_db_session(self, arg: Union["JobWrapper", "JobState"]) -> None:
         """Ensure Job object belongs to current session."""
         try:
             job_wrapper = arg.job_wrapper  # type: ignore[union-attr]
@@ -264,7 +270,7 @@ def url_to_destination(self, url: str):
         """
         return galaxy.jobs.JobDestination(runner=url.split(":")[0])
 
-    def parse_destination_params(self, params: typing.Dict[str, typing.Any]):
+    def parse_destination_params(self, params: Dict[str, Any]):
         """Parse the JobDestination ``params`` dict and return the runner's native representation of those params."""
         raise NotImplementedError()
 
@@ -347,8 +353,8 @@ def build_command_line(
     def get_work_dir_outputs(
         self,
         job_wrapper: "MinimalJobWrapper",
-        job_working_directory: typing.Optional[str] = None,
-        tool_working_directory: typing.Optional[str] = None,
+        job_working_directory: Optional[str] = None,
+        tool_working_directory: Optional[str] = None,
     ):
         """
         Returns list of pairs (source_file, destination) describing path
@@ -527,10 +533,10 @@ def write_executable_script(self, path: str, contents: str, job_io: DescribesScr
     def _find_container(
         self,
         job_wrapper: "MinimalJobWrapper",
-        compute_working_directory: typing.Optional[str] = None,
-        compute_tool_directory: typing.Optional[str] = None,
-        compute_job_directory: typing.Optional[str] = None,
-        compute_tmp_directory: typing.Optional[str] = None,
+        compute_working_directory: Optional[str] = None,
+        compute_tool_directory: Optional[str] = None,
+        compute_job_directory: Optional[str] = None,
+        compute_tmp_directory: Optional[str] = None,
     ):
         job_directory_type = "galaxy" if compute_working_directory is None else "pulsar"
         if not compute_working_directory:
@@ -542,7 +548,7 @@ def _find_container(
         tool = job_wrapper.tool
         assert tool
         if not compute_tool_directory:
-            compute_tool_directory = tool.tool_dir
+            compute_tool_directory = str(tool.tool_dir) if tool.tool_dir is not None else None
 
         if not compute_tmp_directory:
             compute_tmp_directory = job_wrapper.tmp_directory()
@@ -600,7 +606,7 @@ def fail_job(self, job_state: "JobState", exception=False, message="Job failed",
                 fail_message, tool_stdout=tool_stdout, tool_stderr=tool_stderr, exception=exception
             )
 
-    def mark_as_resubmitted(self, job_state: "JobState", info: typing.Optional[str] = None):
+    def mark_as_resubmitted(self, job_state: "JobState", info: Optional[str] = None):
         job_state.job_wrapper.mark_as_resubmitted(info=info)
         if not self.app.config.track_jobs_in_database:
             job_state.job_wrapper.change_state(model.Job.states.QUEUED)

lib/galaxy/tool_util/deps/mulled/mulled_update_singularity_containers.py

@@ -1,7 +1,6 @@
 #!/usr/bin/env python
 
 import argparse
-import os
 import os.path
 import subprocess
 import tempfile
@@ -11,10 +10,10 @@
     Any,
     Dict,
     List,
-    Union,
 )
 
 from galaxy.util import unicodify
+from galaxy.util.path import StrPath
 from .get_tests import (
     hashed_test_search,
     import_test_to_command_list,
@@ -46,7 +45,7 @@ def docker_to_singularity(container, installation, filepath, no_sudo=False):
 
 
 def singularity_container_test(
-    tests: Dict[str, Dict[str, Any]], installation: str, filepath: Union[str, os.PathLike]
+    tests: Dict[str, Dict[str, Any]], installation: str, filepath: StrPath
 ) -> Dict[str, List]:
     """
     Run tests, record if they pass or fail

lib/galaxy/tool_util/parser/factory.py

@@ -1,13 +1,11 @@
 """Constructors for concrete tool and input source objects."""
 
 import logging
-from pathlib import PurePath
 from typing import (
     Callable,
     Dict,
     List,
     Optional,
-    Union,
 )
 
 from yaml import safe_load
@@ -17,6 +15,7 @@
     ElementTree,
     parse_xml_string_to_etree,
 )
+from galaxy.util.path import StrPath
 from galaxy.util.yaml_util import ordered_load
 from .cwl import (
     CwlToolSource,
@@ -61,7 +60,7 @@ def build_yaml_tool_source(yaml_string: str) -> YamlToolSource:
 
 
 def get_tool_source(
-    config_file: Optional[Union[str, PurePath]] = None,
+    config_file: Optional[StrPath] = None,
     xml_tree: Optional[ElementTree] = None,
     enable_beta_formats: bool = True,
     tool_location_fetcher: Optional[ToolLocationFetcher] = None,
@@ -87,8 +86,7 @@ def get_tool_source(
         tool_location_fetcher = ToolLocationFetcher()
 
     assert config_file
-    if isinstance(config_file, PurePath):
-        config_file = str(config_file)
+    config_file = str(config_file)
 
     config_file = tool_location_fetcher.to_tool_path(config_file)
     if not enable_beta_formats:

lib/galaxy/tool_util/parser/interface.py

@@ -155,7 +155,7 @@ def parse_id(self) -> Optional[str]:
     def parse_version(self) -> Optional[str]:
         """Parse a version describing the abstract tool."""
 
-    def parse_tool_module(self):
+    def parse_tool_module(self) -> Optional[Tuple[str, str]]:
         """Load Tool class from a custom module. (Optional).
 
         If not None, return pair containing module and class (as strings).
@@ -169,7 +169,7 @@ def parse_action_module(self):
         """
         return None
 
-    def parse_tool_type(self):
+    def parse_tool_type(self) -> Optional[str]:
         """Load simple tool type string (e.g. 'data_source', 'default')."""
         return None
 

lib/galaxy/tool_util/parser/xml.py

@@ -190,8 +190,7 @@ def parse_action_module(self):
 
     def parse_tool_type(self):
         root = self.root
-        if root.get("tool_type", None) is not None:
-            return root.get("tool_type")
+        return root.get("tool_type")
 
     def parse_name(self):
         return self.root.get("name") or self.parse_id()

lib/galaxy/tool_util/verify/codegen.py

@@ -3,8 +3,6 @@
 # how to use this function...
 # PYTHONPATH=lib python lib/galaxy/tool_util/verify/codegen.py
 
-from __future__ import annotations
-
 import argparse
 import inspect
 import os
@@ -34,7 +32,7 @@
 
 Children = Literal["allowed", "required", "forbidden"]
 
-DESCRIPTION = """This script synchronizes dynamic code aritfacts against models in Galaxy.
+DESCRIPTION = """This script synchronizes dynamic code artifacts against models in Galaxy.
 
 Right now this just synchronizes Galaxy's XSD file against documentation in Galaxy's assertion modules but
 in the future it will also build Pydantic models for these functions.