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:
[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 <actions> tag set or an <actions_group> 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