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.api.dataset_collections

from logging import getLogger
from typing import Optional

from fastapi import (
    Body,
    Path,
    Query,
)

from galaxy.managers.context import ProvidesHistoryContext
from galaxy.schema.fields import EncodedDatabaseIdField
from galaxy.schema.schema import (
    AnyHDCA,
    CreateNewCollectionPayload,
    DatasetCollectionInstanceType,
    HDCADetailed,
)
from galaxy.webapps.galaxy.services.dataset_collections import (
    DatasetCollectionAttributesResult,
    DatasetCollectionContentElements,
    DatasetCollectionsService,
    SuitableConverters,
    UpdateCollectionAttributePayload,
)
from . import (
    depends,
    DependsOnTrans,
    Router,
)

log = getLogger(__name__)

router = Router(tags=["dataset collections"])

DatasetCollectionIdPathParam: EncodedDatabaseIdField = Path(
    ..., description="The encoded identifier of the dataset collection."
)

InstanceTypeQueryParam: DatasetCollectionInstanceType = Query(
    default=DatasetCollectionInstanceType.history,
    description="The type of collection instance. Either `history` (default) or `library`.",
)


[docs]@router.cbv class FastAPIDatasetCollections: service: DatasetCollectionsService = depends(DatasetCollectionsService)
[docs] @router.post( "/api/dataset_collections", summary="Create a new dataset collection instance.", ) def create( self, trans: ProvidesHistoryContext = DependsOnTrans, payload: CreateNewCollectionPayload = Body(...), ) -> HDCADetailed: return self.service.create(trans, payload)
[docs] @router.post( "/api/dataset_collections/{id}/copy", summary="Copy the given collection datasets to a new collection using a new `dbkey` attribute.", ) def copy( self, trans: ProvidesHistoryContext = DependsOnTrans, id: EncodedDatabaseIdField = Path(..., description="The ID of the dataset collection to copy."), payload: UpdateCollectionAttributePayload = Body(...), ): self.service.copy(trans, id, payload)
[docs] @router.get( "/api/dataset_collections/{id}/attributes", summary="Returns `dbkey`/`extension` attributes for all the collection elements.", ) def attributes( self, trans: ProvidesHistoryContext = DependsOnTrans, id: EncodedDatabaseIdField = DatasetCollectionIdPathParam, instance_type: DatasetCollectionInstanceType = InstanceTypeQueryParam, ) -> DatasetCollectionAttributesResult: return self.service.attributes(trans, id, instance_type)
[docs] @router.get( "/api/dataset_collections/{id}/suitable_converters", summary="Returns a list of applicable converters for all datatypes in the given collection.", ) def suitable_converters( self, trans: ProvidesHistoryContext = DependsOnTrans, id: EncodedDatabaseIdField = DatasetCollectionIdPathParam, instance_type: DatasetCollectionInstanceType = InstanceTypeQueryParam, ) -> SuitableConverters: return self.service.suitable_converters(trans, id, instance_type)
[docs] @router.get( "/api/dataset_collections/{id}", summary="Returns detailed information about the given collection.", ) def show( self, trans: ProvidesHistoryContext = DependsOnTrans, id: EncodedDatabaseIdField = DatasetCollectionIdPathParam, instance_type: DatasetCollectionInstanceType = InstanceTypeQueryParam, ) -> AnyHDCA: return self.service.show(trans, id, instance_type)
[docs] @router.get( "/api/dataset_collections/{hdca_id}/contents/{parent_id}", name="contents_dataset_collection", summary="Returns direct child contents of indicated dataset collection parent ID.", ) def contents( self, trans: ProvidesHistoryContext = DependsOnTrans, hdca_id: EncodedDatabaseIdField = DatasetCollectionIdPathParam, parent_id: EncodedDatabaseIdField = Path( ..., description="Parent collection ID describing what collection the contents belongs to.", ), instance_type: DatasetCollectionInstanceType = InstanceTypeQueryParam, limit: Optional[int] = Query( default=None, description="The maximum number of content elements to return.", ), offset: Optional[int] = Query( default=None, description="The number of content elements that will be skipped before returning.", ), ) -> DatasetCollectionContentElements: return self.service.contents(trans, hdca_id, parent_id, instance_type, limit, offset)