Source code for galaxy.tool_shed.galaxy_install.metadata.installed_repository_metadata_manager
import logging
import os
from typing import (
Any,
Dict,
Optional,
)
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 galaxy.tool_shed.galaxy_install.tools import tool_panel_manager
from galaxy.tool_shed.metadata.metadata_generator import GalaxyMetadataGenerator
from galaxy.tool_shed.util.repository_util import (
get_installed_tool_shed_repository,
get_repository_owner,
)
from galaxy.tool_shed.util.tool_util import generate_message_for_invalid_tools
from galaxy.util import inflector
from galaxy.util.tool_shed import (
common_util,
xml_util,
)
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(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 = f"{str(repository.name)} ({owner})"
option_value = f"{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.uninstalled == false())
.order_by(
self.app.install_model.ToolShedRepository.name,
self.app.install_model.ToolShedRepository.owner,
)
)
else:
return self.app.install_model.context.query(self.app.install_model.ToolShedRepository).filter(
self.app.install_model.ToolShedRepository.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:
try:
tool = self.app.toolbox.load_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 # type:ignore[assignment]
self.update_in_shed_tool_config()
session = self.app.install_model.context
session.add(self.repository)
with transaction(session):
session.commit()
log.debug(f"Metadata has been reset on repository {self.repository.name}.")
else:
log.debug(f"Metadata did not need to be reset on repository {self.repository.name}.")
else:
log.debug(
f"Error locating installation directory for repository {self.repository and 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.
"""
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(self.app, 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.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",
repository.name,
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 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(self.app, 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(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 = 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)