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.metadata.repository_metadata_manager

import logging
import tempfile
from typing import (
    Any,
    Dict,
    List,
    Optional,
)

from sqlalchemy import (
    false,
    select,
)

from galaxy import util
from galaxy.model.base import transaction
from galaxy.tool_shed.metadata.metadata_generator import (
    BaseMetadataGenerator,
    HandleResultT,
    InvalidFileT,
)
from galaxy.util import inflector
from galaxy.web.form_builder import SelectField
from tool_shed.context import ProvidesRepositoriesContext
from tool_shed.repository_types import util as rt_util
from tool_shed.repository_types.metadata import TipOnly
from tool_shed.structured_app import ToolShedApp
from tool_shed.util import (
    basic_util,
    common_util,
    hg_util,
    metadata_util,
    repository_util,
    shed_util_common as suc,
    tool_util,
)
from tool_shed.util.metadata_util import repository_metadata_by_changeset_revision
from tool_shed.webapp.model import (
    Repository,
    RepositoryMetadata,
    User,
)

log = logging.getLogger(__name__)


[docs]class ToolShedMetadataGenerator(BaseMetadataGenerator): """A MetadataGenerator building on ToolShed's app and repository constructs.""" app: ToolShedApp repository: Optional[Repository] # why is mypy making me re-annotate these things from the base class, it didn't # when they were in the same file invalid_file_tups: List[InvalidFileT] repository_clone_url: Optional[str]
[docs] def __init__( self, trans: ProvidesRepositoriesContext, repository: Optional[Repository] = None, changeset_revision: Optional[str] = None, repository_clone_url: Optional[str] = None, shed_config_dict: Optional[Dict[str, Any]] = None, relative_install_dir=None, repository_files_dir=None, resetting_all_metadata_on_repository=False, updating_installed_repository=False, persist=False, metadata_dict=None, user=None, ): self.trans = trans self.app = trans.app self.user = user self.repository = repository if changeset_revision is None and self.repository is not None: self.changeset_revision = self.repository.tip() else: self.changeset_revision = changeset_revision if repository_clone_url is None and self.repository is not None: self.repository_clone_url = common_util.generate_clone_url_for(self.trans, self.repository) else: self.repository_clone_url = repository_clone_url if shed_config_dict is None: self.shed_config_dict = {} else: self.shed_config_dict = shed_config_dict if relative_install_dir is None and self.repository is not None: relative_install_dir = self.repository.repo_path(self.app) if repository_files_dir is None and self.repository is not None: repository_files_dir = self.repository.repo_path(self.app) if metadata_dict is None: self.metadata_dict = {} else: self.metadata_dict = metadata_dict self.relative_install_dir = relative_install_dir self.repository_files_dir = repository_files_dir self.resetting_all_metadata_on_repository = resetting_all_metadata_on_repository self.updating_installed_repository = updating_installed_repository self.persist = persist self.invalid_file_tups = [] self.sa_session = trans.app.model.session
[docs] def initial_metadata_dict(self) -> Dict[str, Any]: return {}
[docs] def set_repository( self, repository, relative_install_dir: Optional[str] = None, changeset_revision: Optional[str] = None ): self.repository = repository if relative_install_dir is None and self.repository is not None: relative_install_dir = repository.repo_path(self.app) if changeset_revision is None and self.repository is not None: self.set_changeset_revision(self.repository.tip()) else: self.set_changeset_revision(changeset_revision) self.shed_config_dict = {} self._reset_attributes_after_repository_update(relative_install_dir)
[docs] def handle_repository_elem(self, repository_elem, only_if_compiling_contained_td=False) -> HandleResultT: """ Process the received repository_elem which is a <repository> tag either from a repository_dependencies.xml file or a tool_dependencies.xml file. If the former, we're generating repository dependencies metadata for a repository in the Tool Shed. If the latter, we're generating package dependency metadata within Galaxy or the Tool Shed. """ is_valid = True error_message = "" toolshed = repository_elem.get("toolshed", None) name = repository_elem.get("name", None) owner = repository_elem.get("owner", None) changeset_revision = repository_elem.get("changeset_revision", None) prior_installation_required = str(repository_elem.get("prior_installation_required", False)) repository_dependency_tup = [ toolshed, name, owner, changeset_revision, prior_installation_required, str(only_if_compiling_contained_td), ] if not toolshed: # Default to the current tool shed. toolshed = self.trans.repositories_hostname log.warning(f"\n\n\n\n\n\nin not toolshed with {toolshed}\n\n\n\n") # toolshed = str(url_for("/", qualified=True)).rstrip("/") repository_dependency_tup[0] = toolshed else: log.warning(f"moooocww.....{toolshed}\n\n\n\n\n") toolshed = common_util.remove_protocol_from_tool_shed_url(toolshed) if suc.tool_shed_is_this_tool_shed(toolshed, trans=self.trans): try: user = get_user_by_username(self.sa_session, owner) except Exception: error_message = ( f"Ignoring repository dependency definition for tool shed {toolshed}, name {name}, owner {owner}, " ) error_message += f"changeset revision {changeset_revision} because the owner is invalid." log.debug(error_message) is_valid = False return repository_dependency_tup, is_valid, error_message try: repository = get_repository(self.sa_session, name, user.id) except Exception: error_message = f"Ignoring repository dependency definition for tool shed {toolshed}," error_message += f"name {name}, owner {owner}, " error_message += f"changeset revision {changeset_revision} because the name is invalid. " log.debug(error_message) is_valid = False return repository_dependency_tup, is_valid, error_message repo = repository.hg_repo # The received changeset_revision may be None since defining it in the dependency definition is optional. # If this is the case, the default will be to set its value to the repository dependency tip revision. # This probably occurs only when handling circular dependency definitions. tip_ctx = repo[repo.changelog.tip()] # Make sure the repo.changlog includes at least 1 revision. if changeset_revision is None and tip_ctx.rev() >= 0: changeset_revision = str(tip_ctx) repository_dependency_tup = [ toolshed, name, owner, changeset_revision, prior_installation_required, str(only_if_compiling_contained_td), ] return repository_dependency_tup, is_valid, error_message else: # Find the specified changeset revision in the repository's changelog to see if it's valid. found = False for changeset in repo.changelog: changeset_hash = str(repo[changeset]) if changeset_hash == changeset_revision: found = True break if not found: error_message = f"Ignoring repository dependency definition for tool shed {toolshed}, name {name}, owner {owner}, " error_message += ( f"changeset revision {changeset_revision} because the changeset revision is invalid. " ) log.debug(error_message) is_valid = False return repository_dependency_tup, is_valid, error_message else: # Repository dependencies are currently supported within a single tool shed. error_message = "Repository dependencies are currently supported only within the same tool shed. Ignoring " error_message += f"repository dependency definition for tool shed {toolshed}, name {name}, owner {owner}, changeset revision {changeset_revision}. " log.debug(error_message) is_valid = False return repository_dependency_tup, is_valid, error_message return repository_dependency_tup, is_valid, error_message
[docs]class RepositoryMetadataManager(ToolShedMetadataGenerator):
[docs] def __init__( self, trans: ProvidesRepositoriesContext, 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().__init__( trans, 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=trans.user, ) app = trans.app user = trans.user self.sa_session = app.model.context self.app = app self.user = user # Repository metadata comparisons for changeset revisions. self.EQUAL = "equal" self.NO_METADATA = "no metadata" self.NOT_EQUAL_AND_NOT_SUBSET = "not equal and not subset" self.SUBSET = "subset" self.SUBSET_VALUES = [self.EQUAL, self.SUBSET]
def _add_tool_versions(self, id: int, repository_metadata, changeset_revisions): # Build a dictionary of { 'tool id' : 'parent tool id' } pairs for each tool in repository_metadata. metadata = repository_metadata.metadata tool_versions_dict = {} for tool_dict in metadata.get("tools", []): # We have at least 2 changeset revisions to compare tool guids and tool ids. parent_id = self._get_parent_id( id, tool_dict["id"], tool_dict["version"], tool_dict["guid"], changeset_revisions ) tool_versions_dict[tool_dict["guid"]] = parent_id if tool_versions_dict: repository_metadata.tool_versions = tool_versions_dict self.sa_session.add(repository_metadata) session = self.sa_session() with transaction(session): session.commit()
[docs] def build_repository_ids_select_field( self, name="repository_ids", multiple=True, display="checkboxes", my_writable=False ): """Generate the current list of repositories for resetting metadata.""" repositories_select_field = SelectField(name=name, multiple=multiple, display=display) for repository in self.get_repositories_for_setting_metadata(my_writable=my_writable, order=True): owner = str(repository.user.username) 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
def _clean_repository_metadata(self, changeset_revisions): assert self.repository # Delete all repository_metadata records associated with the repository that have # a changeset_revision that is not in changeset_revisions. We sometimes see multiple # records with the same changeset revision value - no idea how this happens. We'll # assume we can delete the older records, so we'll order by update_time descending and # delete records that have the same changeset_revision we come across later. for repository_metadata in get_repository_metadata(self.sa_session, self.repository.id): changeset_revision = repository_metadata.changeset_revision if changeset_revision not in changeset_revisions: self.sa_session.delete(repository_metadata) session = self.sa_session() with transaction(session): session.commit()
[docs] def compare_changeset_revisions(self, ancestor_changeset_revision, ancestor_metadata_dict): """ Compare the contents of two changeset revisions to determine if a new repository metadata revision should be created. """ # The metadata associated with ancestor_changeset_revision is ancestor_metadata_dict. # This changeset_revision is an ancestor of self.changeset_revision which is associated # with self.metadata_dict. A new repository_metadata record will be created only # when this method returns the constant value self.NOT_EQUAL_AND_NOT_SUBSET. ancestor_tools = ancestor_metadata_dict.get("tools", []) ancestor_guids = [tool_dict["guid"] for tool_dict in ancestor_tools] ancestor_guids.sort() ancestor_repository_dependencies_dict = ancestor_metadata_dict.get("repository_dependencies", {}) ancestor_repository_dependencies = ancestor_repository_dependencies_dict.get("repository_dependencies", []) ancestor_tool_dependencies = ancestor_metadata_dict.get("tool_dependencies", {}) ancestor_data_manager = ancestor_metadata_dict.get("data_manager", {}) current_tools = self.metadata_dict.get("tools", []) current_guids = [tool_dict["guid"] for tool_dict in current_tools] current_guids.sort() current_repository_dependencies_dict = self.metadata_dict.get("repository_dependencies", {}) current_repository_dependencies = current_repository_dependencies_dict.get("repository_dependencies", []) current_tool_dependencies = self.metadata_dict.get("tool_dependencies", {}) current_data_manager = self.metadata_dict.get("data_manager", {}) # Handle case where no metadata exists for either changeset. no_repository_dependencies = not ancestor_repository_dependencies and not current_repository_dependencies no_tool_dependencies = not ancestor_tool_dependencies and not current_tool_dependencies no_tools = not ancestor_guids and not current_guids no_data_manager = not ancestor_data_manager and not current_data_manager if no_repository_dependencies and no_tool_dependencies and no_tools and no_data_manager: return self.NO_METADATA repository_dependency_comparison = self.compare_repository_dependencies( ancestor_repository_dependencies, current_repository_dependencies ) tool_dependency_comparison = self.compare_tool_dependencies( ancestor_tool_dependencies, current_tool_dependencies ) data_manager_comparison = self.compare_data_manager(ancestor_data_manager, current_data_manager) # Handle case where all metadata is the same. if ( ancestor_guids == current_guids and repository_dependency_comparison == self.EQUAL and tool_dependency_comparison == self.EQUAL and data_manager_comparison == self.EQUAL ): return self.EQUAL # Handle case where ancestor metadata is a subset of current metadata. # readme_file_is_subset = readme_file_comparision in [ self.EQUAL, self.SUBSET ] repository_dependency_is_subset = repository_dependency_comparison in self.SUBSET_VALUES tool_dependency_is_subset = tool_dependency_comparison in self.SUBSET_VALUES datamanager_is_subset = data_manager_comparison in self.SUBSET_VALUES if repository_dependency_is_subset and tool_dependency_is_subset and datamanager_is_subset: is_subset = True for guid in ancestor_guids: if guid not in current_guids: is_subset = False break if is_subset: return self.SUBSET return self.NOT_EQUAL_AND_NOT_SUBSET
[docs] def compare_data_manager(self, ancestor_metadata, current_metadata): """Determine if ancestor_metadata is the same as or a subset of current_metadata for data_managers.""" def __data_manager_dict_to_tuple_list(metadata_dict): # we do not check tool_guid or tool conf file name return set( sorted( ( name, tuple(sorted(value.get("data_tables", []))), value.get("guid"), value.get("version"), value.get("name"), value.get("id"), ) for name, value in metadata_dict.items() ) ) # only compare valid entries, any invalid entries are ignored ancestor_metadata = __data_manager_dict_to_tuple_list(ancestor_metadata.get("data_managers", {})) current_metadata = __data_manager_dict_to_tuple_list(current_metadata.get("data_managers", {})) # use set comparisons if ancestor_metadata.issubset(current_metadata): if ancestor_metadata == current_metadata: return self.EQUAL return self.SUBSET return self.NOT_EQUAL_AND_NOT_SUBSET
[docs] def compare_repository_dependencies(self, ancestor_repository_dependencies, current_repository_dependencies): """ Determine if ancestor_repository_dependencies is the same as or a subset of current_repository_dependencies. """ # The list of repository_dependencies looks something like: # [["http://localhost:9009", "emboss_datatypes", "test", "ab03a2a5f407", "False", "False"]]. # Create a string from each tuple in the list for easier comparison. if len(ancestor_repository_dependencies) <= len(current_repository_dependencies): for ancestor_tup in ancestor_repository_dependencies: ( a_tool_shed, a_repo_name, a_repo_owner, a_changeset_revision, a_prior_installation_required, a_only_if_compiling_contained_td, ) = ancestor_tup cleaned_a_tool_shed = common_util.remove_protocol_from_tool_shed_url(a_tool_shed) found_in_current = False for current_tup in current_repository_dependencies: ( c_tool_shed, c_repo_name, c_repo_owner, c_changeset_revision, c_prior_installation_required, c_only_if_compiling_contained_td, ) = current_tup cleaned_c_tool_shed = common_util.remove_protocol_from_tool_shed_url(c_tool_shed) if ( cleaned_c_tool_shed == cleaned_a_tool_shed and c_repo_name == a_repo_name and c_repo_owner == a_repo_owner and c_changeset_revision == a_changeset_revision and util.string_as_bool(c_prior_installation_required) == util.string_as_bool(a_prior_installation_required) and util.string_as_bool(c_only_if_compiling_contained_td) == util.string_as_bool(a_only_if_compiling_contained_td) ): found_in_current = True break if not found_in_current: # In some cases, the only difference between a dependency definition in the lists # is the changeset_revision value. We'll check to see if this is the case, and if # the defined dependency is a repository that has metadata set only on its tip. if not self.different_revision_defines_tip_only_repository_dependency( ancestor_tup, current_repository_dependencies ): return self.NOT_EQUAL_AND_NOT_SUBSET return self.SUBSET if len(ancestor_repository_dependencies) == len(current_repository_dependencies): return self.EQUAL else: return self.SUBSET return self.NOT_EQUAL_AND_NOT_SUBSET
[docs] def compare_tool_dependencies(self, ancestor_tool_dependencies, current_tool_dependencies): """ Determine if ancestor_tool_dependencies is the same as or a subset of current_tool_dependencies. """ # The tool_dependencies dictionary looks something like: # {'bwa/0.5.9': {'readme': 'some string', 'version': '0.5.9', 'type': 'package', 'name': 'bwa'}} if len(ancestor_tool_dependencies) <= len(current_tool_dependencies): for ancestor_td_key in ancestor_tool_dependencies.keys(): if ancestor_td_key in current_tool_dependencies: # The only values that could have changed between the 2 dictionaries are the # "readme" or "type" values. Changing the readme value makes no difference. # Changing the type will change the installation process, but for now we'll # assume it was a typo, so new metadata shouldn't be generated. continue else: return self.NOT_EQUAL_AND_NOT_SUBSET # At this point we know that ancestor_tool_dependencies is at least a subset of current_tool_dependencies. if len(ancestor_tool_dependencies) == len(current_tool_dependencies): return self.EQUAL else: return self.SUBSET return self.NOT_EQUAL_AND_NOT_SUBSET
[docs] def create_or_update_repository_metadata(self, changeset_revision, metadata_dict): """Create or update a repository_metadata record in the tool shed.""" has_repository_dependencies = False has_repository_dependencies_only_if_compiling_contained_td = False includes_tools = False includes_tool_dependencies = False if metadata_dict: repository_dependencies_dict = metadata_dict.get("repository_dependencies", {}) repository_dependencies = repository_dependencies_dict.get("repository_dependencies", []) ( has_repository_dependencies, has_repository_dependencies_only_if_compiling_contained_td, ) = repository_util.get_repository_dependency_types(repository_dependencies) if "tools" in metadata_dict: includes_tools = True if "tool_dependencies" in metadata_dict: includes_tool_dependencies = True if ( has_repository_dependencies or has_repository_dependencies_only_if_compiling_contained_td or includes_tools or includes_tool_dependencies ): downloadable = True else: downloadable = False assert self.repository repository_metadata = repository_metadata_by_changeset_revision( self.app.model, self.repository.id, changeset_revision ) if repository_metadata: repository_metadata.metadata = metadata_dict repository_metadata.downloadable = downloadable repository_metadata.has_repository_dependencies = has_repository_dependencies repository_metadata.includes_datatypes = False repository_metadata.includes_tools = includes_tools repository_metadata.includes_tool_dependencies = includes_tool_dependencies repository_metadata.includes_workflows = False else: repository_metadata = self.app.model.RepositoryMetadata( repository_id=self.repository.id, changeset_revision=changeset_revision, metadata=metadata_dict, downloadable=downloadable, has_repository_dependencies=has_repository_dependencies, includes_datatypes=False, includes_tools=includes_tools, includes_tool_dependencies=includes_tool_dependencies, includes_workflows=False, ) assert repository_metadata # Always set the default values for the following columns. When resetting all metadata # on a repository this will reset the values. assert repository_metadata repository_metadata.missing_test_components = False self.sa_session.add(repository_metadata) session = self.sa_session() with transaction(session): session.commit() return repository_metadata
[docs] def different_revision_defines_tip_only_repository_dependency(self, rd_tup, repository_dependencies): """ Determine if the only difference between rd_tup and a dependency definition in the list of repository_dependencies is the changeset_revision value. """ ( rd_tool_shed, rd_name, rd_owner, rd_changeset_revision, rd_prior_installation_required, rd_only_if_compiling_contained_td, ) = common_util.parse_repository_dependency_tuple(rd_tup) cleaned_rd_tool_shed = common_util.remove_protocol_from_tool_shed_url(rd_tool_shed) for repository_dependency in repository_dependencies: ( tool_shed, name, owner, changeset_revision, prior_installation_required, only_if_compiling_contained_td, ) = common_util.parse_repository_dependency_tuple(repository_dependency) cleaned_tool_shed = common_util.remove_protocol_from_tool_shed_url(tool_shed) if cleaned_rd_tool_shed == cleaned_tool_shed and rd_name == name and rd_owner == owner: # Determine if the repository represented by the dependency tuple is an instance of the repository type TipOnly. required_repository = repository_util.get_repository_by_name_and_owner(self.app, name, owner) repository_type_class = self.app.repository_types_registry.get_class_by_label(required_repository.type) return isinstance(repository_type_class, TipOnly) return False
def _get_parent_id(self, id: int, old_id, version, guid, changeset_revisions): parent_id = None # Compare from most recent to oldest. changeset_revisions.reverse() for changeset_revision in changeset_revisions: repository_metadata = repository_metadata_by_changeset_revision(self.app.model, id, changeset_revision) assert repository_metadata metadata = repository_metadata.metadata tools_dicts = metadata.get("tools", []) for tool_dict in tools_dicts: if tool_dict["guid"] == guid: # The tool has not changed between the compared changeset revisions. continue if tool_dict["id"] == old_id and tool_dict["version"] != version: # The tool version is different, so we've found the parent. return tool_dict["guid"] if parent_id is None: # The tool did not change through all of the changeset revisions. return old_id
[docs] def get_repositories_for_setting_metadata(self, my_writable=False, order=True): """ Return a list of 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 Tool Shed API, order is False. """ # When called from the Tool Shed API, the metadata is reset on all repositories of types # repository_suite_definition and tool_dependency_definition in addition to other selected # repositories. if my_writable: username = self.user.username repo_ids = [] for repository in get_current_repositories(self.sa_session): # Always reset metadata on all repositories of types repository_suite_definition and # tool_dependency_definition. if repository.type in [rt_util.REPOSITORY_SUITE_DEFINITION, rt_util.TOOL_DEPENDENCY_DEFINITION]: repo_ids.append(repository.id) else: allow_push = repository.allow_push() if allow_push: # Include all repositories that are writable by the current user. allow_push_usernames = allow_push.split(",") if username in allow_push_usernames: repo_ids.append(repository.id) if repo_ids: return get_filtered_repositories(self.sa_session, repo_ids, order) else: return [] else: return get_current_repositories(self.sa_session, order)
[docs] def new_metadata_required_for_utilities(self): """ This method compares the last stored repository_metadata record associated with self.repository against the contents of self.metadata_dict and returns True or False for the union set of Galaxy utilities contained in both metadata dictionaries. The metadata contained in self.metadata_dict may not be a subset of that contained in the last stored repository_metadata record associated with self.repository because one or more Galaxy utilities may have been deleted from self.repository in the new tip. """ assert self.repository repository_metadata = metadata_util.get_latest_repository_metadata( self.app, self.repository.id, downloadable=False ) repository_dependencies_required = self.new_repository_dependency_metadata_required(repository_metadata) tools_required = self.new_tool_metadata_required(repository_metadata) tool_dependencies_required = self.new_tool_dependency_metadata_required(repository_metadata) data_managers_required = self.new_data_manager_required(repository_metadata) if repository_dependencies_required or tools_required or tool_dependencies_required or data_managers_required: return True return False
[docs] def new_repository_dependency_metadata_required(self, repository_metadata): """ Compare the last saved metadata for each repository dependency in the repository with the new metadata in self.metadata_dict to determine if a new repository_metadata table record is required or if the last saved metadata record can be updated for repository_dependencies instead. """ if repository_metadata: metadata = repository_metadata.metadata if "repository_dependencies" in metadata: saved_repository_dependencies = metadata["repository_dependencies"]["repository_dependencies"] new_repository_dependencies_metadata = self.metadata_dict.get("repository_dependencies", None) if new_repository_dependencies_metadata: new_repository_dependencies = self.metadata_dict["repository_dependencies"][ "repository_dependencies" ] # TODO: We used to include the following here to handle the case where repository # dependency definitions were deleted. However this erroneously returned True in # cases where is should not have done so. This usually occurred where multiple single # files were uploaded when a single tarball should have been. We need to implement # support for handling deleted repository dependency definitions so that we can guarantee # reproducibility, but we need to do it in a way that is better than the following. # for new_repository_dependency in new_repository_dependencies: # if new_repository_dependency not in saved_repository_dependencies: # return True # The saved metadata must be a subset of the new metadata. for saved_repository_dependency in saved_repository_dependencies: if saved_repository_dependency not in new_repository_dependencies: # In some cases, the only difference between a dependency definition in the lists # is the changeset_revision value. We'll check to see if this is the case, and if # the defined dependency is a repository that has metadata set only on its tip. if not self.different_revision_defines_tip_only_repository_dependency( saved_repository_dependency, new_repository_dependencies ): return True return False else: # The repository_dependencies.xml file must have been deleted, so create a new # repository_metadata record so we always have access to the deleted file. return True else: return False else: if "repository_dependencies" in self.metadata_dict: # There is no saved repository metadata, so we need to create a new repository_metadata record. return True else: # self.metadata_dict includes no metadata for repository dependencies, so a new repository_metadata # record is not needed. return False
[docs] def new_data_manager_required(self, repository_metadata): if self.metadata_dict and repository_metadata and repository_metadata.metadata: return self.compare_data_manager(self.metadata_dict, repository_metadata.metadata) != self.EQUAL else: return bool( repository_metadata and repository_metadata.metadata and repository_metadata.metadata.get("data_managers") )
[docs] def new_tool_metadata_required(self, repository_metadata): """ Compare the last saved metadata for each tool in the repository with the new metadata in self.metadata_dict to determine if a new repository_metadata table record is required, or if the last saved metadata record can be updated instead. """ if "tools" in self.metadata_dict: if repository_metadata: metadata = repository_metadata.metadata if metadata: if "tools" in metadata: saved_tool_ids = [] # The metadata for one or more tools was successfully generated in the past # for this repository, so we first compare the version string for each tool id # in self.metadata_dict with what was previously saved to see if we need to create # a new table record or if we can simply update the existing record. for new_tool_metadata_dict in self.metadata_dict["tools"]: for saved_tool_metadata_dict in metadata["tools"]: if saved_tool_metadata_dict["id"] not in saved_tool_ids: saved_tool_ids.append(saved_tool_metadata_dict["id"]) if new_tool_metadata_dict["id"] == saved_tool_metadata_dict["id"]: if new_tool_metadata_dict["version"] != saved_tool_metadata_dict["version"]: return True # So far, a new metadata record is not required, but we still have to check to see if # any new tool ids exist in self.metadata_dict that are not in the saved metadata. We do # this because if a new tarball was uploaded to a repository that included tools, it # may have removed existing tool files if they were not included in the uploaded tarball. for new_tool_metadata_dict in self.metadata_dict["tools"]: if new_tool_metadata_dict["id"] not in saved_tool_ids: return True return False else: # The new metadata includes tools, but the stored metadata does not, so we can # update the stored metadata. return False else: # There is no stored metadata, so we can update the metadata column in the # repository_metadata table. return False else: # There is no stored repository metadata, so we need to create a new repository_metadata # table record. return True # self.metadata_dict includes no metadata for tools, so a new repository_metadata table # record is not needed. return False
[docs] def new_tool_dependency_metadata_required(self, repository_metadata): """ Compare the last saved metadata for each tool dependency in the repository with the new metadata in self.metadata_dict to determine if a new repository_metadata table record is required or if the last saved metadata record can be updated for tool_dependencies instead. """ if repository_metadata: metadata = repository_metadata.metadata if metadata: if "tool_dependencies" in metadata: saved_tool_dependencies = metadata["tool_dependencies"] new_tool_dependencies = self.metadata_dict.get("tool_dependencies", None) if new_tool_dependencies: # TODO: We used to include the following here to handle the case where # tool dependency definitions were deleted. However, this erroneously # returned True in cases where is should not have done so. This usually # occurred where multiple single files were uploaded when a single tarball # should have been. We need to implement support for handling deleted # tool dependency definitions so that we can guarantee reproducibility, # but we need to do it in a way that is better than the following. # for new_tool_dependency in new_tool_dependencies: # if new_tool_dependency not in saved_tool_dependencies: # return True # The saved metadata must be a subset of the new metadata. for saved_tool_dependency in saved_tool_dependencies: if saved_tool_dependency not in new_tool_dependencies: return True return False else: # The tool_dependencies.xml file must have been deleted, so create a new # repository_metadata record so we always have # access to the deleted file. return True else: return False else: # We have repository metadata that does not include metadata for any tool dependencies # in the repository, so we can update the existing repository metadata. return False else: if "tool_dependencies" in self.metadata_dict: # There is no saved repository metadata, so we need to create a new repository_metadata # record. return True else: # self.metadata_dict includes no metadata for tool dependencies, so a new repository_metadata # record is not needed. return False
[docs] def reset_all_metadata_on_repository_in_tool_shed(self, repository_clone_url=None): """Reset all metadata on a single repository in a tool shed.""" assert self.repository log.debug(f"Resetting all metadata on repository: {self.repository.name}") repo = self.repository.hg_repo # The list of changeset_revisions refers to repository_metadata records that have been created # or updated. When the following loop completes, we'll delete all repository_metadata records # for this repository that do not have a changeset_revision value in this list. changeset_revisions: List[Optional[str]] = [] # When a new repository_metadata record is created, it always uses the values of # metadata_changeset_revision and metadata_dict. metadata_changeset_revision = None metadata_dict = None ancestor_changeset_revision = None ancestor_metadata_dict = None for changeset in self.repository.get_changesets_for_setting_metadata(self.app): work_dir = tempfile.mkdtemp(prefix="tmp-toolshed-ramorits") ctx = repo[changeset] log.debug("Cloning repository changeset revision: %s", str(ctx.rev())) assert self.repository_clone_url repository_clone_url = repository_clone_url or self.repository_clone_url cloned_ok, error_message = hg_util.clone_repository(repository_clone_url, work_dir, str(ctx.rev())) if cloned_ok: log.debug("Generating metadata for changeset revision: %s", str(ctx.rev())) self.set_changeset_revision(str(ctx)) self.set_repository_files_dir(work_dir) self.generate_metadata_for_changeset_revision() if self.metadata_dict: if metadata_changeset_revision is None and metadata_dict is None: # We're at the first change set in the change log. metadata_changeset_revision = self.changeset_revision metadata_dict = self.metadata_dict if ancestor_changeset_revision: # Compare metadata from ancestor and current. The value of comparison will be one of: # self.NO_METADATA - no metadata for either ancestor or current, so continue from current # self.EQUAL - ancestor metadata is equivalent to current metadata, so continue from current # self.SUBSET - ancestor metadata is a subset of current metadata, so continue from current # self.NOT_EQUAL_AND_NOT_SUBSET - ancestor metadata is neither equal to nor a subset of current # metadata, so persist ancestor metadata. log.info(f"amd {ancestor_metadata_dict}") comparison = self.compare_changeset_revisions( ancestor_changeset_revision, ancestor_metadata_dict ) log.info(f"comparison {comparison}") if comparison in [self.NO_METADATA, self.EQUAL, self.SUBSET]: ancestor_changeset_revision = self.changeset_revision ancestor_metadata_dict = self.metadata_dict elif comparison == self.NOT_EQUAL_AND_NOT_SUBSET: metadata_changeset_revision = ancestor_changeset_revision metadata_dict = ancestor_metadata_dict self.create_or_update_repository_metadata(metadata_changeset_revision, metadata_dict) changeset_revisions.append(metadata_changeset_revision) ancestor_changeset_revision = self.changeset_revision ancestor_metadata_dict = self.metadata_dict else: # We're at the beginning of the change log. ancestor_changeset_revision = self.changeset_revision ancestor_metadata_dict = self.metadata_dict if not ctx.children(): metadata_changeset_revision = self.changeset_revision metadata_dict = self.metadata_dict # We're at the end of the change log. self.create_or_update_repository_metadata(metadata_changeset_revision, metadata_dict) changeset_revisions.append(metadata_changeset_revision) ancestor_changeset_revision = None ancestor_metadata_dict = None elif ancestor_metadata_dict: # We reach here only if self.metadata_dict is empty and ancestor_metadata_dict is not. if not ctx.children(): # We're at the end of the change log. self.create_or_update_repository_metadata(metadata_changeset_revision, metadata_dict) changeset_revisions.append(metadata_changeset_revision) ancestor_changeset_revision = None ancestor_metadata_dict = None basic_util.remove_dir(work_dir) # Delete all repository_metadata records for this repository that do not have a changeset_revision # value in changeset_revisions. self._clean_repository_metadata(changeset_revisions) # Set tool version information for all downloadable changeset revisions. Get the list of changeset # revisions from the changelog. self._reset_all_tool_versions(repo)
def _reset_all_tool_versions(self, repo): """Reset tool version lineage for those changeset revisions that include valid tools.""" assert self.repository changeset_revisions_that_contain_tools = _get_changeset_revisions_that_contain_tools( self.app, repo, self.repository ) # The list of changeset_revisions_that_contain_tools is now filtered to contain only those that # are downloadable and contain tools. If a repository includes tools, build a dictionary of # { 'tool id' : 'parent tool id' } pairs for each tool in each changeset revision. for index, changeset_revision in enumerate(changeset_revisions_that_contain_tools): tool_versions_dict = {} repository_metadata = repository_metadata_by_changeset_revision( self.app.model, self.repository.id, changeset_revision ) assert repository_metadata metadata = repository_metadata.metadata tool_dicts = metadata["tools"] if index == 0: # The first changeset_revision is a special case because it will have no ancestor # changeset_revisions in which to match tools. The parent tool id for tools in the # first changeset_revision will be the "old_id" in the tool config. for tool_dict in tool_dicts: tool_versions_dict[tool_dict["guid"]] = tool_dict["id"] else: for tool_dict in tool_dicts: parent_id = self._get_parent_id( self.repository.id, tool_dict["id"], tool_dict["version"], tool_dict["guid"], changeset_revisions_that_contain_tools[0:index], ) tool_versions_dict[tool_dict["guid"]] = parent_id if tool_versions_dict: repository_metadata.tool_versions = tool_versions_dict self.sa_session.add(repository_metadata) session = self.sa_session() with transaction(session): session.commit()
[docs] def reset_metadata_on_selected_repositories(self, **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 = repository_util.get_repository_in_tool_shed(self.app, repository_id) self.set_repository(repository) self.resetting_all_metadata_on_repository = True self.reset_all_metadata_on_repository_in_tool_shed() 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", repository.name, repository.user.username, ) 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, repository_clone_url=None): super().set_repository(repository) self.repository_clone_url = repository_clone_url or common_util.generate_clone_url_for(self.trans, repository)
[docs] def set_repository_metadata(self, host, content_alert_str="", **kwd): """ Set metadata using the self.repository's current disk files, returning specific error messages (if any) to alert the repository owner that the changeset has problems. """ assert self.repository message = "" status = "done" repository_id = self.repository.id repo = self.repository.hg_repo self.generate_metadata_for_changeset_revision() if self.metadata_dict: repository_metadata = None repository_type_class = self.app.repository_types_registry.get_class_by_label(self.repository.type) tip_only = isinstance(repository_type_class, TipOnly) if not tip_only and self.new_metadata_required_for_utilities(): # Create a new repository_metadata table row. repository_metadata = self.create_or_update_repository_metadata( self.repository.tip(), self.metadata_dict ) # If this is the first record stored for this repository, see if we need to send any email alerts. if len(self.repository.downloadable_revisions) == 1: suc.handle_email_alerts( self.app, host, self.repository, content_alert_str="", new_repo_alert=True, admin_only=False ) else: # Update the latest stored repository metadata with the contents and attributes of self.metadata_dict. repository_metadata = metadata_util.get_latest_repository_metadata( self.app, repository_id, downloadable=False ) if repository_metadata: downloadable = metadata_util.is_downloadable(self.metadata_dict) # Update the last saved repository_metadata table row. repository_metadata.changeset_revision = self.repository.tip() repository_metadata.metadata = self.metadata_dict repository_metadata.downloadable = downloadable repository_metadata.includes_datatypes = False # We don't store information about the special type of repository dependency that is needed only for # compiling a tool dependency defined for the dependent repository. repository_dependencies_dict = self.metadata_dict.get("repository_dependencies", {}) repository_dependencies = repository_dependencies_dict.get("repository_dependencies", []) ( has_repository_dependencies, has_repository_dependencies_only_if_compiling_contained_td, ) = repository_util.get_repository_dependency_types(repository_dependencies) repository_metadata.has_repository_dependencies = has_repository_dependencies if "tool_dependencies" in self.metadata_dict: repository_metadata.includes_tool_dependencies = True else: repository_metadata.includes_tool_dependencies = False if "tools" in self.metadata_dict: repository_metadata.includes_tools = True else: repository_metadata.includes_tools = False repository_metadata.includes_workflows = False repository_metadata.missing_test_components = False self.sa_session.add(repository_metadata) session = self.sa_session() with transaction(session): session.commit() else: # There are no metadata records associated with the repository. repository_metadata = self.create_or_update_repository_metadata( self.repository.tip(), self.metadata_dict ) if "tools" in self.metadata_dict and repository_metadata and status != "error": # Set tool versions on the new downloadable change set. The order of the list of changesets is # critical, so we use the repo's changelog. changeset_revisions = [] for changeset in repo.changelog: changeset_revision = str(repo[changeset]) if repository_metadata_by_changeset_revision(self.app.model, repository_id, changeset_revision): changeset_revisions.append(changeset_revision) self._add_tool_versions(repository_id, repository_metadata, changeset_revisions) elif len(repo) == 1 and not self.invalid_file_tups: message = "Revision <b>%s</b> includes no Galaxy utilities for which metadata can " % str( self.repository.tip() ) message += "be defined so this revision cannot be automatically installed into a local Galaxy instance." status = "error" if self.invalid_file_tups: message = tool_util.generate_message_for_invalid_tools( self.app, self.invalid_file_tups, self.repository, self.metadata_dict ) status = "error" return message, status
[docs] def set_repository_metadata_due_to_new_tip(self, host, content_alert_str=None, **kwd): """Set metadata on the tip of self.repository in the tool shed.""" error_message, status = self.set_repository_metadata(host, content_alert_str=content_alert_str, **kwd) return status, error_message
def _get_changeset_revisions_that_contain_tools(app: "ToolShedApp", repo, repository) -> List[str]: changeset_revisions_that_contain_tools = [] for changeset in repo.changelog: changeset_revision = str(repo[changeset]) repository_metadata = repository_metadata_by_changeset_revision(app.model, repository.id, changeset_revision) if repository_metadata: metadata = repository_metadata.metadata if metadata: if metadata.get("tools", None): changeset_revisions_that_contain_tools.append(changeset_revision) return changeset_revisions_that_contain_tools
[docs]def get_user_by_username(session, username): stmt = select(User).where(User.username == username) return session.execute(stmt).scalar_one()
[docs]def get_repository(session, name, user_id): stmt = select(Repository).where(Repository.name == name).where(Repository.user_id == user_id) return session.execute(stmt).scalar_one()
[docs]def get_repository_metadata(session, repository_id): stmt = ( select(RepositoryMetadata) .where(RepositoryMetadata.repository_id == repository_id) .order_by(RepositoryMetadata.changeset_revision, RepositoryMetadata.update_time.desc()) # type: ignore[attr-defined] # mapped attribute ) return session.scalars(stmt)
[docs]def get_current_repositories(session, order=False): stmt = select(Repository).where(Repository.deleted == false()) if order: stmt = stmt.order_by(Repository.name, Repository.user_id) return session.scalars(stmt)
[docs]def get_filtered_repositories(session, repo_ids, order): stmt = select(Repository).where(Repository.id.in_(repo_ids)) if order: stmt = stmt.order_by(Repository.name, Repository.user_id) return session.scalars(stmt)