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 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.legacy_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(trans.app, 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(app=trans.app, 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' : 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 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(trans.app.model.RepositoryMetadata) \ .filter(and_(*clause_list)) \ .order_by(trans.app.model.RepositoryMetadata.table.c.repository_id.desc()): 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('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(trans.app, name, owner) if repository_dependency is None: log.dbug('Cannot locate repository dependency %s owned by %s.' % (name, 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 += '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.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('Cannot locate repository_metadata with id %s' % str(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('Cannot locate repository_metadata with id %s' % str(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.' % (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