Merge pull request #14371 from mr-c/tool_utils_types_22.05

lib/galaxy/tool_util/deps/__init__.py

@@ -6,6 +6,13 @@ import json
 import logging
 import os.path
 import shutil
+from typing import (
+    Any,
+    Dict,
+    List,
+    Optional,
+    TYPE_CHECKING,
+)
 
 from galaxy.util import (
     hash_util,
@@ -26,14 +33,20 @@ from .resolvers import (
 )
 from .resolvers.tool_shed_packages import ToolShedPackageDependencyResolver
 
+if TYPE_CHECKING:
+    from galaxy.jobs import JobDestination
+
 log = logging.getLogger(__name__)
 
 CONFIG_VAL_NOT_FOUND = object()
 
 
 def build_dependency_manager(
-    app_config_dict=None, resolution_config_dict=None, conf_file=None, default_tool_dependency_dir=None
-):
+    app_config_dict: Optional[Dict[str, Any]] = None,
+    resolution_config_dict: Optional[Dict[str, Any]] = None,
+    conf_file: Optional[str] = None,
+    default_tool_dependency_dir: Optional[str] = None,
+) -> "DependencyManager":
     """Build a DependencyManager object from app and/or resolution config.
 
     If app_config_dict is specified, it should be application configuration information
@@ -88,13 +101,9 @@ def build_dependency_manager(
             "app_config": app_config_dict,
         }
         if string_as_bool(app_config_dict.get("use_cached_dependency_manager")):
-            dependency_manager = CachedDependencyManager(**dependency_manager_kwds)
-        else:
-            dependency_manager = DependencyManager(**dependency_manager_kwds)
-    else:
-        dependency_manager = NullDependencyManager()
-
-    return dependency_manager
+            return CachedDependencyManager(**dependency_manager_kwds)
+        return DependencyManager(**dependency_manager_kwds)
+    return NullDependencyManager()
 
 
 class DependencyManager:
@@ -111,7 +120,9 @@ class DependencyManager:
 
     cached = False
 
-    def __init__(self, default_base_path, conf_file=None, app_config=None):
+    def __init__(
+        self, default_base_path: str, conf_file: Optional[str] = None, app_config: Optional[Dict[str, Any]] = None
+    ) -> None:
         """
         Create a new dependency manager looking for packages under the paths listed
         in `base_paths`.  The default base path is app.config.tool_dependency_dir.
@@ -133,8 +144,8 @@ class DependencyManager:
         else:
             plugin_source = self.__build_dependency_resolvers_plugin_source(conf_file)
         self.dependency_resolvers = self.__parse_resolver_conf_plugins(plugin_source)
-        self._enabled_container_types = []
-        self._destination_for_container_type = {}
+        self._enabled_container_types: List[str] = []
+        self._destination_for_container_type: Dict[str, Dict[str, "JobDestination"]] = {}
 
     def set_enabled_container_types(self, container_types_to_destinations):
         """Set the union of all enabled container types."""
@@ -184,7 +195,7 @@ class DependencyManager:
     def precache(self):
         return string_as_bool(self.get_app_option("precache_dependencies", True))
 
-    def dependency_shell_commands(self, requirements, **kwds):
+    def dependency_shell_commands(self, requirements: ToolRequirements, **kwds: Any) -> List[str]:
         requirements_to_dependencies = self.requirements_to_dependencies(requirements, **kwds)
         ordered_dependencies = OrderedSet(requirements_to_dependencies.values())
         return [

lib/galaxy/tool_util/deps/container_resolvers/explicit.py

@@ -1,6 +1,7 @@
 """This module describes the :class:`ExplicitContainerResolver` ContainerResolver plugin."""
 import logging
 import os
+from typing import cast
 
 from galaxy.util.commands import shell
 from .mulled import CliContainerResolver
@@ -88,7 +89,7 @@ class CachedExplicitSingularityContainerResolver(CliContainerResolver):
                 return None
             if not self.cli_available:
                 return container_description
-            image_id = container_description.identifier
+            image_id = cast(str, container_description.identifier)
             cache_path = os.path.normpath(os.path.join(self.cache_directory_path, image_id))
             if install and not os.path.exists(cache_path):
                 destination_info = {}

lib/galaxy/tool_util/deps/containers.py

@@ -1,6 +1,13 @@
 import collections
 import logging
 import os
+from typing import (
+    Any,
+    Dict,
+    List,
+    Optional,
+    TYPE_CHECKING,
+)
 
 from galaxy.util import (
     asbool,
@@ -27,6 +34,14 @@ from .container_resolvers.mulled import (
 )
 from .requirements import ContainerDescription
 
+if TYPE_CHECKING:
+    from beaker.cache import Cache
+
+    from .dependencies import (
+        AppInfo,
+        ToolInfo,
+    )
+
 log = logging.getLogger(__name__)
 
 
@@ -218,7 +233,12 @@ class NullContainerFinder:
 class ContainerRegistry:
     """Loop through enabled ContainerResolver plugins and find first match."""
 
-    def __init__(self, app_info, destination_info=None, mulled_resolution_cache=None):
+    def __init__(
+        self,
+        app_info: "AppInfo",
+        destination_info: Optional[Dict[str, Any]] = None,
+        mulled_resolution_cache: Optional["Cache"] = None,
+    ) -> None:
         self.resolver_classes = self.__resolvers_dict()
         self.enable_mulled_containers = app_info.enable_mulled_containers
         self.app_info = app_info
@@ -292,7 +312,9 @@ class ContainerRegistry:
             cache.mulled_resolution_cache = self.mulled_resolution_cache
         return cache
 
-    def find_best_container_description(self, enabled_container_types, tool_info, **kwds):
+    def find_best_container_description(
+        self, enabled_container_types: List[str], tool_info: "ToolInfo", **kwds: Any
+    ) -> Optional[ContainerDescription]:
         """Yield best container description of supplied types matching tool info."""
         try:
             resolved_container_description = self.resolve(enabled_container_types, tool_info, **kwds)

lib/galaxy/tool_util/deps/dependencies.py

@@ -1,4 +1,16 @@
-from galaxy.tool_util.deps.requirements import ToolRequirements
+from typing import (
+    Any,
+    Dict,
+    List,
+    Optional,
+    Union,
+)
+
+from galaxy.tool_util.deps.requirements import (
+    ContainerDescription,
+    ToolRequirement,
+    ToolRequirements,
+)
 from galaxy.util import bunch
 from .mulled.mulled_build import DEFAULT_CHANNELS
 
@@ -6,20 +18,20 @@ from .mulled.mulled_build import DEFAULT_CHANNELS
 class AppInfo:
     def __init__(
         self,
-        galaxy_root_dir=None,
-        default_file_path=None,
-        tool_data_path=None,
-        shed_tool_data_path=None,
-        outputs_to_working_directory=False,
-        container_image_cache_path=None,
-        library_import_dir=None,
-        enable_mulled_containers=False,
-        container_resolvers_config_file=None,
-        container_resolvers_config_dict=None,
-        involucro_path=None,
-        involucro_auto_init=True,
-        mulled_channels=DEFAULT_CHANNELS,
-    ):
+        galaxy_root_dir: Optional[str] = None,
+        default_file_path: Optional[str] = None,
+        tool_data_path: Optional[str] = None,
+        shed_tool_data_path: Optional[str] = None,
+        outputs_to_working_directory: bool = False,
+        container_image_cache_path: Optional[str] = None,
+        library_import_dir: Optional[str] = None,
+        enable_mulled_containers: bool = False,
+        container_resolvers_config_file: Optional[str] = None,
+        container_resolvers_config_dict: Optional[Dict[str, Any]] = None,
+        involucro_path: Optional[str] = None,
+        involucro_auto_init: bool = True,
+        mulled_channels: List[str] = DEFAULT_CHANNELS,
+    ) -> None:
         self.galaxy_root_dir = galaxy_root_dir
         self.default_file_path = default_file_path
         self.tool_data_path = tool_data_path
@@ -43,14 +55,14 @@ class ToolInfo:
 
     def __init__(
         self,
-        container_descriptions=None,
-        requirements=None,
-        requires_galaxy_python_environment=False,
+        container_descriptions: Optional[List["ContainerDescription"]] = None,
+        requirements: Optional[Union["ToolRequirements", List["ToolRequirement"]]] = None,
+        requires_galaxy_python_environment: bool = False,
         env_pass_through=None,
         guest_ports=None,
-        tool_id=None,
-        tool_version=None,
-        profile=-1,
+        tool_id: Optional[str] = None,
+        tool_version: Optional[str] = None,
+        profile: float = -1,
     ):
         if env_pass_through is None:
             env_pass_through = ["GALAXY_SLOTS", "GALAXY_MEMORY_MB", "GALAXY_MEMORY_MB_PER_SLOT"]

lib/galaxy/tool_util/deps/requirements.py

@@ -1,5 +1,6 @@
 import copy
 from typing import (
+    Any,
     Callable,
     cast,
     Dict,
@@ -46,7 +47,7 @@ class ToolRequirement:
         return copy.deepcopy(self)
 
     @staticmethod
-    def from_dict(d):
+    def from_dict(d: Dict[str, Any]) -> "ToolRequirement":
         version = d.get("version")
         name = d.get("name")
         type = d.get("type")
@@ -123,7 +124,7 @@ class ToolRequirements:
             self.tool_requirements = OrderedSet()
 
     @staticmethod
-    def from_list(requirements):
+    def from_list(requirements: Union[List[ToolRequirement], Dict[str, Any]]) -> "ToolRequirements":
         return ToolRequirements(requirements)
 
     @property
@@ -180,11 +181,11 @@ DEFAULT_CONTAINER_SHELL = "/bin/sh"  # Galaxy assumes bash, but containers are u
 class ContainerDescription:
     def __init__(
         self,
-        identifier=None,
-        type=DEFAULT_CONTAINER_TYPE,
-        resolve_dependencies=DEFAULT_CONTAINER_RESOLVE_DEPENDENCIES,
-        shell=DEFAULT_CONTAINER_SHELL,
-    ):
+        identifier: Optional[str] = None,
+        type: str = DEFAULT_CONTAINER_TYPE,
+        resolve_dependencies: bool = DEFAULT_CONTAINER_RESOLVE_DEPENDENCIES,
+        shell: str = DEFAULT_CONTAINER_SHELL,
+    ) -> None:
         # Force to lowercase because container image names must be lowercase
         self.identifier = identifier.lower() if identifier else None
         self.type = type

lib/galaxy/util/__init__.py

@@ -1006,7 +1006,7 @@ def asbool(obj):
     return bool(obj)
 
 
-def string_as_bool(string: str) -> bool:
+def string_as_bool(string: typing.Any) -> bool:
     if str(string).lower() in ("true", "yes", "on", "1"):
         return True
     else:

packages/tool_util/setup.cfg

@@ -55,6 +55,7 @@ console_scripts =
 [options.extras_require]
 cwl =
     cwltool==3.1.20211107152837
+    schema-salad==8.3.20220525163636
 mulled =
     jinja2
     Whoosh