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:
@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)