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.utility_container_manager
import logging
from galaxy import util
from tool_shed.util import common_util
from tool_shed.util import container_util
from tool_shed.util import repository_util
log = logging.getLogger(__name__)
[docs]class Folder(object):
"""Container object."""
[docs] def __init__(self, id=None, key=None, label=None, parent=None):
self.id = id
self.key = key
self.label = label
self.parent = parent
self.current_repository_installation_errors = []
self.current_repository_successful_installations = []
self.description = None
self.datatypes = []
self.folders = []
self.invalid_data_managers = []
self.invalid_repository_dependencies = []
self.invalid_tool_dependencies = []
self.invalid_tools = []
self.missing_test_components = []
self.readme_files = []
self.repository_dependencies = []
self.repository_installation_errors = []
self.repository_successful_installations = []
self.test_environments = []
self.tool_dependencies = []
self.valid_tools = []
self.valid_data_managers = []
self.workflows = []
[docs] def contains_folder(self, folder):
for index, contained_folder in enumerate(self.folders):
if folder == contained_folder:
return index, contained_folder
return 0, None
[docs] def contains_repository_dependency(self, repository_dependency):
listified_repository_dependency = repository_dependency.listify
for contained_repository_dependency in self.repository_dependencies:
if contained_repository_dependency.listify == listified_repository_dependency:
return True
return False
[docs] def remove_repository_dependency(self, repository_dependency):
listified_repository_dependency = repository_dependency.listify
for contained_repository_dependency in self.repository_dependencies:
if contained_repository_dependency.listify == listified_repository_dependency:
self.repository_dependencies.remove(contained_repository_dependency)
[docs]class DataManager(object):
"""Data Manager object"""
[docs] def __init__(self, id=None, name=None, version=None, data_tables=None):
self.id = id
self.name = name
self.version = version
self.data_tables = data_tables
[docs]class Datatype(object):
"""Datatype object"""
[docs] def __init__(self, id=None, extension=None, type=None, mimetype=None, subclass=None, converters=None, display_app_containers=None):
self.id = id
self.extension = extension
self.type = type
self.mimetype = mimetype
self.subclass = subclass
self.converters = converters
self.display_app_containers = display_app_containers
[docs]class InvalidDataManager(object):
"""Invalid data Manager object"""
[docs] def __init__(self, id=None, index=None, error=None):
self.id = id
self.index = index
self.error = error
[docs]class InvalidTool(object):
"""Invalid tool object"""
[docs] def __init__(self, id=None, tool_config=None, repository_id=None, changeset_revision=None, repository_installation_status=None):
self.id = id
self.tool_config = tool_config
self.repository_id = repository_id
self.changeset_revision = changeset_revision
self.repository_installation_status = repository_installation_status
[docs]class ReadMe(object):
"""Readme text object"""
[docs] def __init__(self, id=None, name=None, text=None):
self.id = id
self.name = name
self.text = text
[docs]class RepositoryDependency(object):
"""Repository dependency 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, installation_status=None, tool_shed_repository_id=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.installation_status = installation_status
self.tool_shed_repository_id = tool_shed_repository_id
@property
def listify(self):
return [self.toolshed,
self.repository_name,
self.repository_owner,
self.changeset_revision,
self.prior_installation_required,
self.only_if_compiling_contained_td]
[docs]class Tool(object):
"""Tool object"""
[docs] def __init__(self, id=None, tool_config=None, tool_id=None, name=None, description=None, version=None, profile=None, requirements=None,
repository_id=None, changeset_revision=None, repository_installation_status=None):
self.id = id
self.tool_config = tool_config
self.tool_id = tool_id
self.name = name
self.description = description
self.version = version
self.profile = profile
self.requirements = requirements
self.repository_id = repository_id
self.changeset_revision = changeset_revision
self.repository_installation_status = repository_installation_status
[docs]class ToolDependency(object):
"""Tool dependency object"""
[docs] def __init__(self, id=None, name=None, version=None, type=None, readme=None, installation_status=None, repository_id=None,
tool_dependency_id=None):
self.id = id
self.name = name
self.version = version
self.type = type
self.readme = readme
self.installation_status = installation_status
self.repository_id = repository_id
self.tool_dependency_id = tool_dependency_id
@property
def listify(self):
return [self.name, self.version, self.type]
[docs]class Workflow(object):
"""Workflow object."""
[docs] def __init__(self, id=None, workflow_name=None, steps=None, format_version=None, annotation=None,
repository_metadata_id=None, repository_id=None):
"""
When rendered in the tool shed, repository_metadata_id will have a value and repository_id will
be None. When rendered in Galaxy, repository_id will have a value and repository_metadata_id will
be None.
"""
self.id = id
self.workflow_name = workflow_name
self.steps = steps
self.format_version = format_version
self.annotation = annotation
self.repository_metadata_id = repository_metadata_id
self.repository_id = repository_id
[docs]class UtilityContainerManager(object):
[docs] def build_data_managers_folder(self, folder_id, data_managers, label=None):
"""Return a folder hierarchy containing Data Managers."""
if data_managers:
if label is None:
label = "Data Managers"
data_manager_id = 0
folder_id += 1
data_managers_root_folder = Folder(id=folder_id, key='root', label='root', parent=None)
folder_id += 1
key = "valid_data_managers"
folder = Folder(id=folder_id, key=key, label=label, parent=data_managers_root_folder)
data_managers_root_folder.folders.append(folder)
# Insert a header row.
data_manager_id += 1
data_manager = DataManager(id=data_manager_id,
name='Name',
version='Version',
data_tables='Data Tables')
folder.valid_data_managers.append(data_manager)
for data_manager_dict in data_managers.values():
data_manager_id += 1
try:
name = data_manager_dict.get('name', '')
version = data_manager_dict.get('version', '')
data_tables = ", ".join(data_manager_dict.get('data_tables', ''))
except Exception as e:
name = str(e)
version = 'unknown'
data_tables = 'unknown'
data_manager = DataManager(id=data_manager_id,
name=name,
version=version,
data_tables=data_tables)
folder.valid_data_managers.append(data_manager)
else:
data_managers_root_folder = None
return folder_id, data_managers_root_folder
[docs] def build_datatypes_folder(self, folder_id, datatypes, label='Datatypes'):
"""Return a folder hierarchy containing datatypes."""
if datatypes:
datatype_id = 0
folder_id += 1
datatypes_root_folder = Folder(id=folder_id, key='root', label='root', parent=None)
folder_id += 1
folder = Folder(id=folder_id, key='datatypes', label=label, parent=datatypes_root_folder)
datatypes_root_folder.folders.append(folder)
# Insert a header row.
datatype_id += 1
datatype = Datatype(id=datatype_id,
extension='extension',
type='type',
mimetype='mimetype',
subclass='subclass')
folder.datatypes.append(datatype)
for datatypes_dict in datatypes:
# {"converters":
# [{"target_datatype": "gff",
# "tool_config": "bed_to_gff_converter.xml",
# "guid": "localhost:9009/repos/test/bed_to_gff_converter/CONVERTER_bed_to_gff_0/2.0.0"}],
# "display_in_upload": "true",
# "dtype": "galaxy.datatypes.interval:Bed",
# "extension": "bed"}
# TODO: converters and display_app information is not currently rendered. Should it be?
# Handle defined converters, if any.
converters = datatypes_dict.get('converters', None)
if converters:
num_converters = len(converters)
else:
num_converters = 0
# Handle defined display applications, if any.
display_app_containers = datatypes_dict.get('display_app_containers', None)
if display_app_containers:
num_display_app_containers = len(display_app_containers)
else:
num_display_app_containers = 0
datatype_id += 1
try:
extension = datatypes_dict.get('extension', '')
type = datatypes_dict.get('dtype', '')
mimetype = datatypes_dict.get('mimetype', '')
subclass = datatypes_dict.get('subclass', '')
converters = num_converters
display_app_containers = num_display_app_containers
except Exception as e:
extension = str(e)
type = 'unknown'
mimetype = 'unknown'
subclass = 'unknown'
converters = 'unknown'
display_app_containers = 'unknown'
datatype = Datatype(id=datatype_id,
extension=extension,
type=type,
mimetype=mimetype,
subclass=subclass,
converters=converters,
display_app_containers=display_app_containers)
folder.datatypes.append(datatype)
else:
datatypes_root_folder = None
return folder_id, datatypes_root_folder
[docs] def build_invalid_data_managers_folder(self, folder_id, data_managers, error_messages=None, label=None):
"""Return a folder hierarchy containing invalid Data Managers."""
if data_managers or error_messages:
if label is None:
label = "Invalid Data Managers"
data_manager_id = 0
folder_id += 1
data_managers_root_folder = Folder(id=folder_id, key='root', label='root', parent=None)
folder_id += 1
key = "invalid_data_managers"
folder = Folder(id=folder_id, key=key, label=label, parent=data_managers_root_folder)
data_managers_root_folder.folders.append(folder)
# Insert a header row.
data_manager_id += 1
data_manager = InvalidDataManager(id=data_manager_id,
index='Element Index',
error='Error')
folder.invalid_data_managers.append(data_manager)
if error_messages:
for error_message in error_messages:
data_manager_id += 1
data_manager = InvalidDataManager(id=data_manager_id,
index=0,
error=error_message)
folder.invalid_data_managers.append(data_manager)
for data_manager_dict in data_managers:
data_manager_id += 1
data_manager = InvalidDataManager(id=data_manager_id,
index=data_manager_dict.get('index', 0) + 1,
error=data_manager_dict.get('error_message', ''))
folder.invalid_data_managers.append(data_manager)
else:
data_managers_root_folder = None
return folder_id, data_managers_root_folder
[docs] def build_invalid_tools_folder(self, folder_id, invalid_tool_configs, changeset_revision, repository=None,
label='Invalid tools'):
"""Return a folder hierarchy containing invalid tools."""
# TODO: Should we display invalid tools on the tool panel selection page when installing the
# repository into Galaxy?
if invalid_tool_configs:
invalid_tool_id = 0
folder_id += 1
invalid_tools_root_folder = Folder(id=folder_id, key='root', label='root', parent=None)
folder_id += 1
folder = Folder(id=folder_id, key='invalid_tools', label=label, parent=invalid_tools_root_folder)
invalid_tools_root_folder.folders.append(folder)
for invalid_tool_config in invalid_tool_configs:
invalid_tool_id += 1
if repository:
repository_id = repository.id
if self.app.name == 'galaxy':
repository_installation_status = repository.status
else:
repository_installation_status = None
else:
repository_id = None
repository_installation_status = None
invalid_tool = InvalidTool(id=invalid_tool_id,
tool_config=invalid_tool_config,
repository_id=repository_id,
changeset_revision=changeset_revision,
repository_installation_status=repository_installation_status)
folder.invalid_tools.append(invalid_tool)
else:
invalid_tools_root_folder = None
return folder_id, invalid_tools_root_folder
[docs] def build_readme_files_folder(self, folder_id, readme_files_dict, label='Readme files'):
"""Return a folder hierarchy containing readme text files."""
if readme_files_dict:
readme_id = 0
folder_id += 1
readme_files_root_folder = Folder(id=folder_id, key='root', label='root', parent=None)
folder_id += 1
readme_files_folder = Folder(id=folder_id, key='readme_files', label=label, parent=readme_files_root_folder)
multiple_readme_files = len(readme_files_dict) > 1
readme_files_root_folder.folders.append(readme_files_folder)
for readme_file_name, readme_file_text in readme_files_dict.items():
readme_id += 1
readme = ReadMe(id=readme_id, name=readme_file_name, text=readme_file_text)
if multiple_readme_files:
folder_id += 1
folder = Folder(id=folder_id, key=readme_file_name, label=readme_file_name, parent=readme_files_folder)
folder.readme_files.append(readme)
readme_files_folder.folders.append(folder)
else:
readme_files_folder.readme_files.append(readme)
else:
readme_files_root_folder = None
return folder_id, readme_files_root_folder
[docs] def build_repository_dependencies_folder(self, folder_id, repository_dependencies, label='Repository dependencies',
installed=False):
"""Return a folder hierarchy containing repository dependencies."""
if repository_dependencies:
repository_dependency_id = 0
folder_id += 1
# Create the root folder.
repository_dependencies_root_folder = Folder(id=folder_id, key='root', label='root', parent=None)
folder_id += 1
# Create the Repository dependencies folder and add it to the root folder.
repository_dependencies_folder_key = repository_dependencies['root_key']
repository_dependencies_folder = Folder(id=folder_id,
key=repository_dependencies_folder_key,
label=label,
parent=repository_dependencies_root_folder)
del repository_dependencies['root_key']
# The received repository_dependencies is a dictionary with keys: 'root_key', 'description', and one or more
# repository_dependency keys. We want the description value associated with the repository_dependencies_folder.
repository_dependencies_folder.description = repository_dependencies.get('description', None)
repository_dependencies_root_folder.folders.append(repository_dependencies_folder)
del repository_dependencies['description']
repository_dependencies_folder, folder_id, repository_dependency_id = \
self.populate_repository_dependencies_container(repository_dependencies_folder,
repository_dependencies,
folder_id,
repository_dependency_id)
repository_dependencies_folder = self.prune_repository_dependencies(repository_dependencies_folder)
else:
repository_dependencies_root_folder = None
return folder_id, repository_dependencies_root_folder
[docs] def build_tools_folder(self, folder_id, tool_dicts, repository, changeset_revision, valid=True, label='Valid tools'):
"""Return a folder hierarchy containing valid tools."""
if tool_dicts:
container_object_tool_id = 0
folder_id += 1
tools_root_folder = Folder(id=folder_id, key='root', label='root', parent=None)
folder_id += 1
folder = Folder(id=folder_id, key='tools', label=label, parent=tools_root_folder)
if self.app.name == 'galaxy':
folder.description = 'click the name to inspect the tool metadata'
tools_root_folder.folders.append(folder)
# Insert a header row.
container_object_tool_id += 1
tool = Tool(id=container_object_tool_id,
tool_config='',
tool_id='',
name='Name',
description='Description',
version='Version',
profile='Minimum Galaxy Version',
requirements='',
repository_id='',
changeset_revision='')
folder.valid_tools.append(tool)
if repository:
repository_id = repository.id
if self.app.name == 'galaxy':
repository_installation_status = repository.status
else:
repository_installation_status = None
else:
repository_id = None
repository_installation_status = None
for tool_dict in tool_dicts:
if not isinstance(tool_dict, dict):
# Due to some previous bug (hopefully not current) invalid tool strings may be included in the received
# list of tool_dicts. For example, the picard repository metadata has 2 invalid tools in the received
# list of supposedly valid tools: 'rgPicardASMetrics.xml', 'rgPicardGCBiasMetrics.xml'.
continue
container_object_tool_id += 1
requirements = tool_dict.get('requirements', None)
if requirements is not None:
# 'requirements': [{'version': '1.56.0', 'type': 'package', 'name': 'picard'}],
requirements_str = ''
for requirement_dict in requirements:
try:
requirement_name = str(requirement_dict.get('name', 'unknown'))
requirement_type = str(requirement_dict.get('type', 'unknown'))
except Exception as e:
requirement_name = str(e)
requirement_type = 'unknown'
requirements_str += '%s (%s), ' % (requirement_name, requirement_type)
requirements_str = requirements_str.rstrip(', ')
else:
requirements_str = 'none'
try:
tool_config = str(tool_dict.get('tool_config', 'missing'))
tool_id = str(tool_dict.get('id', 'unknown'))
name = str(tool_dict.get('name', 'unknown'))
description = str(tool_dict.get('description', ''))
version = str(tool_dict.get('version', 'unknown'))
profile = str(tool_dict.get('profile', 'any'))
except Exception as e:
tool_config = str(e)
tool_id = 'unknown'
name = 'unknown'
description = ''
version = 'unknown'
tool = Tool(id=container_object_tool_id,
tool_config=tool_config,
tool_id=tool_id,
name=name,
description=description,
version=version,
profile=profile,
requirements=requirements_str,
repository_id=repository_id,
changeset_revision=changeset_revision,
repository_installation_status=repository_installation_status)
folder.valid_tools.append(tool)
else:
tools_root_folder = None
return folder_id, tools_root_folder
[docs] def build_tool_dependencies_folder(self, folder_id, tool_dependencies, label='Tool dependencies', missing=False,
new_install=False, reinstalling=False):
"""Return a folder hierarchy containing tool dependencies."""
# When we're in Galaxy (not the tool shed) and the tool dependencies are not installed or are in an error state,
# they are considered missing. The tool dependency status will be displayed only if a record exists for the tool
# dependency in the Galaxy database, but the tool dependency is not installed. The value for new_install will be
# True only if the associated repository in being installed for the first time. This value is used in setting the
# container description.
if tool_dependencies:
tool_dependency_id = 0
folder_id += 1
tool_dependencies_root_folder = Folder(id=folder_id, key='root', label='root', parent=None)
folder_id += 1
folder = Folder(id=folder_id, key='tool_dependencies', label=label, parent=tool_dependencies_root_folder)
if self.app.name == 'galaxy':
if new_install or reinstalling:
folder.description = "repository tools require handling of these dependencies"
elif missing and not new_install and not reinstalling:
folder.description = 'click the name to install the missing dependency'
else:
folder.description = 'click the name to browse the dependency installation directory'
tool_dependencies_root_folder.folders.append(folder)
# Insert a header row.
tool_dependency_id += 1
if self.app.name == 'galaxy':
tool_dependency = ToolDependency(id=tool_dependency_id,
name='Name',
version='Version',
type='Type',
readme=None,
installation_status='Installation status',
repository_id=None,
tool_dependency_id=None)
else:
tool_dependency = ToolDependency(id=tool_dependency_id,
name='Name',
version='Version',
type='Type',
readme=None,
installation_status=None,
repository_id=None,
tool_dependency_id=None)
folder.tool_dependencies.append(tool_dependency)
for dependency_key, requirements_dict in tool_dependencies.items():
tool_dependency_id += 1
if dependency_key in ['set_environment']:
for set_environment_dict in requirements_dict:
try:
name = set_environment_dict.get('name', None)
type = set_environment_dict['type']
repository_id = set_environment_dict.get('repository_id', None)
td_id = set_environment_dict.get('tool_dependency_id', None)
except Exception as e:
name = str(e)
type = 'unknown'
repository_id = 'unknown'
td_id = 'unknown'
if self.app.name == 'galaxy':
try:
installation_status = set_environment_dict.get('status', 'Never installed')
except Exception as e:
installation_status = str(e)
else:
installation_status = None
tool_dependency = ToolDependency(id=tool_dependency_id,
name=name,
version=None,
type=type,
readme=None,
installation_status=installation_status,
repository_id=repository_id,
tool_dependency_id=td_id)
folder.tool_dependencies.append(tool_dependency)
else:
try:
name = requirements_dict['name']
version = requirements_dict['version']
type = requirements_dict['type']
repository_id = requirements_dict.get('repository_id', None)
td_id = requirements_dict.get('tool_dependency_id', None)
except Exception as e:
name = str(e)
version = 'unknown'
type = 'unknown'
repository_id = 'unknown'
td_id = 'unknown'
if self.app.name == 'galaxy':
try:
installation_status = requirements_dict.get('status', 'Never installed')
except Exception as e:
installation_status = str(e)
else:
installation_status = None
tool_dependency = ToolDependency(id=tool_dependency_id,
name=name,
version=version,
type=type,
readme=None,
installation_status=installation_status,
repository_id=repository_id,
tool_dependency_id=td_id)
folder.tool_dependencies.append(tool_dependency)
else:
tool_dependencies_root_folder = None
return folder_id, tool_dependencies_root_folder
[docs] def build_workflows_folder(self, folder_id, workflows, repository_metadata_id=None, repository_id=None,
label='Workflows'):
"""
Return a folder hierarchy containing workflow objects for each workflow dictionary in the
received workflows list. When this method is called from the tool shed, repository_metadata_id
will have a value and repository_id will be None. When this method is called from Galaxy,
repository_id will have a value only if the repository is not currenlty being installed and
repository_metadata_id will be None.
"""
if workflows:
workflow_id = 0
folder_id += 1
workflows_root_folder = Folder(id=folder_id, key='root', label='root', parent=None)
folder_id += 1
folder = Folder(id=folder_id, key='workflows', label=label, parent=workflows_root_folder)
workflows_root_folder.folders.append(folder)
# Insert a header row.
workflow_id += 1
workflow = Workflow(id=workflow_id,
workflow_name='Name',
steps='steps',
format_version='format-version',
annotation='annotation',
repository_metadata_id=repository_metadata_id,
repository_id=repository_id)
folder.workflows.append(workflow)
for workflow_tup in workflows:
workflow_dict = workflow_tup[1]
steps = workflow_dict.get('steps', [])
if steps:
steps = str(len(steps))
else:
steps = 'unknown'
workflow_id += 1
workflow = Workflow(id=workflow_id,
workflow_name=workflow_dict.get('name', ''),
steps=steps,
format_version=workflow_dict.get('format-version', ''),
annotation=workflow_dict.get('annotation', ''),
repository_metadata_id=repository_metadata_id,
repository_id=repository_id)
folder.workflows.append(workflow)
else:
workflows_root_folder = None
return folder_id, workflows_root_folder
[docs] def generate_repository_dependencies_folder_label_from_key(self, repository_name, repository_owner, changeset_revision,
prior_installation_required, only_if_compiling_contained_td, key):
"""Return a repository dependency label based on the repository dependency key."""
if self.key_is_current_repositorys_key(repository_name,
repository_owner,
changeset_revision,
prior_installation_required,
only_if_compiling_contained_td, key):
label = 'Repository dependencies'
else:
if util.asbool(prior_installation_required):
prior_installation_required_str = " <i>(prior install required)</i>"
else:
prior_installation_required_str = ""
label = "Repository <b>%s</b> revision <b>%s</b> owned by <b>%s</b>%s" % \
(repository_name, changeset_revision, repository_owner, prior_installation_required_str)
return label
[docs] def get_components_from_repository_dependency_for_installed_repository(self, repository_dependency):
"""
Parse a repository dependency and return components necessary for proper display
in Galaxy on the Manage repository page.
"""
# Default prior_installation_required and only_if_compiling_contained_td to False.
prior_installation_required = 'False'
only_if_compiling_contained_td = 'False'
if len(repository_dependency) == 6:
# Metadata should have been reset on this installed repository, but it wasn't.
tool_shed_repository_id = repository_dependency[4]
installation_status = repository_dependency[5]
tool_shed, name, owner, changeset_revision = repository_dependency[0:4]
repository_dependency = [tool_shed, name, owner, changeset_revision, prior_installation_required, only_if_compiling_contained_td]
elif len(repository_dependency) == 7:
# We have a repository dependency tuple that includes a prior_installation_required value but not a only_if_compiling_contained_td value.
tool_shed_repository_id = repository_dependency[5]
installation_status = repository_dependency[6]
tool_shed, name, owner, changeset_revision, prior_installation_required = repository_dependency[0:5]
repository_dependency = \
[tool_shed, name, owner, changeset_revision, prior_installation_required, only_if_compiling_contained_td]
elif len(repository_dependency) == 8:
# We have a repository dependency tuple that includes both a prior_installation_required value
# and a only_if_compiling_contained_td value.
tool_shed_repository_id = repository_dependency[6]
installation_status = repository_dependency[7]
repository_dependency = repository_dependency[0:6]
else:
tool_shed_repository_id = None
installation_status = 'unknown'
if tool_shed_repository_id:
tool_shed_repository = repository_util.get_tool_shed_repository_by_id(self.app,
self.app.security.encode_id(tool_shed_repository_id))
if tool_shed_repository:
if tool_shed_repository.missing_repository_dependencies:
installation_status = '%s, missing repository dependencies' % installation_status
elif tool_shed_repository.missing_tool_dependencies:
installation_status = '%s, missing tool dependencies' % installation_status
return tool_shed_repository_id, installation_status, repository_dependency
[docs] def get_folder(self, folder, key):
if folder.key == key:
return folder
for sub_folder in folder.folders:
return self.get_folder(sub_folder, key)
return None
[docs] def handle_repository_dependencies_container_entry(self, repository_dependencies_folder, rd_key, rd_value, folder_id,
repository_dependency_id, folder_keys):
repository_components_tuple = container_util.get_components_from_key(rd_key)
components_list = repository_util.extract_components_from_tuple(repository_components_tuple)
toolshed, repository_name, repository_owner, changeset_revision = components_list[0:4]
# For backward compatibility to the 12/20/12 Galaxy release.
if len(components_list) == 4:
prior_installation_required = 'False'
only_if_compiling_contained_td = 'False'
elif len(components_list) == 5:
prior_installation_required = components_list[4]
only_if_compiling_contained_td = 'False'
elif len(components_list) == 6:
prior_installation_required = components_list[4]
only_if_compiling_contained_td = components_list[5]
folder = self.get_folder(repository_dependencies_folder, rd_key)
label = self.generate_repository_dependencies_folder_label_from_key(repository_name,
repository_owner,
changeset_revision,
prior_installation_required,
only_if_compiling_contained_td,
repository_dependencies_folder.key)
if folder:
if rd_key not in folder_keys:
folder_id += 1
sub_folder = Folder(id=folder_id, key=rd_key, label=label, parent=folder)
folder.folders.append(sub_folder)
else:
sub_folder = folder
else:
folder_id += 1
sub_folder = Folder(id=folder_id, key=rd_key, label=label, parent=repository_dependencies_folder)
repository_dependencies_folder.folders.append(sub_folder)
if self.app.name == 'galaxy':
# Insert a header row.
repository_dependency_id += 1
repository_dependency = RepositoryDependency(id=repository_dependency_id,
repository_name='Name',
changeset_revision='Revision',
repository_owner='Owner',
installation_status='Installation status')
# Insert the header row into the folder.
sub_folder.repository_dependencies.append(repository_dependency)
for repository_dependency in rd_value:
if self.app.name == 'galaxy':
tool_shed_repository_id, installation_status, repository_dependency = \
self.get_components_from_repository_dependency_for_installed_repository(repository_dependency)
else:
tool_shed_repository_id = None
installation_status = None
can_create_dependency = not self.is_subfolder_of(sub_folder, repository_dependency)
if can_create_dependency:
toolshed, repository_name, repository_owner, changeset_revision, prior_installation_required, only_if_compiling_contained_td = \
common_util.parse_repository_dependency_tuple(repository_dependency)
repository_dependency_id += 1
repository_dependency = RepositoryDependency(id=repository_dependency_id,
toolshed=toolshed,
repository_name=repository_name,
repository_owner=repository_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),
installation_status=installation_status,
tool_shed_repository_id=tool_shed_repository_id)
# Insert the repository_dependency into the folder.
sub_folder.repository_dependencies.append(repository_dependency)
return repository_dependencies_folder, folder_id, repository_dependency_id
[docs] def is_subfolder_of(self, folder, repository_dependency):
toolshed, repository_name, repository_owner, changeset_revision, prior_installation_required, only_if_compiling_contained_td = \
common_util.parse_repository_dependency_tuple(repository_dependency)
key = container_util.generate_repository_dependencies_key_for_repository(toolshed,
repository_name,
repository_owner,
changeset_revision,
prior_installation_required,
only_if_compiling_contained_td)
for sub_folder in folder.folders:
if key == sub_folder.key:
return True
return False
[docs] def key_is_current_repositorys_key(self, repository_name, repository_owner, changeset_revision,
prior_installation_required, only_if_compiling_contained_td, key):
repository_components_tuple = container_util.get_components_from_key(key)
components_list = repository_util.extract_components_from_tuple(repository_components_tuple)
toolshed, key_name, key_owner, key_changeset_revision = components_list[0:4]
# For backward compatibility to the 12/20/12 Galaxy release.
if len(components_list) == 4:
key_prior_installation_required = 'False'
key_only_if_compiling_contained_td = 'False'
elif len(components_list) == 5:
key_prior_installation_required = components_list[4]
key_only_if_compiling_contained_td = 'False'
elif len(components_list) == 6:
key_prior_installation_required = components_list[4]
key_only_if_compiling_contained_td = components_list[5]
if (repository_name == key_name and
repository_owner == key_owner and
changeset_revision == key_changeset_revision and
prior_installation_required == key_prior_installation_required and
only_if_compiling_contained_td == key_only_if_compiling_contained_td):
return True
return False
[docs] def populate_repository_dependencies_container(self, repository_dependencies_folder, repository_dependencies, folder_id,
repository_dependency_id):
folder_keys = []
for key in repository_dependencies.keys():
if key not in folder_keys:
folder_keys.append(key)
for key, value in repository_dependencies.items():
repository_dependencies_folder, folder_id, repository_dependency_id = \
self.handle_repository_dependencies_container_entry(repository_dependencies_folder,
key,
value,
folder_id,
repository_dependency_id,
folder_keys)
return repository_dependencies_folder, folder_id, repository_dependency_id
[docs] def prune_folder(self, folder, repository_dependency):
listified_repository_dependency = repository_dependency.listify
if self.is_subfolder_of(folder, listified_repository_dependency):
folder.repository_dependencies.remove(repository_dependency)
[docs] def prune_repository_dependencies(self, folder):
"""
Since the object used to generate a repository dependencies container is a dictionary
and not an odict() (it must be json-serialize-able), the order in which the dictionary
is processed to create the container sometimes results in repository dependency entries
in a folder that also includes the repository dependency as a sub-folder (if the repository
dependency has its own repository dependency). This method will remove all repository
dependencies from folder that are also sub-folders of folder.
"""
repository_dependencies = [rd for rd in folder.repository_dependencies]
for repository_dependency in repository_dependencies:
self.prune_folder(folder, repository_dependency)
for sub_folder in folder.folders:
return self.prune_repository_dependencies(sub_folder)
return folder