Warning
This document is for an in-development version 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.tools.deps.containers
import logging
import os
from galaxy.util import (
asbool,
plugin_config
)
from .container_classes import (
CONTAINER_CLASSES,
DOCKER_CONTAINER_TYPE,
NULL_CONTAINER,
SINGULARITY_CONTAINER_TYPE,
)
from .container_resolvers.explicit import (
ExplicitContainerResolver,
ExplicitSingularityContainerResolver,
)
from .container_resolvers.mulled import (
BuildMulledDockerContainerResolver,
BuildMulledSingularityContainerResolver,
CachedMulledDockerContainerResolver,
CachedMulledSingularityContainerResolver,
MulledDockerContainerResolver,
MulledSingularityContainerResolver,
)
from .requirements import (
ContainerDescription,
)
log = logging.getLogger(__name__)
DEFAULT_CONTAINER_TYPE = DOCKER_CONTAINER_TYPE
ALL_CONTAINER_TYPES = [DOCKER_CONTAINER_TYPE, SINGULARITY_CONTAINER_TYPE]
[docs]class ContainerFinder(object):
[docs] def __init__(self, app_info):
self.app_info = app_info
self.container_registry = ContainerRegistry(app_info)
def _enabled_container_types(self, destination_info):
return [t for t in ALL_CONTAINER_TYPES if self.__container_type_enabled(t, destination_info)]
[docs] def find_best_container_description(self, enabled_container_types, tool_info):
"""Regardless of destination properties - find best container for tool.
Given container types and container.ToolInfo description of the tool."""
container_description = self.container_registry.find_best_container_description(enabled_container_types, tool_info)
return container_description
[docs] def find_container(self, tool_info, destination_info, job_info):
enabled_container_types = self._enabled_container_types(destination_info)
# Short-cut everything else and just skip checks if no container type is enabled.
if not enabled_container_types:
return NULL_CONTAINER
def __destination_container(container_description=None, container_id=None, container_type=None):
if container_description:
container_id = container_description.identifier
container_type = container_description.type
container = self.__destination_container(
container_id,
container_type,
tool_info,
destination_info,
job_info,
container_description,
)
return container
def container_from_description_from_dicts(destination_container_dicts):
for destination_container_dict in destination_container_dicts:
container_description = ContainerDescription.from_dict(destination_container_dict)
if container_description:
container = __destination_container(container_description)
if container:
return container
if "container_override" in destination_info:
container = container_from_description_from_dicts(destination_info["container_override"])
if container:
return container
# If destination forcing Galaxy to use a particular container do it,
# this is likely kind of a corner case. For instance if deployers
# do not trust the containers annotated in tools.
for container_type in CONTAINER_CLASSES.keys():
container_id = self.__overridden_container_id(container_type, destination_info)
if container_id:
container = __destination_container(container_type=container_type, container_id=container_id)
if container:
return container
# Otherwise lets see if we can find container for the tool.
container_description = self.find_best_container_description(enabled_container_types, tool_info)
container = __destination_container(container_description)
if container:
return container
# If we still don't have a container, check to see if any container
# types define a default container id and use that.
if "container" in destination_info:
container = container_from_description_from_dicts(destination_info["container"])
if container:
return container
for container_type in CONTAINER_CLASSES.keys():
container_id = self.__default_container_id(container_type, destination_info)
if container_id:
container = __destination_container(container_type=container_type, container_id=container_id)
if container:
return container
return NULL_CONTAINER
def __overridden_container_id(self, container_type, destination_info):
if not self.__container_type_enabled(container_type, destination_info):
return None
if "%s_container_id_override" % container_type in destination_info:
return destination_info.get("%s_container_id_override" % container_type)
if "%s_image_override" % container_type in destination_info:
return self.__build_container_id_from_parts(container_type, destination_info, mode="override")
def __build_container_id_from_parts(self, container_type, destination_info, mode):
repo = ""
owner = ""
repo_key = "%s_repo_%s" % (container_type, mode)
owner_key = "%s_owner_%s" % (container_type, mode)
if repo_key in destination_info:
repo = destination_info[repo_key] + "/"
if owner_key in destination_info:
owner = destination_info[owner_key] + "/"
cont_id = repo + owner + destination_info["%s_image_%s" % (container_type, mode)]
tag_key = "%s_tag_%s" % (container_type, mode)
if tag_key in destination_info:
cont_id += ":" + destination_info[tag_key]
return cont_id
def __default_container_id(self, container_type, destination_info):
if not self.__container_type_enabled(container_type, destination_info):
return None
key = "%s_default_container_id" % container_type
# Also allow docker_image...
if key not in destination_info:
key = "%s_image" % container_type
if key in destination_info:
return destination_info.get(key)
elif "%s_image_default" in destination_info:
return self.__build_container_id_from_parts(container_type, destination_info, mode="default")
return None
def __destination_container(self, container_id, container_type, tool_info, destination_info, job_info, container_description=None):
# TODO: ensure destination_info is dict-like
if not self.__container_type_enabled(container_type, destination_info):
return NULL_CONTAINER
# TODO: Right now this assumes all containers available when a
# container type is - there should be more thought put into this.
# Checking which are availalbe - settings policies for what can be
# auto-fetched, etc....
return CONTAINER_CLASSES[container_type](container_id, self.app_info, tool_info, destination_info, job_info, container_description)
def __container_type_enabled(self, container_type, destination_info):
return asbool(destination_info.get("%s_enabled" % container_type, False))
[docs]class ContainerRegistry(object):
"""Loop through enabled ContainerResolver plugins and find first match."""
[docs] def __init__(self, app_info):
self.resolver_classes = self.__resolvers_dict()
self.enable_beta_mulled_containers = app_info.enable_beta_mulled_containers
self.app_info = app_info
self.container_resolvers = self.__build_container_resolvers(app_info)
def __build_container_resolvers(self, app_info):
conf_file = getattr(app_info, 'containers_resolvers_config_file', None)
if not conf_file:
return self.__default_containers_resolvers()
if not os.path.exists(conf_file):
log.debug("Unable to find config file '%s'", conf_file)
return self.__default_containers_resolvers()
plugin_source = plugin_config.plugin_source_from_path(conf_file)
return self.__parse_resolver_conf_xml(plugin_source)
def __parse_resolver_conf_xml(self, plugin_source):
extra_kwds = {
'app_info': self.app_info
}
return plugin_config.load_plugins(self.resolver_classes, plugin_source, extra_kwds)
def __default_containers_resolvers(self):
default_resolvers = [
ExplicitContainerResolver(self.app_info),
ExplicitSingularityContainerResolver(self.app_info),
]
if self.enable_beta_mulled_containers:
default_resolvers.extend([
CachedMulledDockerContainerResolver(self.app_info, namespace="biocontainers"),
CachedMulledSingularityContainerResolver(self.app_info, namespace="biocontainers"),
MulledDockerContainerResolver(self.app_info, namespace="biocontainers"),
MulledSingularityContainerResolver(self.app_info, namespace="biocontainers"),
BuildMulledDockerContainerResolver(self.app_info),
BuildMulledSingularityContainerResolver(self.app_info),
])
return default_resolvers
def __resolvers_dict(self):
import galaxy.tools.deps.container_resolvers
return plugin_config.plugins_dict(galaxy.tools.deps.container_resolvers, 'resolver_type')
[docs] def find_best_container_description(self, enabled_container_types, tool_info):
"""Yield best container description of supplied types matching tool info."""
for container_resolver in self.container_resolvers:
if hasattr(container_resolver, "container_type"):
if container_resolver.container_type not in enabled_container_types:
continue
container_description = container_resolver.resolve(enabled_container_types, tool_info)
log.info("Checking with container resolver [%s] found description [%s]" % (container_resolver, container_description))
if container_description:
assert container_description.type in enabled_container_types
return container_description
return None