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 galaxy.tool_shed.util.tool_dependency_util

import logging
import os
import shutil

from sqlalchemy import and_

from galaxy import util

log = logging.getLogger(__name__)

[docs]def get_tool_dependency(app, id): """Get a tool_dependency from the database via id""" return app.install_model.context.query(app.install_model.ToolDependency).get(app.security.decode_id(id))
[docs]def get_tool_dependency_by_name_type_repository(app, repository, name, type): context = app.install_model.context return ( context.query(app.install_model.ToolDependency) .filter( and_( app.install_model.ToolDependency.table.c.tool_shed_repository_id == repository.id, app.install_model.ToolDependency.table.c.name == name, app.install_model.ToolDependency.table.c.type == type, ) ) .first() )
[docs]def get_tool_dependency_by_name_version_type(app, name, version, type): context = app.install_model.context return ( context.query(app.install_model.ToolDependency) .filter( and_( app.install_model.ToolDependency.table.c.name == name, app.install_model.ToolDependency.table.c.version == version, app.install_model.ToolDependency.table.c.type == type, ) ) .first() )
[docs]def get_tool_dependency_by_name_version_type_repository(app, repository, name, version, type): context = app.install_model.context return ( context.query(app.install_model.ToolDependency) .filter( and_( app.install_model.ToolDependency.table.c.tool_shed_repository_id == repository.id, app.install_model.ToolDependency.table.c.name == name, app.install_model.ToolDependency.table.c.version == version, app.install_model.ToolDependency.table.c.type == type, ) ) .first() )
[docs]def get_tool_dependency_ids(as_string=False, **kwd): tool_dependency_id = kwd.get("tool_dependency_id", None) if "tool_dependency_ids" in kwd: tool_dependency_ids = util.listify(kwd["tool_dependency_ids"]) elif "id" in kwd: tool_dependency_ids = util.listify(kwd["id"]) elif "inst_td_ids" in kwd: tool_dependency_ids = util.listify(kwd["inst_td_ids"]) elif "uninstalled_tool_dependency_ids" in kwd: tool_dependency_ids = util.listify(kwd["uninstalled_tool_dependency_ids"]) else: tool_dependency_ids = [] if tool_dependency_id and tool_dependency_id not in tool_dependency_ids: tool_dependency_ids.append(tool_dependency_id) if as_string: return ",".join(tool_dependency_ids) return tool_dependency_ids
[docs]def parse_package_elem(package_elem, platform_info_dict=None, include_after_install_actions=True): """ Parse a <package> element within a tool dependency definition and return a list of action tuples. This method is called when setting metadata on a repository that includes a tool_dependencies.xml file or when installing a repository that includes a tool_dependencies.xml file. If installing, platform_info_dict must be a valid dictionary and include_after_install_actions must be True. """ # The actions_elem_tuples list contains <actions> tag sets (possibly inside of an <actions_group> # tag set) to be processed in the order they are defined in the tool_dependencies.xml file. actions_elem_tuples = [] # The tag sets that will go into the actions_elem_list are those that install a compiled binary if # the architecture and operating system match its defined attributes. If compiled binary is not # installed, the first <actions> tag set [following those that have the os and architecture attributes] # that does not have os or architecture attributes will be processed. This tag set must contain the # recipe for downloading and compiling source. actions_elem_list = [] for elem in package_elem: if elem.tag == "actions": # We have an <actions> tag that should not be matched against a specific combination of # architecture and operating system. in_actions_group = False actions_elem_tuples.append((in_actions_group, elem)) elif elem.tag == "actions_group": # We have an actions_group element, and its child <actions> elements should therefore be compared # with the current operating system # and processor architecture. in_actions_group = True # Record the number of <actions> elements so we can filter out any <action> elements that precede # <actions> elements. actions_elem_count = len(elem.findall("actions")) # Record the number of <actions> elements that have both architecture and os specified, in order # to filter out any platform-independent <actions> elements that come before platform-specific # <actions> elements. platform_actions_elements = [] for actions_elem in elem.findall("actions"): if actions_elem.get("architecture") is not None and actions_elem.get("os") is not None: platform_actions_elements.append(actions_elem) platform_actions_element_count = len(platform_actions_elements) platform_actions_elements_processed = 0 actions_elems_processed = 0 # The tag sets that will go into the after_install_actions list are <action> tags instead of <actions> # tags. These will be processed only if they are at the very end of the <actions_group> tag set (after # all <actions> tag sets). See below for details. after_install_actions = [] # Inspect the <actions_group> element and build the actions_elem_list and the after_install_actions list. for child_element in elem: if child_element.tag == "actions": actions_elems_processed += 1 system = child_element.get("os") architecture = child_element.get("architecture") # Skip <actions> tags that have only one of architecture or os specified, in order for the # count in platform_actions_elements_processed to remain accurate. if (system and not architecture) or (architecture and not system): log.debug("Error: Both architecture and os attributes must be specified in an <actions> tag.") continue # Since we are inside an <actions_group> tag set, compare it with our current platform information # and filter the <actions> tag sets that don't match. Require both the os and architecture attributes # to be defined in order to find a match. if system and architecture: platform_actions_elements_processed += 1 # If either the os or architecture do not match the platform, this <actions> tag will not be # considered a match. Skip it and proceed with checking the next one. if platform_info_dict: if platform_info_dict["os"] != system or platform_info_dict["architecture"] != architecture: continue else: # We must not be installing a repository into Galaxy, so determining if we can install a # binary is not necessary. continue else: # <actions> tags without both os and architecture attributes are only allowed to be specified # after platform-specific <actions> tags. If we find a platform-independent <actions> tag before # all platform-specific <actions> tags have been processed. if platform_actions_elements_processed < platform_actions_element_count: debug_msg = "Error: <actions> tags without os and architecture attributes are only allowed " debug_msg += ( "after all <actions> tags with os and architecture attributes have been defined. " ) debug_msg += ( "Skipping the <actions> tag set with no os or architecture attributes that has " ) debug_msg += ( "been defined between two <actions> tag sets that have these attributes defined. " ) log.debug(debug_msg) continue # If we reach this point, it means one of two things: 1) The system and architecture attributes are # not defined in this <actions> tag, or 2) The system and architecture attributes are defined, and # they are an exact match for the current platform. Append the child element to the list of elements # to process. actions_elem_list.append(child_element) elif child_element.tag == "action": # Any <action> tags within an <actions_group> tag set must come after all <actions> tags. if actions_elems_processed == actions_elem_count: # If all <actions> elements have been processed, then this <action> element can be appended to the # list of actions to execute within this group. after_install_actions.append(child_element) else: # If any <actions> elements remain to be processed, then log a message stating that <action> # elements are not allowed to precede any <actions> elements within an <actions_group> tag set. debug_msg = ( "Error: <action> tags are only allowed at the end of an <actions_group> tag set after " ) debug_msg += f"all <actions> tags. Skipping <{child_element.tag}> element with type {child_element.get('type', 'unknown')}." log.debug(debug_msg) continue if platform_info_dict is None and not include_after_install_actions: # We must be setting metadata on a repository. if len(actions_elem_list) >= 1: actions_elem_tuples.append((in_actions_group, actions_elem_list[0])) else: # We are processing a recipe that contains only an <actions_group> tag set for installing a binary, # but does not include an additional recipe for installing and compiling from source. actions_elem_tuples.append((in_actions_group, [])) elif platform_info_dict is not None and include_after_install_actions: # We must be installing a repository. if after_install_actions: actions_elem_list.extend(after_install_actions) actions_elem_tuples.append((in_actions_group, actions_elem_list)) else: # Skip any element that is not <actions> or <actions_group> - this will skip comments, <repository> tags # and <readme> tags. in_actions_group = False continue return actions_elem_tuples
[docs]def remove_tool_dependency(app, tool_dependency): """The received tool_dependency must be in an error state.""" context = app.install_model.context dependency_install_dir = tool_dependency.installation_directory(app) removed, error_message = remove_tool_dependency_installation_directory(dependency_install_dir) if removed: tool_dependency.status = app.install_model.ToolDependency.installation_status.UNINSTALLED tool_dependency.error_message = None context.add(tool_dependency) context.flush() # Since the received tool_dependency is in an error state, nothing will need to be changed in any # of the in-memory dictionaries in the installed_repository_manager because changing the state from # error to uninstalled requires no in-memory changes.. return removed, error_message
[docs]def remove_tool_dependency_installation_directory(dependency_install_dir): if os.path.exists(dependency_install_dir): try: shutil.rmtree(dependency_install_dir) removed = True error_message = "" log.debug(f"Removed tool dependency installation directory: {dependency_install_dir}") except Exception as e: removed = False error_message = f"Error removing tool dependency installation directory {dependency_install_dir}: {e}" log.warning(error_message) else: removed = True error_message = "" return removed, error_message