Warning

This document is for an in-development version 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.galaxy.api.dataset_collections

from logging import getLogger

import routes

from galaxy import managers
from galaxy.exceptions import ObjectNotFound
from galaxy.managers.base import decode_id
from galaxy.managers.collections_util import (
    api_payload_to_create_params,
    dictify_dataset_collection_instance,
    dictify_element_reference
)
from galaxy.web import expose_api
from galaxy.webapps.base.controller import (
    BaseAPIController,
    UsesLibraryMixinItems
)

log = getLogger(__name__)


[docs]class DatasetCollectionsController( BaseAPIController, UsesLibraryMixinItems, ):
[docs] def __init__(self, app): super().__init__(app) self.history_manager = managers.histories.HistoryManager(app)
[docs] @expose_api def index(self, trans, **kwd): trans.response.status = 501 return 'not implemented'
[docs] @expose_api def create(self, trans, payload, **kwd): """ * POST /api/dataset_collections: create a new dataset collection instance. :type payload: dict :param payload: (optional) dictionary structure containing: * collection_type: dataset colltion type to create. * instance_type: Instance type - 'history' or 'library'. * name: the new dataset collections's name * datasets: object describing datasets for collection :rtype: dict :returns: element view of new dataset collection """ # TODO: Error handling... create_params = api_payload_to_create_params(payload) instance_type = payload.pop("instance_type", "history") if instance_type == "history": history_id = payload.get('history_id') history_id = decode_id(self.app, history_id) history = self.history_manager.get_owned(history_id, trans.user, current_history=trans.history) create_params["parent"] = history elif instance_type == "library": folder_id = payload.get('folder_id') library_folder = self.get_library_folder(trans, folder_id, check_accessible=True) self.check_user_can_add_to_library_item(trans, library_folder, check_accessible=False) create_params["parent"] = library_folder else: trans.status = 501 return dataset_collection_instance = self.__service(trans).create(trans=trans, **create_params) return dictify_dataset_collection_instance(dataset_collection_instance, security=trans.security, parent=create_params["parent"])
[docs] @expose_api def show(self, trans, instance_type, id, **kwds): dataset_collection_instance = self.__service(trans).get_dataset_collection_instance( trans, id=id, instance_type=instance_type, ) if instance_type == 'history': parent = dataset_collection_instance.history elif instance_type == 'library': parent = dataset_collection_instance.folder else: trans.status = 501 return return dictify_dataset_collection_instance( dataset_collection_instance, security=trans.security, parent=parent, view='element' )
[docs] @expose_api def contents(self, trans, hdca_id, parent_id, instance_type='history', limit=None, offset=None, **kwds): """ * GET /api/dataset_collection/{hdca_id}/contents/{parent_id}?limit=100&offset=0 Shows direct child contents of indicated dataset collection parent id :type string: encoded string id :param id: HDCA.id :type string: encoded string id :param parent_id: parent dataset_collection.id for the dataset contents to be viewed :type integer: int :param limit: pagination limit for returned dataset collection elements :type integer: int :param offset: pagination offset for returned dataset collection elements :rtype: list :returns: list of dataset collection elements and contents """ svc = self.__service(trans) encode_id = trans.app.security.encode_id # validate HDCA for current user, will throw error if not permitted # TODO: refactor get_dataset_collection_instance hdca = svc.get_dataset_collection_instance(trans, id=hdca_id, check_ownership=True, instance_type=instance_type) # check to make sure the dsc is part of the validated hdca decoded_parent_id = decode_id(self.app, parent_id) if not hdca.contains_collection(decoded_parent_id): errmsg = 'Requested dataset collection is not contained within indicated history content' raise ObjectNotFound(errmsg) # retrieve contents contents_qry = svc.get_collection_contents_qry(decoded_parent_id, limit=limit, offset=offset) # dictify and tack on a collection_url for drilling down into nested collections def process_element(dsc_element): result = dictify_element_reference(dsc_element, recursive=False, security=trans.security) if result["element_type"] == "dataset_collection": result["object"]["contents_url"] = routes.url_for('contents_dataset_collection', hdca_id=encode_id(hdca.id), parent_id=encode_id(result["object"]["id"])) trans.security.encode_all_ids(result, recursive=True) return result results = contents_qry.with_session(trans.sa_session()).all() return [process_element(el) for el in results]
def __service(self, trans): service = trans.app.dataset_collections_service return service