Warning

This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.tool_util.deps.container_resolvers

"""The module defines the abstract interface for resolving container images for tool execution."""
from abc import (
    ABCMeta,
    abstractmethod,
    abstractproperty,
)
from typing import (
    Any,
    Container,
    Optional,
    TYPE_CHECKING,
)

from galaxy.util.bunch import Bunch
from galaxy.util.dictifiable import Dictifiable

if TYPE_CHECKING:
    from beaker.cache import Cache

    from ..dependencies import (
        AppInfo,
        ToolInfo,
    )
    from ..requirements import ContainerDescription


[docs]class ResolutionCache(Bunch): """Simple cache for duplicated computation created once per set of requests (likely web request in Galaxy context). This should not be assumed to be thread safe - resolution using a given cache should all occur one resolution at a time in a single thread. """ mulled_resolution_cache: Optional["Cache"] = None
[docs]class ContainerResolver(Dictifiable, metaclass=ABCMeta): """Description of a technique for resolving container images for tool execution.""" # Keys for dictification. dict_collection_visible_keys = ["resolver_type", "can_uninstall_dependencies", "builds_on_resolution"] can_uninstall_dependencies = False builds_on_resolution = False read_only = True # not used for containers, but set for when they are used like dependency resolvers
[docs] def __init__(self, app_info: "AppInfo", **kwds) -> None: """Default initializer for ``ContainerResolver`` subclasses.""" self.app_info = app_info self.resolver_kwds = kwds
def _get_config_option(self, key: str, default: Any = None) -> Any: """Look in resolver-specific settings for option and then fallback to global settings. """ return getattr(self.app_info, key, default)
[docs] @abstractmethod def resolve( self, enabled_container_types: Container[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.dependencies.ToolInfo` description of the tool and its requirements. """
@abstractproperty def resolver_type(self) -> str: """Short label for the type of container resolution.""" 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) -> str: return f"{self.__class__.__name__}[]"