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