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.tool.tag_attribute_handler

import copy
import logging

log = logging.getLogger(__name__)


[docs]class TagAttributeHandler(object):
[docs] def __init__(self, app, rdd, unpopulate): self.app = app self.altered = False self.rdd = rdd self.unpopulate = unpopulate
[docs] def process_action_tag_set(self, elem, message): # Here we're inside of an <actions> tag set. See http://localhost:9009/view/devteam/package_r_2_11_0 . # <action> # <repository name="package_readline_6_2" owner="devteam"> # <package name="readline" version="6.2" /> # </repository> # </action> elem_altered = False new_elem = copy.deepcopy(elem) for sub_index, sub_elem in enumerate(elem): altered = False error_message = '' if sub_elem.tag == 'repository': altered, new_sub_elem, error_message = \ self.process_repository_tag_set(parent_elem=elem, elem_index=sub_index, elem=sub_elem, message=message) if error_message and error_message not in message: message += error_message if altered: if not self.altered: self.altered = True if not elem_altered: elem_altered = True new_elem[sub_index] = new_sub_elem return elem_altered, new_elem, message
[docs] def process_actions_tag_set(self, elem, message, skip_actions_tags=True): # <actions> # <package name="libgtextutils" version="0.6"> # <repository name="package_libgtextutils_0_6" owner="test" prior_installation_required="True" /> # </package> elem_altered = False new_elem = copy.deepcopy(elem) for sub_index, sub_elem in enumerate(elem): altered = False error_message = '' if sub_elem.tag == 'package': altered, new_sub_elem, error_message = self.process_package_tag_set(elem=sub_elem, message=message, skip_actions_tags=skip_actions_tags) elif sub_elem.tag == 'action': # <action type="set_environment_for_install"> # <repository name="package_readline_6_2" owner="devteam""> # <package name="readline" version="6.2" /> # </repository> # </action> altered, new_sub_elem, error_message = self.process_action_tag_set(elem=sub_elem, message=message) else: # Inspect the sub elements of elem to locate all <repository> tags and # populate them with toolshed and changeset_revision attributes if necessary. altered, new_sub_elem, error_message = self.rdd.handle_sub_elem(parent_elem=elem, elem_index=sub_index, elem=sub_elem) if error_message and error_message not in message: message += error_message if altered: if not self.altered: self.altered = True if not elem_altered: elem_altered = True new_elem[sub_index] = new_sub_elem return elem_altered, new_elem, message
[docs] def process_actions_group_tag_set(self, elem, message, skip_actions_tags=False): # Inspect all entries in the <actions_group> tag set, skipping <actions> # tag sets that define os and architecture attributes. We want to inspect # only the last <actions> tag set contained within the <actions_group> tag # set to see if a complex repository dependency is defined. elem_altered = False new_elem = copy.deepcopy(elem) for sub_index, sub_elem in enumerate(elem): altered = False error_message = '' if sub_elem.tag == 'actions': if skip_actions_tags: # Skip all actions tags that include os or architecture attributes. system = sub_elem.get('os') architecture = sub_elem.get('architecture') if system or architecture: continue altered, new_sub_elem, error_message = \ self.process_actions_tag_set(elem=sub_elem, message=message, skip_actions_tags=skip_actions_tags) if error_message and error_message not in message: message += error_message if altered: if not self.altered: self.altered = True if not elem_altered: elem_altered = True new_elem[sub_index] = new_sub_elem return elem_altered, new_elem, message
[docs] def process_config(self, root, skip_actions_tags=True): error_message = '' new_root = copy.deepcopy(root) if root.tag == 'tool_dependency': for elem_index, elem in enumerate(root): altered = False if elem.tag == 'package': # <package name="eigen" version="2.0.17"> altered, new_elem, error_message = \ self.process_package_tag_set(elem=elem, message=error_message, skip_actions_tags=skip_actions_tags) if altered: if not self.altered: self.altered = True new_root[elem_index] = new_elem else: error_message = "Invalid tool_dependencies.xml file." return self.altered, new_root, error_message
[docs] def process_install_tag_set(self, elem, message, skip_actions_tags=True): # <install version="1.0"> elem_altered = False new_elem = copy.deepcopy(elem) for sub_index, sub_elem in enumerate(elem): altered = False error_message = '' if sub_elem.tag == 'actions_group': altered, new_sub_elem, error_message = \ self.process_actions_group_tag_set(elem=sub_elem, message=message, skip_actions_tags=skip_actions_tags) elif sub_elem.tag == 'actions': altered, new_sub_elem, error_message = \ self.process_actions_tag_set(elem=sub_elem, message=message, skip_actions_tags=skip_actions_tags) else: package_name = elem.get('name', '') package_version = elem.get('version', '') error_message += 'Version %s of the %s package cannot be installed because ' % \ (str(package_version), str(package_name)) error_message += 'the recipe for installing the package is missing either an ' error_message += '&lt;actions&gt; tag set or an &lt;actions_group&gt; tag set.' if error_message and error_message not in message: message += error_message if altered: if not self.altered: self.altered = True if not elem_altered: elem_altered = True new_elem[sub_index] = new_sub_elem return elem_altered, new_elem, message
[docs] def process_package_tag_set(self, elem, message, skip_actions_tags=True): elem_altered = False new_elem = copy.deepcopy(elem) for sub_index, sub_elem in enumerate(elem): altered = False error_message = '' if sub_elem.tag == 'install': altered, new_sub_elem, error_message = \ self.process_install_tag_set(elem=sub_elem, message=message, skip_actions_tags=skip_actions_tags) elif sub_elem.tag == 'repository': altered, new_sub_elem, error_message = \ self.process_repository_tag_set(parent_elem=elem, elem_index=sub_index, elem=sub_elem, message=message) if error_message and error_message not in message: message += error_message if altered: if not self.altered: self.altered = True if not elem_altered: elem_altered = True new_elem[sub_index] = new_sub_elem return elem_altered, new_elem, message
[docs] def process_repository_tag_set(self, parent_elem, elem_index, elem, message): # We have a complex repository dependency. altered, new_elem, error_message = self.rdd.handle_complex_dependency_elem(parent_elem=parent_elem, elem_index=elem_index, elem=elem) if error_message and error_message not in message: message += error_message if altered: if not self.altered: self.altered = True return altered, new_elem, message