Warning

This document is for an in-development version 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:
[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 += f"Version {package_version} of the {package_name} package cannot be " error_message += "installed because the recipe for installing the package is missing " error_message += "either an &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