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 galaxy.tool_shed.galaxy_install.tools.data_manager

import errno
import logging
import os
import time

from galaxy.util import (
    etree,
    parse_xml_string,
    xml_to_string,
)
from galaxy.util.renamed_temporary_file import RenamedTemporaryFile
from galaxy.util.tool_shed.xml_util import parse_xml
from . import tool_panel_manager

log = logging.getLogger(__name__)

SHED_DATA_MANAGER_CONF_XML = """<?xml version="1.0"?>
<data_managers>
</data_managers>
"""


[docs]class DataManagerHandler:
[docs] def __init__(self, app): self.app = app
@property def data_managers_path(self): tree, error_message = parse_xml(self.app.config.shed_data_manager_config_file) if tree: root = tree.getroot() return root.get('tool_path', None) return None
[docs] def data_manager_config_elems_to_xml_file(self, config_elems, config_filename): """ Persist the current in-memory list of config_elems to a file named by the value of config_filename. """ data_managers_path = self.data_managers_path if data_managers_path: root_str = f'<?xml version="1.0"?><data_managers tool_path="{data_managers_path}"></data_managers>' else: root_str = '<?xml version="1.0"?><data_managers></data_managers>' root = parse_xml_string(root_str) for elem in config_elems: root.append(elem) try: with RenamedTemporaryFile(config_filename, mode='w') as fh: fh.write(xml_to_string(root)) except Exception: log.exception("Exception in DataManagerHandler.data_manager_config_elems_to_xml_file")
[docs] def install_data_managers(self, shed_data_manager_conf_filename, metadata_dict, shed_config_dict, relative_install_dir, repository, repository_tools_tups): rval = [] if 'data_manager' in metadata_dict: tpm = tool_panel_manager.ToolPanelManager(self.app) repository_tools_by_guid = {} for tool_tup in repository_tools_tups: repository_tools_by_guid[tool_tup[1]] = dict(tool_config_filename=tool_tup[0], tool=tool_tup[2]) # Load existing data managers. try: tree, error_message = parse_xml(shed_data_manager_conf_filename, check_exists=False) except OSError as exc: if exc.errno == errno.ENOENT: with open(shed_data_manager_conf_filename, 'w') as fh: fh.write(SHED_DATA_MANAGER_CONF_XML) tree, error_message = parse_xml(shed_data_manager_conf_filename) else: raise if tree is None: return rval config_elems = [elem for elem in tree.getroot()] repo_data_manager_conf_filename = metadata_dict['data_manager'].get('config_filename', None) if repo_data_manager_conf_filename is None: log.debug("No data_manager_conf.xml file has been defined.") return rval data_manager_config_has_changes = False relative_repo_data_manager_dir = os.path.join(shed_config_dict.get('tool_path', ''), relative_install_dir) repo_data_manager_conf_filename = os.path.join(relative_repo_data_manager_dir, repo_data_manager_conf_filename) tree, error_message = parse_xml(repo_data_manager_conf_filename) if tree is None: return rval root = tree.getroot() for elem in root: if elem.tag == 'data_manager': data_manager_id = elem.get('id', None) if data_manager_id is None: log.error("A data manager was defined that does not have an id and will not be installed:\n%s" % xml_to_string(elem)) continue data_manager_dict = metadata_dict['data_manager'].get('data_managers', {}).get(data_manager_id, None) if data_manager_dict is None: log.error(f"Data manager metadata is not defined properly for '{data_manager_id}'.") continue guid = data_manager_dict.get('guid', None) if guid is None: log.error(f"Data manager guid '{guid}' is not set in metadata for '{data_manager_id}'.") continue elem.set('guid', guid) tool_guid = data_manager_dict.get('tool_guid', None) if tool_guid is None: log.error(f"Data manager tool guid '{tool_guid}' is not set in metadata for '{data_manager_id}'.") continue tool_dict = repository_tools_by_guid.get(tool_guid, None) if tool_dict is None: log.error("Data manager tool guid '%s' could not be found for '%s'. Perhaps the tool is invalid?" % (tool_guid, data_manager_id)) continue tool = tool_dict.get('tool', None) if tool is None: log.error("Data manager tool with guid '%s' could not be found for '%s'. Perhaps the tool is invalid?" % (tool_guid, data_manager_id)) continue tool_config_filename = tool_dict.get('tool_config_filename', None) if tool_config_filename is None: log.error(f"Data manager metadata is missing 'tool_config_file' for '{data_manager_id}'.") continue elem.set('shed_conf_file', shed_config_dict['config_filename']) if elem.get('tool_file', None) is not None: del elem.attrib['tool_file'] # remove old tool_file info tool_elem = tpm.generate_tool_elem(repository.tool_shed, repository.name, repository.installed_changeset_revision, repository.owner, tool_config_filename, tool, None) elem.insert(0, tool_elem) data_manager = \ self.app.data_managers.load_manager_from_elem(elem, tool_path=shed_config_dict.get('tool_path', '')) if data_manager: rval.append(data_manager) elif elem.tag is etree.Comment: pass else: log.warning(f"Encountered unexpected element '{elem.tag}':\n{xml_to_string(elem)}") config_elems.append(elem) data_manager_config_has_changes = True # Persist the altered shed_data_manager_config file. if data_manager_config_has_changes: reload_count = self.app.data_managers._reload_count self.data_manager_config_elems_to_xml_file(config_elems, shed_data_manager_conf_filename) while self.app.data_managers._reload_count <= reload_count: time.sleep(0.1) # Wait for shed_data_manager watcher thread to pick up changes return rval
[docs] def remove_from_data_manager(self, repository): metadata_dict = repository.metadata_ if metadata_dict and 'data_manager' in metadata_dict: shed_data_manager_conf_filename = self.app.config.shed_data_manager_config_file tree, error_message = parse_xml(shed_data_manager_conf_filename) if tree: root = tree.getroot() assert root.tag == 'data_managers', f'The file provided ({shed_data_manager_conf_filename}) for removing data managers from is not a valid data manager xml file.' guids = [data_manager_dict.get('guid') for data_manager_dict in metadata_dict.get('data_manager', {}).get('data_managers', {}).values() if 'guid' in data_manager_dict] load_old_data_managers_by_guid = {} data_manager_config_has_changes = False config_elems = [] for elem in root: # Match Data Manager elements by guid and installed_changeset_revision elem_matches_removed_data_manager = False if elem.tag == 'data_manager': guid = elem.get('guid', None) if guid in guids: tool_elem = elem.find('tool') if tool_elem is not None: installed_changeset_revision_elem = tool_elem.find('installed_changeset_revision') if installed_changeset_revision_elem is not None: if installed_changeset_revision_elem.text == repository.installed_changeset_revision: elem_matches_removed_data_manager = True else: # This is a different version, which had been previously overridden load_old_data_managers_by_guid[guid] = elem if elem_matches_removed_data_manager: data_manager_config_has_changes = True else: config_elems.append(elem) # Remove data managers from in memory self.app.data_managers.remove_manager(guids) # Load other versions of any now uninstalled data managers, if any for elem in load_old_data_managers_by_guid.values(): self.app.data_managers.load_manager_from_elem(elem) # Persist the altered shed_data_manager_config file. if data_manager_config_has_changes: self.data_manager_config_elems_to_xml_file(config_elems, shed_data_manager_conf_filename)