Merge Target class with CondaTarget

lib/galaxy/tool_util/deps/conda_util.py

@@ -409,21 +409,26 @@ def installed_conda_targets(conda_context: CondaContext) -> Iterator["CondaTarge
     for name in dir_contents:
         versioned_match = VERSIONED_ENV_DIR_NAME.match(name)
         if versioned_match:
-            yield CondaTarget(versioned_match.group(1), versioned_match.group(2))
+            yield CondaTarget(versioned_match.group(1), version=versioned_match.group(2))
 
         unversioned_match = UNVERSIONED_ENV_DIR_NAME.match(name)
         if unversioned_match:
             yield CondaTarget(unversioned_match.group(1))
 
 
 class CondaTarget:
-    def __init__(self, package: str, version: Optional[str] = None, channel: Optional[str] = None) -> None:
+    def __init__(
+        self, package: str, version: Optional[str] = None, build: Optional[str] = None, channel: Optional[str] = None
+    ) -> None:
         if SHELL_UNSAFE_PATTERN.search(package) is not None:
             raise ValueError(f"Invalid package [{package}] encountered.")
-        self.package = package
+        self.package = package.lower()
         if version and SHELL_UNSAFE_PATTERN.search(version) is not None:
             raise ValueError(f"Invalid version [{version}] encountered.")
         self.version = version
+        if build is not None and SHELL_UNSAFE_PATTERN.search(build) is not None:
+            raise ValueError(f"Invalid build [{build}] encountered.")
+        self.build = build
         if channel and SHELL_UNSAFE_PATTERN.search(channel) is not None:
             raise ValueError(f"Invalid version [{channel}] encountered.")
         self.channel = channel
@@ -432,8 +437,8 @@ def __str__(self) -> str:
         attributes = f"package={self.package}"
         if self.version is not None:
             attributes += f",version={self.version}"
-        else:
-            attributes += ",unversioned"
+        if self.build is not None:
+            attributes += f",build={self.build}"
 
         if self.channel:
             attributes += f",channel={self.channel}"
@@ -446,9 +451,12 @@ def __str__(self) -> str:
     def package_specifier(self) -> str:
         """Return a package specifier as consumed by conda install/create."""
         if self.version:
-            return f"{self.package}={self.version}"
+            spec = f"{self.package}={self.version}"
         else:
-            return self.package
+            spec = f"{self.package}=*"
+        if self.build:
+            spec += f"={self.build}"
+        return spec
 
     @property
     def install_environment(self) -> str:
@@ -462,11 +470,16 @@ def install_environment(self) -> str:
             return f"__{self.package}@_uv_"
 
     def __hash__(self) -> int:
-        return hash((self.package, self.version, self.channel))
+        return hash((self.package, self.version, self.build, self.channel))
 
     def __eq__(self, other: Any) -> bool:
         if isinstance(other, self.__class__):
-            return (self.package, self.version, self.channel) == (other.package, other.version, other.channel)
+            return (self.package, self.version, self.build, self.channel) == (
+                other.package,
+                other.version,
+                other.build,
+                other.channel,
+            )
         return False
 
 
@@ -557,7 +570,7 @@ def best_search_result(
 ) -> Union[Tuple[None, None], Tuple[Dict[str, Any], bool]]:
     """Find best "conda search" result for specified target.
 
-    Return ``None`` if no results match.
+    Return (``None``, ``None``) if no results match.
     """
     # Cannot specify the version here (i.e. conda_target.package_specifier)
     # because if the version is not found, the exec_search() call would fail.
@@ -589,10 +602,15 @@ def best_search_result(
 
 
 def is_search_hit_exact(conda_target: CondaTarget, search_hit: Dict[str, Any]) -> bool:
-    target_version = conda_target.version
     # It'd be nice to make request verson of 1.0 match available
     # version of 1.0.3 or something like that.
-    return bool(not target_version or search_hit["version"] == target_version)
+    target_version = conda_target.version
+    if target_version and search_hit["version"] != target_version:
+        return False
+    target_build = conda_target.build
+    if target_build and search_hit["build"] != target_build:
+        return False
+    return True
 
 
 def is_conda_target_installed(conda_target: CondaTarget, conda_context: CondaContext) -> bool:
@@ -671,6 +689,7 @@ def build_isolated_environment(
 def requirement_to_conda_targets(requirement: "ToolRequirement") -> Optional[CondaTarget]:
     conda_target = None
     if requirement.type == "package":
+        assert requirement.name
         conda_target = CondaTarget(requirement.name, version=requirement.version)
     return conda_target
 

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

@@ -6,6 +6,8 @@
 )
 from typing import (
     Any,
+    Container,
+    List,
     Optional,
     TYPE_CHECKING,
 )
@@ -16,7 +18,11 @@
 if TYPE_CHECKING:
     from beaker.cache import Cache
 
-    from ..dependencies import AppInfo
+    from ..dependencies import (
+        AppInfo,
+        ToolInfo,
+    )
+    from ..requirements import ContainerDescription
 
 
 class ResolutionCache(Bunch):
@@ -53,20 +59,24 @@ def _get_config_option(self, key: str, default: Any = None) -> Any:
             return default
 
     @abstractmethod
-    def resolve(self, enabled_container_types, tool_info, resolution_cache=None, **kwds):
+    def resolve(
+        self, enabled_container_types: List[str], tool_info: "ToolInfo", **kwds
+    ) -> Optional["ContainerDescription"]:
         """Find a container matching all supplied requirements for tool.
 
-        The supplied argument is a :class:`galaxy.tool_util.deps.containers.ToolInfo` description
+        The supplied argument is a :class:`galaxy.tool_util.deps.dependencies.ToolInfo` description
         of the tool and its requirements.
         """
 
     @abstractproperty
-    def resolver_type(self):
+    def resolver_type(self) -> str:
         """Short label for the type of container resolution."""
 
-    def _container_type_enabled(self, container_description, enabled_container_types):
+    def _container_type_enabled(
+        self, container_description: "ContainerDescription", enabled_container_types: Container[str]
+    ) -> bool:
         """Return a boolean indicating if the specified container type is enabled."""
         return container_description.type in enabled_container_types
 
-    def __str__(self):
+    def __str__(self) -> str:
         return f"{self.__class__.__name__}[]"

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

@@ -2,14 +2,21 @@
 import copy
 import logging
 import os
-from typing import cast
+from typing import (
+    cast,
+    Optional,
+    TYPE_CHECKING,
+)
 
 from galaxy.util.commands import shell
 from . import ContainerResolver
 from .mulled import CliContainerResolver
 from ..container_classes import SingularityContainer
 from ..requirements import ContainerDescription
 
+if TYPE_CHECKING:
+    from ..dependencies import AppInfo
+
 log = logging.getLogger(__name__)
 
 DEFAULT_SHELL = "/bin/bash"
@@ -62,8 +69,8 @@ class CachedExplicitSingularityContainerResolver(CliContainerResolver):
     container_type = "singularity"
     cli = "singularity"
 
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
+    def __init__(self, app_info: Optional["AppInfo"] = None, **kwargs) -> None:
+        super().__init__(app_info=app_info, **kwargs)
         self.cache_directory_path = kwargs.get(
             "cache_directory", os.path.join(kwargs["app_info"].container_image_cache_path, "singularity", "explicit")
         )
@@ -116,8 +123,8 @@ def __str__(self):
 
 
 class BaseAdminConfiguredContainerResolver(ContainerResolver):
-    def __init__(self, app_info=None, shell=DEFAULT_SHELL, **kwds):
-        super().__init__(app_info, **kwds)
+    def __init__(self, app_info: Optional["AppInfo"] = None, shell=DEFAULT_SHELL, **kwds) -> None:
+        super().__init__(app_info=app_info, **kwds)
         self.shell = shell
 
     def _container_description(self, identifier, container_type):
@@ -135,8 +142,8 @@ class FallbackContainerResolver(BaseAdminConfiguredContainerResolver):
     resolver_type = "fallback"
     container_type = "docker"
 
-    def __init__(self, app_info=None, identifier="", **kwds):
-        super().__init__(app_info, **kwds)
+    def __init__(self, app_info: Optional["AppInfo"] = None, identifier="", **kwds) -> None:
+        super().__init__(app_info=app_info, **kwds)
         assert identifier, "fallback container resolver must be specified with non-empty identifier"
         self.identifier = identifier
 
@@ -187,8 +194,8 @@ class RequiresGalaxyEnvironmentSingularityContainerResolver(RequiresGalaxyEnviro
 class MappingContainerResolver(BaseAdminConfiguredContainerResolver):
     resolver_type = "mapping"
 
-    def __init__(self, app_info=None, **kwds):
-        super().__init__(app_info, **kwds)
+    def __init__(self, app_info: Optional["AppInfo"] = None, **kwds) -> None:
+        super().__init__(app_info=app_info, **kwds)
         mappings = self.resolver_kwds["mappings"]
         assert isinstance(mappings, list), "mapping container resolver must be specified with mapping list"
         self.mappings = mappings