Source code for tool_shed.webapp.api.repository_revisions

import logging
from typing import (
    Callable,
    Dict,
)

from sqlalchemy import select

from galaxy import (
    util,
    web,
)
from galaxy.model.base import transaction
from galaxy.webapps.base.controller import HTTPBadRequest
from tool_shed.util import (
    metadata_util,
    repository_util,
)
from tool_shed.webapp.model import RepositoryMetadata
from . import BaseShedAPIController

log = logging.getLogger(__name__)


[docs]class RepositoryRevisionsController(BaseShedAPIController): """RESTful controller for interactions with tool shed repository revisions.""" def __get_value_mapper(self, trans) -> Dict[str, Callable]: value_mapper = { "id": trans.security.encode_id, "repository_id": trans.security.encode_id, "user_id": trans.security.encode_id, } return value_mapper
[docs] @web.legacy_expose_api_anonymous def index(self, trans, **kwd): """ GET /api/repository_revisions Displays a collection (list) of repository revisions. """ # Example URL: http://localhost:9009/api/repository_revisions downloadable = kwd.get("downloadable", None) malicious = kwd.get("malicious", None) missing_test_components = kwd.get("missing_test_components", None) includes_tools = kwd.get("includes_tools", None) repository_metadata_dicts = [] all_repository_metadata = get_repository_metadata( trans.sa_session, downloadable, malicious, missing_test_components, includes_tools ) for repository_metadata in all_repository_metadata: repository_metadata_dict = repository_metadata.to_dict( view="collection", value_mapper=self.__get_value_mapper(trans) ) repository_metadata_dict["url"] = web.url_for( controller="repository_revisions", action="show", id=trans.security.encode_id(repository_metadata.id) ) repository_metadata_dicts.append(repository_metadata_dict) return repository_metadata_dicts
[docs] @web.legacy_expose_api_anonymous def repository_dependencies(self, trans, id, **kwd): """ GET /api/repository_revisions/{encoded repository_metadata id}/repository_dependencies Returns a list of dictionaries that each define a specific downloadable revision of a repository in the Tool Shed. This method returns dictionaries with more information in them than other methods in this controller. The information about repository_metdata is enhanced to include information about the repository (e.g., name, owner, etc) associated with the repository_metadata record. :param id: the encoded id of the `RepositoryMetadata` object """ # Example URL: http://localhost:9009/api/repository_revisions/repository_dependencies/bb125606ff9ea620 repository_dependencies_dicts = [] repository_metadata = metadata_util.get_repository_metadata_by_id(trans.app, id) if repository_metadata is None: log.debug(f"Invalid repository_metadata id received: {id}") return repository_dependencies_dicts metadata = repository_metadata.metadata if metadata is None: log.debug(f"The repository_metadata record with id {id} has no metadata.") return repository_dependencies_dicts if "repository_dependencies" in metadata: rd_tups = metadata["repository_dependencies"]["repository_dependencies"] for rd_tup in rd_tups: tool_shed, name, owner, changeset_revision = rd_tup[0:4] repository_dependency = repository_util.get_repository_by_name_and_owner(trans.app, name, owner) if repository_dependency is None: log.dbug(f"Cannot locate repository dependency {name} owned by {owner}.") continue repository_dependency_id = trans.security.encode_id(repository_dependency.id) repository_dependency_repository_metadata = metadata_util.get_repository_metadata_by_changeset_revision( trans.app, repository_dependency_id, changeset_revision ) if repository_dependency_repository_metadata is None: # The changeset_revision column in the repository_metadata table has been updated with a new # value value, so find the changeset_revision to which we need to update. new_changeset_revision = metadata_util.get_next_downloadable_changeset_revision( trans.app, repository_dependency, changeset_revision ) if new_changeset_revision != changeset_revision: repository_dependency_repository_metadata = ( metadata_util.get_repository_metadata_by_changeset_revision( trans.app, repository_dependency_id, new_changeset_revision ) ) changeset_revision = new_changeset_revision else: decoded_repository_dependency_id = trans.security.decode_id(repository_dependency_id) debug_msg = ( "Cannot locate repository_metadata with id %d for repository dependency %s owned by %s " % (decoded_repository_dependency_id, str(name), str(owner)) ) debug_msg += f"using either of these changeset_revisions: {changeset_revision}, {new_changeset_revision}." log.debug(debug_msg) continue repository_dependency_metadata_dict = repository_dependency_repository_metadata.to_dict( view="element", value_mapper=self.__get_value_mapper(trans) ) repository_dependency_dict = repository_dependency.to_dict( view="element", value_mapper=self.__get_value_mapper(trans) ) # We need to be careful with the entries in our repository_dependency_dict here since this Tool Shed API # controller is working with repository_metadata records. The above to_dict() method returns a dictionary # with an id entry for the repository record. However, all of the other methods in this controller have # the id entry associated with a repository_metadata record id. To avoid confusion, we'll update the # repository_dependency_metadata_dict with entries from the repository_dependency_dict without using the # Python dictionary update() method because we do not want to overwrite existing entries. for k, v in repository_dependency_dict.items(): if k not in repository_dependency_metadata_dict: repository_dependency_metadata_dict[k] = v repository_dependency_metadata_dict["url"] = web.url_for( controller="repositories", action="show", id=repository_dependency_id ) repository_dependencies_dicts.append(repository_dependency_metadata_dict) return repository_dependencies_dicts
[docs] @web.legacy_expose_api_anonymous def show(self, trans, id, **kwd): """ GET /api/repository_revisions/{encoded_repository_metadata_id} Displays information about a repository_metadata record in the Tool Shed. :param id: the encoded id of the `RepositoryMetadata` object """ # Example URL: http://localhost:9009/api/repository_revisions/bb125606ff9ea620 repository_metadata = metadata_util.get_repository_metadata_by_id(trans.app, id) if repository_metadata is None: log.debug(f"Cannot locate repository_metadata with id {id}") return {} encoded_repository_id = trans.security.encode_id(repository_metadata.repository_id) repository_metadata_dict = repository_metadata.to_dict( view="element", value_mapper=self.__get_value_mapper(trans) ) repository_metadata_dict["url"] = web.url_for( controller="repositories", action="show", id=encoded_repository_id ) return repository_metadata_dict
[docs] @web.legacy_expose_api def update(self, trans, payload, **kwd): """ PUT /api/repository_revisions/{encoded_repository_metadata_id}/{payload} Updates the value of specified columns of the repository_metadata table based on the key / value pairs in payload. :param id: the encoded id of the `RepositoryMetadata` object """ repository_metadata_id = kwd.get("id", None) if repository_metadata_id is None: raise HTTPBadRequest(detail="Missing required parameter 'id'.") repository_metadata = metadata_util.get_repository_metadata_by_id(trans.app, repository_metadata_id) if repository_metadata is None: decoded_repository_metadata_id = trans.security.decode_id(repository_metadata_id) log.debug(f"Cannot locate repository_metadata with id {decoded_repository_metadata_id}") return {} else: decoded_repository_metadata_id = repository_metadata.id flush_needed = False for key, new_value in payload.items(): if hasattr(repository_metadata, key): # log information when setting attributes associated with the Tool Shed's install and test framework. if key in ["includes_tools", "missing_test_components"]: log.debug( "Setting repository_metadata column %s to value %s for changeset_revision %s via the Tool Shed API.", key, new_value, repository_metadata.changeset_revision, ) setattr(repository_metadata, key, new_value) flush_needed = True if flush_needed: log.debug( "Updating repository_metadata record with id %s and changeset_revision %s.", decoded_repository_metadata_id, repository_metadata.changeset_revision, ) trans.sa_session.add(repository_metadata) with transaction(trans.sa_session): trans.sa_session.commit() trans.sa_session.refresh(repository_metadata) repository_metadata_dict = repository_metadata.to_dict( view="element", value_mapper=self.__get_value_mapper(trans) ) repository_metadata_dict["url"] = web.url_for( controller="repository_revisions", action="show", id=repository_metadata_id ) return repository_metadata_dict
[docs]def get_repository_metadata(session, downloadable, malicious, missing_test_components, includes_tools): stmt = select(RepositoryMetadata) if downloadable is not None: stmt = stmt.where(RepositoryMetadata.downloadable == util.asbool(downloadable)) if malicious is not None: stmt = stmt.where(RepositoryMetadata.malicious == util.asbool(malicious)) if missing_test_components is not None: stmt = stmt.where(RepositoryMetadata.missing_test_components == util.asbool(missing_test_components)) if includes_tools is not None: stmt = stmt.where(RepositoryMetadata.includes_tools == util.asbool(includes_tools)) stmt = stmt.order_by(RepositoryMetadata.repository_id.desc()) return session.scalars(stmt)