Source code for galaxy.tool_shed.galaxy_install.metadata.installed_repository_metadata_manager

import logging
import os
from typing import (

from sqlalchemy import false

from galaxy import util
from galaxy.model.base import transaction
from galaxy.model.tool_shed_install import ToolShedRepository
from galaxy.tool_shed.galaxy_install.client import InstallationTarget
from import tool_panel_manager
from galaxy.tool_shed.metadata.metadata_generator import GalaxyMetadataGenerator
from galaxy.tool_shed.util.repository_util import (
from galaxy.tool_shed.util.tool_util import generate_message_for_invalid_tools
from galaxy.util import inflector
from galaxy.util.tool_shed import (
from galaxy.web.form_builder import SelectField

log = logging.getLogger(__name__)

[docs]class InstalledRepositoryMetadataManager(GalaxyMetadataGenerator): app: InstallationTarget
[docs] def __init__( self, app: InstallationTarget, tpm: Optional[tool_panel_manager.ToolPanelManager] = None, repository: Optional[ToolShedRepository] = None, changeset_revision: Optional[str] = None, repository_clone_url: Optional[str] = None, shed_config_dict: Optional[Dict[str, Any]] = None, relative_install_dir: Optional[str] = None, repository_files_dir: Optional[str] = None, resetting_all_metadata_on_repository: bool = False, updating_installed_repository: bool = False, persist: bool = False, metadata_dict: Optional[Dict[str, Any]] = None, ): super().__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( 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 = f"{str(} ({owner})" option_value = f"{}" 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 ( .filter( == false()) .order_by(,, ) ) else: return == 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: try: tool = os.path.abspath(load_relative_path), guid=guid, use_cached=False ) except Exception: log.exception("Error while loading tool at path '%s'", load_relative_path) tool = None 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: assert self.repository 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() session = session.add(self.repository) with transaction(session): session.commit() log.debug(f"Metadata has been reset on repository {}.") else: log.debug(f"Metadata did not need to be reset on repository {}.") else: log.debug( f"Error locating installation directory for repository {self.repository and}." )
[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. """ message = "" status = "done" if repository_ids := util.listify(kwd.get("repository_ids", None)): successful_count = 0 unsuccessful_count = 0 for repository_id in repository_ids: try: repository = get_installed_tool_shed_repository(, repository_id) self.set_repository(repository) self.reset_all_metadata_on_installed_repository() if self.invalid_file_tups: message = generate_message_for_invalid_tools(, 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",, repository.owner, ) successful_count += 1 except Exception: log.exception("Error attempting to reset metadata on repository %s", str( 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 galaxy 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().set_repository(repository) self.repository_clone_url = common_util.generate_clone_url_for_installed_repository(, repository)
[docs] def tool_shed_from_repository_clone_url(self) -> str: """Given a repository clone URL, return the tool shed that contains the repository.""" repository_clone_url = self.repository_clone_url assert repository_clone_url cleaned_repository_clone_url = common_util.remove_protocol_and_user_from_clone_url(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. """ assert self.repository shed_conf_dict = self.shed_config_dict or self.repository.get_shed_config_dict( 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.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 = 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.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)