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: """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: """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: """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: """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: """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: """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: """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: """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: """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: """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:
[docs] def __init__(self, app): self.app = app
[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 += f'{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 OrderedDict() (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