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.galaxy_install.metadata.installed_repository_metadata_manager
import logging
import os
from sqlalchemy import false
from galaxy import util
from galaxy.util import inflector
from galaxy.web.form_builder import SelectField
from tool_shed.galaxy_install.tools import tool_panel_manager
from tool_shed.metadata import metadata_generator
from tool_shed.util import common_util
from tool_shed.util import repository_util
from tool_shed.util import tool_util
from tool_shed.util import xml_util
log = logging.getLogger(__name__)
[docs]class InstalledRepositoryMetadataManager(metadata_generator.MetadataGenerator):
[docs] def __init__(self, app, tpm=None, repository=None, changeset_revision=None, repository_clone_url=None,
shed_config_dict=None, relative_install_dir=None, repository_files_dir=None,
resetting_all_metadata_on_repository=False, updating_installed_repository=False,
persist=False, metadata_dict=None):
super(InstalledRepositoryMetadataManager, self).__init__(app, repository, changeset_revision,
repository_clone_url, shed_config_dict,
relative_install_dir, repository_files_dir,
resetting_all_metadata_on_repository,
updating_installed_repository, persist,
metadata_dict=metadata_dict, user=None)
if tpm is None:
self.tpm = tool_panel_manager.ToolPanelManager(self.app)
else:
self.tpm = tpm
[docs] def build_repository_ids_select_field(self, name='repository_ids', multiple=True, display='checkboxes'):
"""Generate the current list of repositories for resetting metadata."""
repositories_select_field = SelectField(name=name, multiple=multiple, display=display)
query = self.get_query_for_setting_metadata_on_repositories(order=True)
for repository in query:
owner = str(repository.owner)
option_label = '%s (%s)' % (str(repository.name), owner)
option_value = '%s' % self.app.security.encode_id(repository.id)
repositories_select_field.add_option(option_label, option_value)
return repositories_select_field
[docs] def get_query_for_setting_metadata_on_repositories(self, order=True):
"""
Return a query containing repositories for resetting metadata. The order parameter
is used for displaying the list of repositories ordered alphabetically for display on
a page. When called from the Galaxy API, order is False.
"""
if order:
return self.app.install_model.context.query(self.app.install_model.ToolShedRepository) \
.filter(self.app.install_model.ToolShedRepository.table.c.uninstalled == false()) \
.order_by(self.app.install_model.ToolShedRepository.table.c.name,
self.app.install_model.ToolShedRepository.table.c.owner)
else:
return self.app.install_model.context.query(self.app.install_model.ToolShedRepository) \
.filter(self.app.install_model.ToolShedRepository.table.c.uninstalled == false())
[docs] def get_repository_tools_tups(self):
"""
Return a list of tuples of the form (relative_path, guid, tool) for each tool defined
in the received tool shed repository metadata.
"""
repository_tools_tups = []
shed_conf_dict = self.tpm.get_shed_tool_conf_dict(self.metadata_dict.get('shed_config_filename'))
if 'tools' in self.metadata_dict:
for tool_dict in self.metadata_dict['tools']:
load_relative_path = relative_path = tool_dict.get('tool_config', None)
if shed_conf_dict.get('tool_path'):
load_relative_path = os.path.join(shed_conf_dict.get('tool_path'), relative_path)
guid = tool_dict.get('guid', None)
if relative_path and guid:
tool = self.app.toolbox.load_tool(os.path.abspath(load_relative_path), guid=guid, use_cached=False)
else:
tool = None
if tool:
repository_tools_tups.append((relative_path, guid, tool))
return repository_tools_tups
[docs] def reset_all_metadata_on_installed_repository(self):
"""Reset all metadata on a single tool shed repository installed into a Galaxy instance."""
if self.relative_install_dir:
original_metadata_dict = self.repository.metadata
self.generate_metadata_for_changeset_revision()
if self.metadata_dict != original_metadata_dict:
self.repository.metadata = self.metadata_dict
self.update_in_shed_tool_config()
self.app.install_model.context.add(self.repository)
self.app.install_model.context.flush()
self.app.tool_shed_repository_cache.rebuild()
log.debug('Metadata has been reset on repository %s.' % self.repository.name)
else:
log.debug('Metadata did not need to be reset on repository %s.' % self.repository.name)
else:
log.debug('Error locating installation directory for repository %s.' % self.repository.name)
[docs] def reset_metadata_on_selected_repositories(self, user, **kwd):
"""
Inspect the repository changelog to reset metadata for all appropriate changeset revisions.
This method is called from both Galaxy and the Tool Shed.
"""
repository_ids = util.listify(kwd.get('repository_ids', None))
message = ''
status = 'done'
if repository_ids:
successful_count = 0
unsuccessful_count = 0
for repository_id in repository_ids:
try:
repository = repository_util.get_installed_tool_shed_repository(self.app, repository_id)
self.set_repository(repository)
self.reset_all_metadata_on_installed_repository()
if self.invalid_file_tups:
message = tool_util.generate_message_for_invalid_tools(self.app,
self.invalid_file_tups,
repository,
None,
as_html=False)
log.debug(message)
unsuccessful_count += 1
else:
log.debug("Successfully reset metadata on repository %s owned by %s" %
(str(repository.name), str(repository.owner)))
successful_count += 1
except Exception:
log.exception("Error attempting to reset metadata on repository %s", str(repository.name))
unsuccessful_count += 1
message = "Successfully reset metadata on %d %s. " % \
(successful_count, inflector.cond_plural(successful_count, "repository"))
if unsuccessful_count:
message += "Error setting metadata on %d %s - see the paster log for details. " % \
(unsuccessful_count, inflector.cond_plural(unsuccessful_count, "repository"))
else:
message = 'Select at least one repository to on which to reset all metadata.'
status = 'error'
return message, status
[docs] def set_repository(self, repository):
super(InstalledRepositoryMetadataManager, self).set_repository(repository)
self.repository_clone_url = common_util.generate_clone_url_for_installed_repository(self.app, repository)
[docs] def tool_shed_from_repository_clone_url(self):
"""Given a repository clone URL, return the tool shed that contains the repository."""
cleaned_repository_clone_url = common_util.remove_protocol_and_user_from_clone_url(self.repository_clone_url)
return common_util.remove_protocol_and_user_from_clone_url(cleaned_repository_clone_url).split('/repos/')[0].rstrip('/')
[docs] def update_in_shed_tool_config(self):
"""
A tool shed repository is being updated so change the shed_tool_conf file. Parse the config
file to generate the entire list of config_elems instead of using the in-memory list.
"""
shed_conf_dict = self.repository.get_shed_config_dict(self.app)
shed_tool_conf = shed_conf_dict['config_filename']
tool_path = shed_conf_dict['tool_path']
self.tpm.generate_tool_panel_dict_from_shed_tool_conf_entries(self.repository)
repository_tools_tups = self.get_repository_tools_tups()
clone_url = common_util.generate_clone_url_for_installed_repository(self.app, self.repository)
tool_shed = self.tool_shed_from_repository_clone_url()
owner = self.repository.owner
if not owner:
cleaned_repository_clone_url = common_util.remove_protocol_and_user_from_clone_url(clone_url)
owner = repository_util.get_repository_owner(cleaned_repository_clone_url)
guid_to_tool_elem_dict = {}
for tool_config_filename, guid, tool in repository_tools_tups:
guid_to_tool_elem_dict[guid] = self.tpm.generate_tool_elem(tool_shed,
self.repository.name,
self.repository.changeset_revision,
self.repository.owner or '',
tool_config_filename,
tool,
None)
config_elems = []
tree, error_message = xml_util.parse_xml(shed_tool_conf)
if tree:
root = tree.getroot()
for elem in root:
if elem.tag == 'section':
for i, tool_elem in enumerate(elem):
guid = tool_elem.attrib.get('guid')
if guid in guid_to_tool_elem_dict:
elem[i] = guid_to_tool_elem_dict[guid]
elif elem.tag == 'tool':
guid = elem.attrib.get('guid')
if guid in guid_to_tool_elem_dict:
elem = guid_to_tool_elem_dict[guid]
config_elems.append(elem)
self.tpm.config_elems_to_xml_file(config_elems, shed_tool_conf, tool_path)