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.galaxy.services.dataset_collections

from logging import getLogger
from typing import (
    List,
    Optional,
    Set,
    TYPE_CHECKING,
    Union,
)

from pydantic import (
    Extra,
    Field,
    ValidationError,
)
from typing_extensions import Literal

from galaxy import exceptions
from galaxy.datatypes.registry import Registry
from galaxy.managers.collections import DatasetCollectionManager
from galaxy.managers.collections_util import (
    api_payload_to_create_params,
    dictify_dataset_collection_instance,
    dictify_element_reference,
)
from galaxy.managers.context import ProvidesHistoryContext
from galaxy.managers.hdcas import HDCAManager
from galaxy.managers.histories import HistoryManager
from galaxy.model import DatasetCollectionElement
from galaxy.schema.fields import (
    DecodedDatabaseIdField,
    ModelClassField,
)
from galaxy.schema.schema import (
    AnyHDCA,
    CreateNewCollectionPayload,
    DatasetCollectionInstanceType,
    DCESummary,
    DCEType,
    HDCADetailed,
    Model,
    TagCollection,
)
from galaxy.security.idencoding import IdEncodingHelper
from galaxy.webapps.base.controller import UsesLibraryMixinItems
from galaxy.webapps.galaxy.services.base import ServiceBase

if TYPE_CHECKING:
    from galaxy.model import (
        HistoryDatasetCollectionAssociation,
        LibraryDatasetCollectionAssociation,
    )

log = getLogger(__name__)


[docs]class UpdateCollectionAttributePayload(Model): """Contains attributes that can be updated for all elements in a dataset collection.""" dbkey: str = Field(..., description="TODO")
[docs] class Config: extra = Extra.forbid # will cause validation to fail if extra attributes are included,
[docs]class DatasetCollectionAttributesResult(Model): dbkey: str = Field(..., description="TODO") # Are the following fields really used/needed? extension: str = Field(..., description="The dataset file extension.", example="txt") model_class: Literal["HistoryDatasetCollectionAssociation"] = ModelClassField("HistoryDatasetCollectionAssociation") dbkeys: Optional[Set[str]] extensions: Optional[Set[str]] tags: TagCollection
[docs]class SuitableConverter(Model): tool_id: str = Field(..., description="The ID of the tool that can perform the type conversion.") name: str = Field(..., description="The name of the converter.") target_type: str = Field(..., description="The type to convert to.") original_type: str = Field(..., description="The type to convert from.")
[docs]class SuitableConverters(Model): """Collection of converters that can be used on a particular dataset collection.""" __root__: List[SuitableConverter]
[docs]class DatasetCollectionContentElements(Model): """Represents a collection of elements contained in the dataset collection.""" __root__: List[DCESummary]
[docs]class DatasetCollectionsService(ServiceBase, UsesLibraryMixinItems):
[docs] def __init__( self, security: IdEncodingHelper, history_manager: HistoryManager, hdca_manager: HDCAManager, collection_manager: DatasetCollectionManager, datatypes_registry: Registry, ): super().__init__(security) self.history_manager = history_manager self.hdca_manager = hdca_manager self.collection_manager = collection_manager self.datatypes_registry = datatypes_registry
[docs] def create(self, trans: ProvidesHistoryContext, payload: CreateNewCollectionPayload) -> HDCADetailed: """ Create a new dataset collection instance. :type payload: dict :param payload: (optional) dictionary structure containing: * collection_type: dataset collection 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.dict(exclude_unset=True)) if payload.instance_type == "history": if payload.history_id is None: raise exceptions.RequestParameterInvalidException("Parameter history_id is required.") history = self.history_manager.get_mutable(payload.history_id, trans.user, current_history=trans.history) create_params["parent"] = history create_params["history"] = history elif payload.instance_type == "library" and payload.folder_id: library_folder = self.get_library_folder(trans, payload.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: raise exceptions.RequestParameterInvalidException() dataset_collection_instance = self.collection_manager.create(trans=trans, **create_params) rval = dictify_dataset_collection_instance( dataset_collection_instance, security=trans.security, url_builder=trans.url_builder, parent=create_params["parent"], ) return rval
[docs] def copy( self, trans: ProvidesHistoryContext, id: DecodedDatabaseIdField, payload: UpdateCollectionAttributePayload ): """ Iterate over all datasets of a collection and copy datasets with new attributes to a new collection. e.g attributes = {'dbkey': 'dm3'} """ self.collection_manager.copy( trans, trans.history, "hdca", id, copy_elements=True, dataset_instance_attributes=payload.dict() )
[docs] def attributes( self, trans: ProvidesHistoryContext, id: DecodedDatabaseIdField, instance_type: DatasetCollectionInstanceType = "history", ) -> DatasetCollectionAttributesResult: """ Returns dbkey/extension for collection elements """ dataset_collection_instance = self.collection_manager.get_dataset_collection_instance( trans, instance_type, id, check_ownership=True ) rval = dataset_collection_instance.to_dict(view="dbkeysandextensions") return rval
[docs] def suitable_converters( self, trans: ProvidesHistoryContext, id: DecodedDatabaseIdField, instance_type: DatasetCollectionInstanceType = "history", ) -> SuitableConverters: """ Returns suitable converters for all datatypes in collection """ rval = self.collection_manager.get_converters_for_collection(trans, id, self.datatypes_registry, instance_type) return rval
[docs] def show( self, trans: ProvidesHistoryContext, id: DecodedDatabaseIdField, instance_type: DatasetCollectionInstanceType = "history", ) -> AnyHDCA: """ Returns information about a particular dataset collection. """ dataset_collection_instance: Union["HistoryDatasetCollectionAssociation", "LibraryDatasetCollectionAssociation"] if instance_type == "history": dataset_collection_instance = self.collection_manager.get_dataset_collection_instance(trans, "history", id) parent = dataset_collection_instance.history elif instance_type == "library": dataset_collection_instance = self.collection_manager.get_dataset_collection_instance(trans, "library", id) parent = dataset_collection_instance.folder else: raise exceptions.RequestParameterInvalidException() rval = dictify_dataset_collection_instance( dataset_collection_instance, security=trans.security, url_builder=trans.url_builder, parent=parent, view="element", ) return rval
[docs] def dce_content(self, trans: ProvidesHistoryContext, dce_id: DecodedDatabaseIdField) -> DCESummary: dce: Optional[DatasetCollectionElement] = trans.model.session.query(DatasetCollectionElement).get(dce_id) if not dce: raise exceptions.ObjectNotFound("No DatasetCollectionElement found") if not trans.user_is_admin: collection = dce.child_collection or dce.collection if not trans.app.security_agent.can_access_collection(trans.get_current_user_roles(), collection): raise exceptions.ItemAccessibilityException("Collection not accessible by user.") serialized_dce = dictify_element_reference(dce, recursive=False, security=trans.security) return trans.security.encode_all_ids(serialized_dce, recursive=True)
[docs] def contents( self, trans: ProvidesHistoryContext, hdca_id: DecodedDatabaseIdField, parent_id: DecodedDatabaseIdField, instance_type: DatasetCollectionInstanceType = "history", limit: Optional[int] = None, offset: Optional[int] = None, ) -> DatasetCollectionContentElements: """ 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 """ # validate HDCA for current user, will throw error if not permitted # TODO: refactor get_dataset_collection_instance if instance_type != "history": raise exceptions.RequestParameterInvalidException( "Parameter instance_type not being 'history' is not yet implemented." ) hdca: "HistoryDatasetCollectionAssociation" = self.collection_manager.get_dataset_collection_instance( trans, "history", hdca_id ) # check to make sure the dsc is part of the validated hdca if not hdca.contains_collection(parent_id): raise exceptions.ObjectNotFound( "Requested dataset collection is not contained within indicated history content" ) # retrieve contents contents = self.collection_manager.get_collection_contents(trans, parent_id, limit=limit, offset=offset) # dictify and tack on a collection_url for drilling down into nested collections def serialize_element(dsc_element) -> DCESummary: result = dictify_element_reference(dsc_element, recursive=False, security=trans.security) if result["element_type"] == DCEType.dataset_collection: assert trans.url_builder result["object"]["contents_url"] = trans.url_builder( "contents_dataset_collection", hdca_id=self.encode_id(hdca.id), parent_id=self.encode_id(result["object"]["id"]), ) trans.security.encode_all_ids(result, recursive=True) return result rval = [serialize_element(el) for el in contents] try: return DatasetCollectionContentElements.construct(__root__=rval) except ValidationError: log.exception( f"Serializing DatasetCollectionContentsElements failed. Collection is populated: {hdca.collection.populated}" ) raise