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 tool_shed.utility_containers
import logging
import threading
from galaxy import util
from galaxy.tool_shed.util import utility_container_manager
from galaxy.tool_shed.util.container_util import (
generate_repository_dependencies_key_for_repository,
STRSEP,
)
from galaxy.util.tool_shed.common_util import parse_repository_dependency_tuple
from tool_shed.util.readme_util import build_readme_files_dict
log = logging.getLogger(__name__)
[docs]class FailedTest:
"""Failed tool tests object"""
[docs] def __init__(self, id=None, stderr=None, test_id=None, tool_id=None, tool_version=None, traceback=None):
self.id = id
self.stderr = stderr
self.test_id = test_id
self.tool_id = tool_id
self.tool_version = tool_version
self.traceback = traceback
[docs]class InvalidRepositoryDependency:
"""Invalid repository dependency definition object"""
[docs] def __init__(
self,
id=None,
toolshed=None,
repository_name=None,
repository_owner=None,
changeset_revision=None,
prior_installation_required=False,
only_if_compiling_contained_td=False,
error=None,
):
self.id = id
self.toolshed = toolshed
self.repository_name = repository_name
self.repository_owner = repository_owner
self.changeset_revision = changeset_revision
self.prior_installation_required = prior_installation_required
self.only_if_compiling_contained_td = only_if_compiling_contained_td
self.error = error
[docs]class InvalidToolDependency:
"""Invalid tool dependency definition object"""
[docs] def __init__(self, id=None, name=None, version=None, type=None, error=None):
self.id = id
self.name = name
self.version = version
self.type = type
self.error = error
[docs]class MissingTestComponent:
"""Missing tool test components object"""
[docs] def __init__(self, id=None, missing_components=None, tool_guid=None, tool_id=None, tool_version=None):
self.id = id
self.missing_components = missing_components
self.tool_guid = tool_guid
self.tool_id = tool_id
self.tool_version = tool_version
[docs]class PassedTest:
"""Passed tool tests object"""
[docs] def __init__(self, id=None, test_id=None, tool_id=None, tool_version=None):
self.id = id
self.test_id = test_id
self.tool_id = tool_id
self.tool_version = tool_version
[docs]class RepositoryInstallationError:
"""Repository installation error object"""
[docs] def __init__(self, id=None, tool_shed=None, name=None, owner=None, changeset_revision=None, error_message=None):
self.id = id
self.tool_shed = tool_shed
self.name = name
self.owner = owner
self.changeset_revision = changeset_revision
self.error_message = error_message
[docs]class RepositorySuccessfulInstallation:
"""Repository installation object"""
[docs] def __init__(self, id=None, tool_shed=None, name=None, owner=None, changeset_revision=None):
self.id = id
self.tool_shed = tool_shed
self.name = name
self.owner = owner
self.changeset_revision = changeset_revision
[docs]class ToolDependencyInstallationError:
"""Tool dependency installation error object"""
[docs] def __init__(self, id=None, type=None, name=None, version=None, error_message=None):
self.id = id
self.type = type
self.name = name
self.version = version
self.error_message = error_message
[docs]class ToolDependencySuccessfulInstallation:
"""Tool dependency installation object"""
[docs] def __init__(self, id=None, type=None, name=None, version=None, installation_directory=None):
self.id = id
self.type = type
self.name = name
self.version = version
self.installation_directory = installation_directory
[docs]class ToolShedUtilityContainerManager(utility_container_manager.UtilityContainerManager):
[docs] def build_invalid_repository_dependencies_root_folder(self, folder_id, invalid_repository_dependencies_dict):
"""Return a folder hierarchy containing invalid repository dependencies."""
label = "Invalid repository dependencies"
if invalid_repository_dependencies_dict:
invalid_repository_dependency_id = 0
folder_id += 1
invalid_repository_dependencies_root_folder = utility_container_manager.Folder(
id=folder_id, key="root", label="root", parent=None
)
folder_id += 1
invalid_repository_dependencies_folder = utility_container_manager.Folder(
id=folder_id,
key="invalid_repository_dependencies",
label=label,
parent=invalid_repository_dependencies_root_folder,
)
invalid_repository_dependencies_root_folder.folders.append(invalid_repository_dependencies_folder)
invalid_repository_dependencies = invalid_repository_dependencies_dict["repository_dependencies"]
for invalid_repository_dependency in invalid_repository_dependencies:
folder_id += 1
invalid_repository_dependency_id += 1
(
toolshed,
name,
owner,
changeset_revision,
prior_installation_required,
only_if_compiling_contained_td,
error,
) = parse_repository_dependency_tuple(invalid_repository_dependency, contains_error=True)
key = generate_repository_dependencies_key_for_repository(
toolshed,
name,
owner,
changeset_revision,
prior_installation_required,
only_if_compiling_contained_td,
)
label = f"Repository <b>{name}</b> revision <b>{changeset_revision}</b> owned by <b>{owner}</b>"
folder = utility_container_manager.Folder(
id=folder_id, key=key, label=label, parent=invalid_repository_dependencies_folder
)
ird = InvalidRepositoryDependency(
id=invalid_repository_dependency_id,
toolshed=toolshed,
repository_name=name,
repository_owner=owner,
changeset_revision=changeset_revision,
prior_installation_required=util.asbool(prior_installation_required),
only_if_compiling_contained_td=util.asbool(only_if_compiling_contained_td),
error=error,
)
folder.invalid_repository_dependencies.append(ird)
invalid_repository_dependencies_folder.folders.append(folder)
else:
invalid_repository_dependencies_root_folder = None
return folder_id, invalid_repository_dependencies_root_folder
[docs] def build_invalid_tool_dependencies_root_folder(self, folder_id, invalid_tool_dependencies_dict):
"""Return a folder hierarchy containing invalid tool dependencies."""
# # INvalid tool dependencies are always packages like:
# {"R/2.15.1": {"name": "R", "readme": "some string", "type": "package", "version": "2.15.1" "error" : "some sting" }
label = "Invalid tool dependencies"
if invalid_tool_dependencies_dict:
invalid_tool_dependency_id = 0
folder_id += 1
invalid_tool_dependencies_root_folder = utility_container_manager.Folder(
id=folder_id, key="root", label="root", parent=None
)
folder_id += 1
invalid_tool_dependencies_folder = utility_container_manager.Folder(
id=folder_id, key="invalid_tool_dependencies", label=label, parent=invalid_tool_dependencies_root_folder
)
invalid_tool_dependencies_root_folder.folders.append(invalid_tool_dependencies_folder)
for requirements_dict in invalid_tool_dependencies_dict.values():
folder_id += 1
invalid_tool_dependency_id += 1
try:
name = requirements_dict["name"]
type = requirements_dict["type"]
version = requirements_dict["version"]
error = requirements_dict["error"]
except Exception as e:
name = "unknown"
type = "unknown"
version = "unknown"
error = str(e)
key = self.generate_tool_dependencies_key(name, version, type)
label = f"Version <b>{version}</b> of the <b>{name}</b> <b>{type}</b>"
folder = utility_container_manager.Folder(
id=folder_id, key=key, label=label, parent=invalid_tool_dependencies_folder
)
itd = InvalidToolDependency(
id=invalid_tool_dependency_id, name=name, version=version, type=type, error=error
)
folder.invalid_tool_dependencies.append(itd)
invalid_tool_dependencies_folder.folders.append(folder)
else:
invalid_tool_dependencies_root_folder = None
return folder_id, invalid_tool_dependencies_root_folder
[docs] def build_repository_containers(
self, repository, changeset_revision, repository_dependencies, repository_metadata, exclude=None
):
"""
Return a dictionary of containers for the received repository's dependencies and
contents for display in the Tool Shed.
"""
if exclude is None:
exclude = []
containers_dict = dict(
datatypes=None,
invalid_tools=None,
readme_files=None,
repository_dependencies=None,
tool_dependencies=None,
valid_tools=None,
valid_data_managers=None,
)
if repository_metadata:
metadata = repository_metadata.metadata
lock = threading.Lock()
lock.acquire(True)
try:
folder_id = 0
# Invalid repository dependencies container.
if metadata:
if (
"invalid_repository_dependencies" not in exclude
and "invalid_repository_dependencies" in metadata
):
invalid_repository_dependencies = metadata["invalid_repository_dependencies"]
(
folder_id,
invalid_repository_dependencies_root_folder,
) = self.build_invalid_repository_dependencies_root_folder(
folder_id, invalid_repository_dependencies
)
containers_dict["invalid_repository_dependencies"] = invalid_repository_dependencies_root_folder
# Invalid tool dependencies container.
if metadata:
if "invalid_tool_dependencies" not in exclude and "invalid_tool_dependencies" in metadata:
invalid_tool_dependencies = metadata["invalid_tool_dependencies"]
(
folder_id,
invalid_tool_dependencies_root_folder,
) = self.build_invalid_tool_dependencies_root_folder(folder_id, invalid_tool_dependencies)
containers_dict["invalid_tool_dependencies"] = invalid_tool_dependencies_root_folder
# Invalid tools container.
if metadata:
if "invalid_tools" not in exclude and "invalid_tools" in metadata:
invalid_tool_configs = metadata["invalid_tools"]
folder_id, invalid_tools_root_folder = self.build_invalid_tools_folder(
folder_id,
invalid_tool_configs,
changeset_revision,
repository=repository,
label="Invalid tools",
)
containers_dict["invalid_tools"] = invalid_tools_root_folder
# Readme files container.
if metadata:
if "readme_files" not in exclude and "readme_files" in metadata:
readme_files_dict = build_readme_files_dict(self.app, repository, changeset_revision, metadata)
folder_id, readme_files_root_folder = self.build_readme_files_folder(
folder_id, readme_files_dict
)
containers_dict["readme_files"] = readme_files_root_folder
if "repository_dependencies" not in exclude:
# Repository dependencies container.
folder_id, repository_dependencies_root_folder = self.build_repository_dependencies_folder(
folder_id=folder_id,
repository_dependencies=repository_dependencies,
label="Repository dependencies",
installed=False,
)
if repository_dependencies_root_folder:
containers_dict["repository_dependencies"] = repository_dependencies_root_folder
# Tool dependencies container.
if metadata:
if "tool_dependencies" not in exclude and "tool_dependencies" in metadata:
tool_dependencies = metadata["tool_dependencies"]
if "orphan_tool_dependencies" in metadata:
# The use of the orphan_tool_dependencies category in metadata has been deprecated,
# but we still need to check in case the metadata is out of date.
orphan_tool_dependencies = metadata["orphan_tool_dependencies"]
tool_dependencies.update(orphan_tool_dependencies)
# Tool dependencies can be categorized as orphans only if the repository contains tools.
if "tools" not in exclude:
tools = metadata.get("tools", [])
tools.extend(metadata.get("invalid_tools", []))
folder_id, tool_dependencies_root_folder = self.build_tool_dependencies_folder(
folder_id, tool_dependencies, missing=False, new_install=False
)
containers_dict["tool_dependencies"] = tool_dependencies_root_folder
# Valid tools container.
if metadata:
if "tools" not in exclude and "tools" in metadata:
valid_tools = metadata["tools"]
folder_id, valid_tools_root_folder = self.build_tools_folder(
folder_id, valid_tools, repository, changeset_revision, label="Valid tools"
)
containers_dict["valid_tools"] = valid_tools_root_folder
# Valid Data Managers container
if metadata:
if "data_manager" not in exclude and "data_manager" in metadata:
data_managers = metadata["data_manager"].get("data_managers", None)
folder_id, data_managers_root_folder = self.build_data_managers_folder(
folder_id, data_managers, label="Data Managers"
)
containers_dict["valid_data_managers"] = data_managers_root_folder
error_messages = metadata["data_manager"].get("error_messages", None)
data_managers = metadata["data_manager"].get("invalid_data_managers", None)
folder_id, data_managers_root_folder = self.build_invalid_data_managers_folder(
folder_id, data_managers, error_messages, label="Invalid Data Managers"
)
containers_dict["invalid_data_managers"] = data_managers_root_folder
except Exception:
log.exception("Exception in build_repository_containers")
finally:
lock.release()
return containers_dict
[docs] def generate_tool_dependencies_key(self, name, version, type):
return f"{str(name)}{STRSEP}{str(version)}{STRSEP}{str(type)}"