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.dependencies.attribute_handlers

import copy
import logging

from galaxy.util import asbool
from galaxy.web import url_for
from tool_shed.dependencies.tool import tag_attribute_handler
from tool_shed.repository_types.util import REPOSITORY_DEPENDENCY_DEFINITION_FILENAME
from tool_shed.repository_types.util import TOOL_DEPENDENCY_DEFINITION_FILENAME
from tool_shed.util import hg_util
from tool_shed.util import metadata_util
from tool_shed.util import repository_util
from tool_shed.util import xml_util

log = logging.getLogger(__name__)


[docs]class RepositoryDependencyAttributeHandler:
[docs] def __init__(self, app, unpopulate): self.app = app self.file_name = REPOSITORY_DEPENDENCY_DEFINITION_FILENAME self.unpopulate = unpopulate
[docs] def check_tag_attributes(self, elem): # <repository name="molecule_datatypes" owner="test" /> error_message = '' name = elem.get('name') if not name: error_message += 'The tag is missing the required name attribute. ' owner = elem.get('owner') if not owner: error_message += 'The tag is missing the required owner attribute. ' log.debug(error_message) return error_message
[docs] def handle_complex_dependency_elem(self, parent_elem, elem_index, elem): """ Populate or unpopulate the toolshed and changeset_revision attributes of a <repository> tag that defines a complex repository dependency. """ # <repository name="package_eigen_2_0" owner="test" prior_installation_required="True" /> altered, new_elem, error_message = self.handle_elem(elem) if error_message: error_message += f' The {TOOL_DEPENDENCY_DEFINITION_FILENAME} file contains an invalid <repository> tag.' return altered, new_elem, error_message
[docs] def handle_elem(self, elem): """Populate or unpopulate the changeset_revision and toolshed attributes of repository tags.""" # <repository name="molecule_datatypes" owner="test" changeset_revision="1a070566e9c6" /> # <repository changeset_revision="xxx" name="package_xorg_macros_1_17_1" owner="test" toolshed="yyy"> # <package name="xorg_macros" version="1.17.1" /> # </repository> error_message = '' name = elem.get('name') owner = elem.get('owner') # The name and owner attributes are always required, so if either are missing, return the error message. if not name or not owner: error_message = self.check_tag_attributes(elem) return False, elem, error_message altered = False toolshed = elem.get('toolshed') changeset_revision = elem.get('changeset_revision') # Over a short period of time a bug existed which caused the prior_installation_required attribute # to be set to False and included in the <repository> tag when a repository was exported along with # its dependencies. The following will eliminate this problematic attribute upon import. prior_installation_required = elem.get('prior_installation_required') if prior_installation_required is not None and not asbool(prior_installation_required): del elem.attrib['prior_installation_required'] sub_elems = [child_elem for child_elem in list(elem)] if len(sub_elems) > 0: # At this point, a <repository> tag will point only to a package. # <package name="xorg_macros" version="1.17.1" /> # Coerce the list to dict. sub_elements = {} packages = [] for sub_elem in sub_elems: sub_elem_type = sub_elem.tag sub_elem_name = sub_elem.get('name') sub_elem_version = sub_elem.get('version') if sub_elem_type and sub_elem_name and sub_elem_version: packages.append((sub_elem_name, sub_elem_version)) sub_elements['packages'] = packages else: # Set to None. sub_elements = None if self.unpopulate: # We're exporting the repository, so eliminate all toolshed and changeset_revision attributes # from the <repository> tag. if toolshed or changeset_revision: attributes = {} attributes['name'] = name attributes['owner'] = owner prior_installation_required = elem.get('prior_installation_required') if asbool(prior_installation_required): attributes['prior_installation_required'] = 'True' new_elem = xml_util.create_element('repository', attributes=attributes, sub_elements=sub_elements) altered = True return altered, new_elem, error_message # From here on we're populating the toolshed and changeset_revision attributes if necessary. if not toolshed: # Default the setting to the current tool shed. toolshed = str(url_for('/', qualified=True)).rstrip('/') elem.attrib['toolshed'] = toolshed altered = True if not changeset_revision: # Populate the changeset_revision attribute with the latest installable metadata revision for # the defined repository. We use the latest installable revision instead of the latest metadata # revision to ensure that the contents of the revision are valid. repository = repository_util.get_repository_by_name_and_owner(self.app, name, owner) if repository: lastest_installable_changeset_revision = \ metadata_util.get_latest_downloadable_changeset_revision(self.app, repository) if lastest_installable_changeset_revision != hg_util.INITIAL_CHANGELOG_HASH: elem.attrib['changeset_revision'] = lastest_installable_changeset_revision altered = True else: error_message = 'Invalid latest installable changeset_revision %s ' % \ str(lastest_installable_changeset_revision) error_message += f'retrieved for repository {str(name)} owned by {str(owner)}. ' else: error_message = f'Unable to locate repository with name {str(name)} and owner {str(owner)}. ' return altered, elem, error_message
[docs] def handle_sub_elem(self, parent_elem, elem_index, elem): """ Populate or unpopulate the toolshed and changeset_revision attributes for each of the following tag sets. <action type="set_environment_for_install"> <action type="setup_r_environment"> <action type="setup_ruby_environment"> """ sub_elem_altered = False error_message = '' for sub_index, sub_elem in enumerate(elem): # Make sure to skip comments and tags that are not <repository>. if sub_elem.tag == 'repository': altered, new_sub_elem, message = self.handle_elem(sub_elem) if message: error_message += 'The %s file contains an invalid <repository> tag. %s' % \ (TOOL_DEPENDENCY_DEFINITION_FILENAME, message) if altered: if not sub_elem_altered: sub_elem_altered = True elem[sub_index] = new_sub_elem if sub_elem_altered: parent_elem[elem_index] = elem return sub_elem_altered, parent_elem, error_message
[docs] def handle_tag_attributes(self, config): """ Populate or unpopulate the toolshed and changeset_revision attributes of a <repository> tag. Populating will occur when a dependency definition file is being uploaded to the repository, while unpopulating will occur when the repository is being exported. """ # Make sure we're looking at a valid repository_dependencies.xml file. tree, error_message = xml_util.parse_xml(config) if tree is None: return False, None, error_message root = tree.getroot() root_altered = False new_root = copy.deepcopy(root) for index, elem in enumerate(root): if elem.tag == 'repository': # <repository name="molecule_datatypes" owner="test" changeset_revision="1a070566e9c6" /> altered, new_elem, error_message = self.handle_elem(elem) if error_message: error_message = f'The {self.file_name} file contains an invalid <repository> tag. {error_message}' return False, None, error_message if altered: if not root_altered: root_altered = True new_root[index] = new_elem return root_altered, new_root, error_message
[docs]class ToolDependencyAttributeHandler:
[docs] def __init__(self, app, unpopulate): self.app = app self.file_name = TOOL_DEPENDENCY_DEFINITION_FILENAME self.unpopulate = unpopulate
[docs] def handle_tag_attributes(self, tool_dependencies_config): """ Populate or unpopulate the tooshed and changeset_revision attributes of each <repository> tag defined within a tool_dependencies.xml file. """ rdah = RepositoryDependencyAttributeHandler(self.app, self.unpopulate) tah = tag_attribute_handler.TagAttributeHandler(self.app, rdah, self.unpopulate) altered = False error_message = '' # Make sure we're looking at a valid tool_dependencies.xml file. tree, error_message = xml_util.parse_xml(tool_dependencies_config) if tree is None: return False, None, error_message root = tree.getroot() altered, new_root, error_message = tah.process_config(root, skip_actions_tags=False) return altered, new_root, error_message