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 import UNKNOWN
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)}"