Source code for galaxy.webapps.tool_shed.api.repository_revisions

import logging

from sqlalchemy import and_

from galaxy import util
from galaxy import web
from galaxy.web.base.controller import BaseAPIController, HTTPBadRequest
from tool_shed.capsule import capsule_manager
from tool_shed.util import metadata_util
from tool_shed.util import repository_util

log = logging.getLogger(__name__)

[docs]class RepositoryRevisionsController(BaseAPIController): """RESTful controller for interactions with tool shed repository revisions."""
[docs] @web.expose_api_anonymous def export(self, trans, payload, **kwd): """ POST /api/repository_revisions/export Creates and saves a gzip compressed tar archive of a repository and optionally all of its repository dependencies. The following parameters are included in the payload. :param tool_shed_url (required): the base URL of the Tool Shed from which the Repository is to be exported :param name (required): the name of the Repository :param owner (required): the owner of the Repository :param changeset_revision (required): the changeset_revision of the RepositoryMetadata object associated with the Repository :param export_repository_dependencies (optional): whether to export repository dependencies - defaults to False :param download_dir (optional): the local directory to which to download the archive - defaults to /tmp """ tool_shed_url = payload.get('tool_shed_url', '') if not tool_shed_url: raise HTTPBadRequest(detail="Missing required parameter 'tool_shed_url'.") tool_shed_url = tool_shed_url.rstrip('/') name = payload.get('name', '') if not name: raise HTTPBadRequest(detail="Missing required parameter 'name'.") owner = payload.get('owner', '') if not owner: raise HTTPBadRequest(detail="Missing required parameter 'owner'.") changeset_revision = payload.get('changeset_revision', '') if not changeset_revision: raise HTTPBadRequest(detail="Missing required parameter 'changeset_revision'.") export_repository_dependencies = payload.get('export_repository_dependencies', False) # We'll currently support only gzip-compressed tar archives. export_repository_dependencies = util.asbool(export_repository_dependencies) # Get the repository information. repository = repository_util.get_repository_by_name_and_owner(, name, owner) if repository is None: error_message = 'Cannot locate repository with name %s and owner %s,' % (str(name), str(owner)) log.debug(error_message) return None, error_message erm = capsule_manager.ExportRepositoryManager(, user=trans.user, tool_shed_url=tool_shed_url, repository=repository, changeset_revision=changeset_revision, export_repository_dependencies=export_repository_dependencies, using_api=True) return erm.export_repository()
def __get_value_mapper(self, trans): value_mapper = {'id' :, 'repository_id' :, 'user_id' :} return value_mapper
[docs] @web.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 repository_metadata_dicts = [] # Build up an anded clause list of filters. clause_list = [] # Filter by downloadable if received. downloadable = kwd.get('downloadable', None) if downloadable is not None: clause_list.append(trans.model.RepositoryMetadata.table.c.downloadable == util.asbool(downloadable)) # Filter by malicious if received. malicious = kwd.get('malicious', None) if malicious is not None: clause_list.append(trans.model.RepositoryMetadata.table.c.malicious == util.asbool(malicious)) # Filter by missing_test_components if received. missing_test_components = kwd.get('missing_test_components', None) if missing_test_components is not None: clause_list.append(trans.model.RepositoryMetadata.table.c.missing_test_components == util.asbool(missing_test_components)) # Filter by includes_tools if received. includes_tools = kwd.get('includes_tools', None) if includes_tools is not None: clause_list.append(trans.model.RepositoryMetadata.table.c.includes_tools == util.asbool(includes_tools)) for repository_metadata in trans.sa_session.query( \ .filter(and_(*clause_list)) \ .order_by( 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', repository_metadata_dicts.append(repository_metadata_dict) return repository_metadata_dicts
[docs] @web.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(, id) if repository_metadata is None: log.debug('Invalid repository_metadata id received: %s' % str(id)) return repository_dependencies_dicts metadata = repository_metadata.metadata if metadata is None: log.debug('The repository_metadata record with id %s has no metadata.' % str(id)) 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(, name, owner) if repository_dependency is None: log.dbug('Cannot locate repository dependency %s owned by %s.' % (name, owner)) continue repository_dependency_id = repository_dependency_repository_metadata = \ metadata_util.get_repository_metadata_by_changeset_revision(, 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(, repository_dependency, changeset_revision) if new_changeset_revision != changeset_revision: repository_dependency_repository_metadata = \ metadata_util.get_repository_metadata_by_changeset_revision(, repository_dependency_id, new_changeset_revision) changeset_revision = new_changeset_revision else: decoded_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 += 'using either of these changeset_revisions: %s, %s.' % \ (str(changeset_revision), str(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.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(, id) if repository_metadata is None: log.debug('Cannot locate repository_metadata with id %s' % str(id)) return {} encoded_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.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(, repository_metadata_id) if repository_metadata is None: decoded_repository_metadata_id = log.debug('Cannot locate repository_metadata with id %s' % str(decoded_repository_metadata_id)) return {} else: decoded_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.' % (str(key), str(new_value), str(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.' % (str(decoded_repository_metadata_id), str(repository_metadata.changeset_revision))) trans.sa_session.add(repository_metadata) trans.sa_session.flush() 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