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.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)