Skip to content

Commit

Permalink
Fix incorrect type in tool_util.deps and fix package structure.
Browse files Browse the repository at this point in the history
  • Loading branch information
jmchilton committed Feb 26, 2025
1 parent ca4faaa commit f2af556
Showing 1 changed file with 21 additions and 7 deletions.
28 changes: 21 additions & 7 deletions lib/galaxy/tool_util/deps/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,10 @@
Dict,
List,
Optional,
TYPE_CHECKING,
)

from typing_extensions import Protocol

from galaxy.util import (
hash_util,
plugin_config,
Expand All @@ -33,9 +34,6 @@
)
from .resolvers.tool_shed_packages import ToolShedPackageDependencyResolver

if TYPE_CHECKING:
from galaxy.jobs import JobDestination

log = logging.getLogger(__name__)

CONFIG_VAL_NOT_FOUND = object()
Expand Down Expand Up @@ -106,6 +104,16 @@ def build_dependency_manager(
return NullDependencyManager()


ContainerType = str
DestinationId = str
DestinationParametersType = Dict[str, Any]


class DestinationProtocol(Protocol):
id: Optional[DestinationId]
params: DestinationParametersType


class DependencyManager:
"""
A DependencyManager attempts to resolve named and versioned dependencies by
Expand All @@ -118,6 +126,7 @@ class DependencyManager:
dependency available in the current shell environment.
"""

_destination_for_container_type: Dict[ContainerType, List[DestinationProtocol]]
cached = False

def __init__(
Expand Down Expand Up @@ -145,21 +154,26 @@ def __init__(
plugin_source = self.__build_dependency_resolvers_plugin_source(conf_file)
self.dependency_resolvers = self.__parse_resolver_conf_plugins(plugin_source)
self._enabled_container_types: List[str] = []
self._destination_for_container_type: Dict[str, Dict[str, JobDestination]] = {}
self._destination_for_container_type = {}

def set_enabled_container_types(self, container_types_to_destinations):
def set_enabled_container_types(
self, container_types_to_destinations: Dict[ContainerType, List[DestinationProtocol]]
):
"""Set the union of all enabled container types."""
self._enabled_container_types = list(container_types_to_destinations.keys())
# Just pick first enabled destination for a container type, probably covers the most common deployment scenarios
self._destination_for_container_type = container_types_to_destinations

def get_destination_info_for_container_type(self, container_type, destination_id=None):
def get_destination_info_for_container_type(
self, container_type: ContainerType, destination_id: Optional[DestinationId] = None
) -> Optional[DestinationParametersType]:
if destination_id is None:
return next(iter(self._destination_for_container_type[container_type])).params
else:
for destination in self._destination_for_container_type[container_type]:
if destination.id == destination_id:
return destination.params
return None

@property
def enabled_container_types(self):
Expand Down

0 comments on commit f2af556

Please sign in to comment.