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.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 (
from galaxy.web import expose_api
from galaxy.webapps.base.controller import (

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 parent_id != hdca_id and 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