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__}[]"