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.history_contents

"""
API operations on the contents of a history.
"""
import logging
from typing import (
    Any,
    Dict,
    List,
    Optional,
    Union,
)

from fastapi import (
    Body,
    Depends,
    Header,
    Path,
    Query,
)
from pydantic.error_wrappers import ValidationError
from starlette import status
from starlette.responses import (
    Response,
    StreamingResponse,
)

from galaxy import util
from galaxy.managers.context import ProvidesHistoryContext
from galaxy.schema import (
    FilterQueryParams,
    SerializationParams,
    ValueFilterQueryParams,
)
from galaxy.schema.fields import DecodedDatabaseIdField
from galaxy.schema.schema import (
    AnyHistoryContentItem,
    AnyJobStateSummary,
    AsyncFile,
    AsyncTaskResultSummary,
    DatasetAssociationRoles,
    DatasetSourceType,
    DeleteHistoryContentPayload,
    DeleteHistoryContentResult,
    HistoryContentBulkOperationPayload,
    HistoryContentBulkOperationResult,
    HistoryContentsArchiveDryRunResult,
    HistoryContentsResult,
    HistoryContentsWithStatsResult,
    HistoryContentType,
    MaterializeDatasetInstanceAPIRequest,
    MaterializeDatasetInstanceRequest,
    StoreExportPayload,
    UpdateDatasetPermissionsPayload,
    UpdateHistoryContentsBatchPayload,
    UpdateHistoryContentsPayload,
    WriteStoreToPayload,
)
from galaxy.web.framework.decorators import validation_error_to_message_exception
from galaxy.webapps.galaxy.api import (
    depends,
    DependsOnTrans,
    Router,
)
from galaxy.webapps.galaxy.api.common import (
    get_filter_query_params,
    get_update_permission_payload,
    get_value_filter_query_params,
    query_serialization_params,
)
from galaxy.webapps.galaxy.services.history_contents import (
    CreateHistoryContentFromStore,
    CreateHistoryContentPayload,
    HistoriesContentsService,
    HistoryContentsIndexJobsSummaryParams,
    HistoryContentsIndexParams,
    LegacyHistoryContentsIndexParams,
)

log = logging.getLogger(__name__)


router = Router(tags=["histories"])

HistoryIDPathParam: DecodedDatabaseIdField = Path(..., title="History ID", description="The ID of the History.")

HistoryItemIDPathParam: DecodedDatabaseIdField = Path(
    ..., title="History Item ID", description="The ID of the item (`HDA`/`HDCA`) contained in the history."
)

HistoryHDCAIDPathParam: DecodedDatabaseIdField = Path(
    ..., title="History Dataset Collection ID", description="The ID of the `HDCA` contained in the history."
)


[docs]def ContentTypeQueryParam(default: Optional[HistoryContentType]): return Query( default=default, title="Content Type", description="The type of the target history element.", example=HistoryContentType.dataset, )
ContentTypePathParam = Path( title="Content Type", description="The type of the target history element.", example=HistoryContentType.dataset, ) FuzzyCountQueryParam = Query( default=None, title="Fuzzy Count", description=( "This value can be used to broadly restrict the magnitude " "of the number of elements returned via the API for large " "collections. The number of actual elements returned may " 'be "a bit" more than this number or "a lot" less - varying ' "on the depth of nesting, balance of nesting at each level, " "and size of target collection. The consumer of this API should " "not expect a stable number or pre-calculable number of " "elements to be produced given this parameter - the only " "promise is that this API will not respond with an order " "of magnitude more elements estimated with this value. " 'The UI uses this parameter to fetch a "balanced" concept of ' 'the "start" of large collections at every depth of the ' "collection." ), ) PurgeQueryParam = Query( default=False, title="Purge", description="Whether to remove from disk the target HDA or child HDAs of the target HDCA.", deprecated=True, ) RecursiveQueryParam = Query( default=False, title="Recursive", description="When deleting a dataset collection, whether to also delete containing datasets.", deprecated=True, ) StopJobQueryParam = Query( default=False, title="Stop Job", description="Whether to stop the creating job if all outputs of the job have been deleted.", deprecated=True, ) CONTENT_DELETE_RESPONSES = { 200: { "description": "Request has been executed.", "model": DeleteHistoryContentResult, }, 202: { "description": "Request accepted, processing will finish later.", "model": DeleteHistoryContentResult, }, }
[docs]def get_index_query_params( v: Optional[str] = Query( # Should this be deprecated at some point and directly use the latest version by default? default=None, title="Version", description=( "Only `dev` value is allowed. Set it to use the latest version of this endpoint. " "**All parameters marked as `deprecated` will be ignored when this parameter is set.**" ), example="dev", ), dataset_details: Optional[str] = Query( default=None, alias="details", title="Dataset Details", description=( "A comma-separated list of encoded dataset IDs that will return additional (full) details " "instead of the *summary* default information." ), deprecated=True, # TODO: remove 'dataset_details' when the UI doesn't need it ), ) -> HistoryContentsIndexParams: """This function is meant to be used as a dependency to render the OpenAPI documentation correctly""" return parse_index_query_params( v=v, dataset_details=dataset_details, )
[docs]def parse_index_query_params( v: Optional[str] = None, dataset_details: Optional[str] = None, **_, # Additional params are ignored ) -> HistoryContentsIndexParams: """Parses query parameters for the history contents `index` operation and returns a model containing the values in the correct type.""" try: return HistoryContentsIndexParams( v=v, dataset_details=parse_dataset_details(dataset_details), ) except ValidationError as e: raise validation_error_to_message_exception(e)
LegacyIdsQueryParam = Query( default=None, title="IDs", description=( "A comma-separated list of encoded `HDA/HDCA` IDs. If this list is provided, only information about the " "specific datasets will be returned. Also, setting this value will return `all` details of the content item." ), deprecated=True, ) LegacyTypesQueryParam = Query( default=None, title="Types", description=( "A list or comma-separated list of kinds of contents to return " "(currently just `dataset` and `dataset_collection` are available). " "If unset, all types will be returned." ), deprecated=True, ) LegacyDetailsQueryParam = Query( default=None, title="Details", description=("Legacy name for the `dataset_details` parameter."), deprecated=True, ) LegacyDeletedQueryParam = Query( default=None, title="Deleted", description="Whether to return deleted or undeleted datasets only. Leave unset for both.", deprecated=True, ) LegacyVisibleQueryParam = Query( default=None, title="Visible", description="Whether to return visible or hidden datasets only. Leave unset for both.", deprecated=True, ) LegacyShareableQueryParam = Query( default=None, title="Shareable", description="Whether to return only shareable or not shareable datasets. Leave unset for both.", ) ArchiveFilenamePathParam = Path( description="The name that the Archive will have (defaults to history name).", ) ArchiveFilenameQueryParam = Query( default=None, description="The name that the Archive will have (defaults to history name).", ) DryRunQueryParam = Query( default=True, description="Whether to return the archive and file paths only (as JSON) and not an actual archive file.", )
[docs]def get_legacy_index_query_params( ids: Optional[str] = LegacyIdsQueryParam, types: Optional[List[str]] = LegacyTypesQueryParam, details: Optional[str] = LegacyDetailsQueryParam, deleted: Optional[bool] = LegacyDeletedQueryParam, visible: Optional[bool] = LegacyVisibleQueryParam, shareable: Optional[bool] = LegacyShareableQueryParam, ) -> LegacyHistoryContentsIndexParams: """This function is meant to be used as a dependency to render the OpenAPI documentation correctly""" return parse_legacy_index_query_params( ids=ids, types=types, details=details, deleted=deleted, visible=visible, shareable=shareable, )
[docs]def parse_legacy_index_query_params( ids: Optional[str] = None, types: Optional[Union[List[str], str]] = None, details: Optional[str] = None, deleted: Optional[bool] = None, visible: Optional[bool] = None, shareable: Optional[bool] = None, **_, # Additional params are ignored ) -> LegacyHistoryContentsIndexParams: """Parses (legacy) query parameters for the history contents `index` operation and returns a model containing the values in the correct type.""" if types: content_types = parse_content_types(types) else: content_types = [e for e in HistoryContentType] id_list = None if ids: id_list = util.listify(ids) # If explicit ids given, always used detailed result. dataset_details = "all" else: dataset_details = parse_dataset_details(details) try: return LegacyHistoryContentsIndexParams( types=content_types, ids=id_list, deleted=deleted, visible=visible, shareable=shareable, dataset_details=dataset_details, ) except ValidationError as e: raise validation_error_to_message_exception(e)
[docs]def parse_content_types(types: Union[List[str], str]) -> List[HistoryContentType]: if isinstance(types, list) and len(types) == 1: # Support ?types=dataset,dataset_collection content_types = util.listify(types[0]) else: # Support ?types=dataset&types=dataset_collection content_types = util.listify(types) return [HistoryContentType[content_type] for content_type in content_types]
[docs]def parse_dataset_details(details: Optional[str]): """Parses the different values that the `dataset_details` parameter can have from a string.""" dataset_details = None if details is not None and details != "all": dataset_details = set(util.listify(details)) else: # either None or 'all' dataset_details = details # type: ignore return dataset_details
[docs]def get_index_jobs_summary_params( ids: Optional[str] = Query( default=None, title="IDs", description=( "A comma-separated list of encoded ids of job summary objects to return - if `ids` " "is specified types must also be specified and have same length." ), ), types: Optional[str] = Query( default=None, title="Types", description=( "A comma-separated list of type of object represented by elements in the `ids` array - any of " "`Job`, `ImplicitCollectionJob`, or `WorkflowInvocation`." ), ), ) -> HistoryContentsIndexJobsSummaryParams: """This function is meant to be used as a dependency to render the OpenAPI documentation correctly""" return parse_index_jobs_summary_params( ids=ids, types=types, )
[docs]def parse_index_jobs_summary_params( ids: Optional[str] = None, types: Optional[str] = None, **_, # Additional params are ignored ) -> HistoryContentsIndexJobsSummaryParams: """Parses query parameters for the history contents `index_jobs_summary` operation and returns a model containing the values in the correct type.""" return HistoryContentsIndexJobsSummaryParams(ids=util.listify(ids), types=util.listify(types))
[docs]@router.cbv class FastAPIHistoryContents: service: HistoriesContentsService = depends(HistoriesContentsService)
[docs] @router.get( "/api/histories/{history_id}/contents/{type}s", summary="Returns the contents of the given history filtered by type.", operation_id="history_contents__index_typed", ) def index_typed( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, index_params: HistoryContentsIndexParams = Depends(get_index_query_params), type: HistoryContentType = ContentTypePathParam, legacy_params: LegacyHistoryContentsIndexParams = Depends(get_legacy_index_query_params), serialization_params: SerializationParams = Depends(query_serialization_params), filter_query_params: FilterQueryParams = Depends(get_filter_query_params), accept: str = Header(default="application/json", include_in_schema=False), ) -> Union[HistoryContentsResult, HistoryContentsWithStatsResult]: """ Return a list of either `HDA`/`HDCA` data for the history with the given ``ID``. - The contents can be filtered and queried using the appropriate parameters. - The amount of information returned for each item can be customized. **Note**: Anonymous users are allowed to get their current history contents. """ legacy_params.types = [type] items = self.service.index( trans, history_id, index_params, legacy_params, serialization_params, filter_query_params, accept, ) return items
[docs] @router.get( "/api/histories/{history_id}/contents", name="history_contents", summary="Returns the contents of the given history.", responses={ 200: { "description": ("The contents of the history that match the query."), "content": { "application/json": { "schema": { # HistoryContentsResult.schema(), "ref": "#/components/schemas/HistoryContentsResult" }, }, HistoryContentsWithStatsResult.__accept_type__: { "schema": { # HistoryContentsWithStatsResult.schema(), "ref": "#/components/schemas/HistoryContentsWithStatsResult" }, }, }, }, }, operation_id="history_contents__index", ) def index( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, index_params: HistoryContentsIndexParams = Depends(get_index_query_params), type: Optional[str] = Query(default=None, include_in_schema=False, deprecated=True), legacy_params: LegacyHistoryContentsIndexParams = Depends(get_legacy_index_query_params), serialization_params: SerializationParams = Depends(query_serialization_params), filter_query_params: FilterQueryParams = Depends(get_filter_query_params), accept: str = Header(default="application/json", include_in_schema=False), ) -> Union[HistoryContentsResult, HistoryContentsWithStatsResult]: """ Return a list of `HDA`/`HDCA` data for the history with the given ``ID``. - The contents can be filtered and queried using the appropriate parameters. - The amount of information returned for each item can be customized. **Note**: Anonymous users are allowed to get their current history contents. """ if type is not None: legacy_params.types = parse_content_types(type) items = self.service.index( trans, history_id, index_params, legacy_params, serialization_params, filter_query_params, accept, ) return items
[docs] @router.get( "/api/histories/{history_id}/contents/{type}s/{id}/jobs_summary", summary="Return detailed information about an `HDA` or `HDCAs` jobs.", ) def show_jobs_summary( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, id: DecodedDatabaseIdField = HistoryItemIDPathParam, type: HistoryContentType = ContentTypePathParam, ) -> AnyJobStateSummary: """Return detailed information about an `HDA` or `HDCAs` jobs. **Warning**: We allow anyone to fetch job state information about any object they can guess an encoded ID for - it isn't considered protected data. This keeps polling IDs as part of state calculation for large histories and collections as efficient as possible. """ return self.service.show_jobs_summary(trans, id, contents_type=type)
[docs] @router.get( "/api/histories/{history_id}/contents/{type}s/{id}", name="history_content_typed", summary="Return detailed information about a specific HDA or HDCA with the given `ID` within a history.", operation_id="history_contents__show", ) def show( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, id: DecodedDatabaseIdField = HistoryItemIDPathParam, type: HistoryContentType = ContentTypePathParam, fuzzy_count: Optional[int] = FuzzyCountQueryParam, serialization_params: SerializationParams = Depends(query_serialization_params), ) -> AnyHistoryContentItem: """ Return detailed information about an `HDA` or `HDCA` within a history. **Note**: Anonymous users are allowed to get their current history contents. """ return self.service.show( trans, id=id, serialization_params=serialization_params, contents_type=type, fuzzy_count=fuzzy_count, )
[docs] @router.get( "/api/histories/{history_id}/contents/{id}", name="history_content", summary="Return detailed information about an HDA within a history. ``/api/histories/{history_id}/contents/{type}s/{id}`` should be used instead.", deprecated=True, operation_id="history_contents__show_legacy", ) def show_legacy( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, type: HistoryContentType = ContentTypeQueryParam(default=HistoryContentType.dataset), id: DecodedDatabaseIdField = HistoryItemIDPathParam, fuzzy_count: Optional[int] = FuzzyCountQueryParam, serialization_params: SerializationParams = Depends(query_serialization_params), ) -> AnyHistoryContentItem: """ Return detailed information about an `HDA` or `HDCA` within a history. **Note**: Anonymous users are allowed to get their current history contents. """ return self.service.show( trans, id=id, serialization_params=serialization_params, contents_type=type, fuzzy_count=fuzzy_count, )
[docs] @router.post( "/api/histories/{history_id}/contents/{type}s/{id}/prepare_store_download", summary="Prepare a dataset or dataset collection for export-style download.", ) def prepare_store_download( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, id: DecodedDatabaseIdField = HistoryItemIDPathParam, type: HistoryContentType = ContentTypePathParam, payload: StoreExportPayload = Body(...), ) -> AsyncFile: return self.service.prepare_store_download( trans, id, contents_type=type, payload=payload, )
[docs] @router.post( "/api/histories/{history_id}/contents/{type}s/{id}/write_store", summary="Prepare a dataset or dataset collection for export-style download and write to supplied URI.", ) def write_store( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, id: DecodedDatabaseIdField = HistoryItemIDPathParam, type: HistoryContentType = ContentTypePathParam, payload: WriteStoreToPayload = Body(...), ) -> AsyncTaskResultSummary: return self.service.write_store( trans, id, contents_type=type, payload=payload, )
[docs] @router.get( "/api/histories/{history_id}/jobs_summary", summary="Return job state summary info for jobs, implicit groups jobs for collections or workflow invocations.", ) def index_jobs_summary( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, params: HistoryContentsIndexJobsSummaryParams = Depends(get_index_jobs_summary_params), ) -> List[AnyJobStateSummary]: """Return job state summary info for jobs, implicit groups jobs for collections or workflow invocations. **Warning**: We allow anyone to fetch job state information about any object they can guess an encoded ID for - it isn't considered protected data. This keeps polling IDs as part of state calculation for large histories and collections as efficient as possible. """ return self.service.index_jobs_summary(trans, params)
[docs] @router.get( "/api/histories/{history_id}/contents/dataset_collections/{id}/download", summary="Download the content of a dataset collection as a `zip` archive.", response_class=StreamingResponse, operation_id="history_contents__download_collection", ) def download_dataset_collection_history_content( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: Optional[DecodedDatabaseIdField] = Path( description="The encoded database identifier of the History.", ), id: DecodedDatabaseIdField = HistoryHDCAIDPathParam, ): """Download the content of a history dataset collection as a `zip` archive while maintaining approximate collection structure. """ archive = self.service.get_dataset_collection_archive_for_download(trans, id) return StreamingResponse(archive.response(), headers=archive.get_headers())
[docs] @router.get( "/api/dataset_collections/{id}/download", summary="Download the content of a dataset collection as a `zip` archive.", response_class=StreamingResponse, tags=["dataset collections"], operation_id="dataset_collections__download", ) def download_dataset_collection( self, trans: ProvidesHistoryContext = DependsOnTrans, id: DecodedDatabaseIdField = HistoryHDCAIDPathParam, ): """Download the content of a history dataset collection as a `zip` archive while maintaining approximate collection structure. """ archive = self.service.get_dataset_collection_archive_for_download(trans, id) return StreamingResponse(archive.response(), headers=archive.get_headers())
[docs] @router.post( "/api/histories/{history_id}/contents/dataset_collections/{id}/prepare_download", summary="Prepare an short term storage object that the collection will be downloaded to.", include_in_schema=False, ) @router.post( "/api/dataset_collections/{id}/prepare_download", summary="Prepare an short term storage object that the collection will be downloaded to.", responses={ 200: { "description": "Short term storage reference for async monitoring of this download.", }, 501: {"description": "Required asynchronous tasks required for this operation not available."}, }, tags=["dataset collections"], ) def prepare_collection_download( self, trans: ProvidesHistoryContext = DependsOnTrans, id: DecodedDatabaseIdField = HistoryHDCAIDPathParam, ) -> AsyncFile: """The history dataset collection will be written as a `zip` archive to the returned short term storage object. Progress tracking this file's creation can be tracked with the short_term_storage API. """ return self.service.prepare_collection_download(trans, id)
[docs] @router.post( "/api/histories/{history_id}/contents/{type}s", summary="Create a new `HDA` or `HDCA` in the given History.", operation_id="history_contents__create_typed", ) def create_typed( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, type: HistoryContentType = ContentTypePathParam, serialization_params: SerializationParams = Depends(query_serialization_params), payload: CreateHistoryContentPayload = Body(...), ) -> Union[AnyHistoryContentItem, List[AnyHistoryContentItem]]: """Create a new `HDA` or `HDCA` in the given History.""" return self._create(trans, history_id, type, serialization_params, payload)
[docs] @router.post( "/api/histories/{history_id}/contents", summary="Create a new `HDA` or `HDCA` in the given History.", deprecated=True, operation_id="history_contents__create", ) def create( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, type: Optional[HistoryContentType] = ContentTypeQueryParam(default=None), serialization_params: SerializationParams = Depends(query_serialization_params), payload: CreateHistoryContentPayload = Body(...), ) -> Union[AnyHistoryContentItem, List[AnyHistoryContentItem]]: """Create a new `HDA` or `HDCA` in the given History.""" return self._create(trans, history_id, type, serialization_params, payload)
def _create( self, trans: ProvidesHistoryContext, history_id: DecodedDatabaseIdField, type: Optional[HistoryContentType], serialization_params: SerializationParams, payload: CreateHistoryContentPayload, ) -> Union[AnyHistoryContentItem, List[AnyHistoryContentItem]]: """Create a new `HDA` or `HDCA` in the given History.""" payload.type = type or payload.type return self.service.create(trans, history_id, payload, serialization_params)
[docs] @router.put( "/api/histories/{history_id}/contents/{dataset_id}/permissions", summary="Set permissions of the given history dataset to the given role ids.", ) def update_permissions( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, dataset_id: DecodedDatabaseIdField = HistoryItemIDPathParam, # Using a generic Dict here as an attempt on supporting multiple aliases for the permissions params. payload: Dict[str, Any] = Body( default=..., example=UpdateDatasetPermissionsPayload(), ), ) -> DatasetAssociationRoles: """Set permissions of the given history dataset to the given role ids.""" update_payload = get_update_permission_payload(payload) return self.service.update_permissions(trans, dataset_id, update_payload)
[docs] @router.put( "/api/histories/{history_id}/contents", summary="Batch update specific properties of a set items contained in the given History.", ) def update_batch( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, serialization_params: SerializationParams = Depends(query_serialization_params), payload: UpdateHistoryContentsBatchPayload = Body(...), ) -> HistoryContentsResult: """Batch update specific properties of a set items contained in the given History. If you provide an invalid/unknown property key the request will not fail, but no changes will be made to the items. """ result = self.service.update_batch(trans, history_id, payload, serialization_params) return HistoryContentsResult.construct(__root__=result)
[docs] @router.put( "/api/histories/{history_id}/contents/bulk", summary="Executes an operation on a set of items contained in the given History.", ) def bulk_operation( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, filter_query_params: ValueFilterQueryParams = Depends(get_value_filter_query_params), payload: HistoryContentBulkOperationPayload = Body(...), ) -> HistoryContentBulkOperationResult: """Executes an operation on a set of items contained in the given History. The items to be processed can be explicitly set or determined by a dynamic query. """ return self.service.bulk_operation(trans, history_id, filter_query_params, payload)
[docs] @router.put( "/api/histories/{history_id}/contents/{id}/validate", summary="Validates the metadata associated with a dataset within a History.", ) def validate( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, id: DecodedDatabaseIdField = HistoryItemIDPathParam, ) -> dict: # TODO: define a response? """Validates the metadata associated with a dataset within a History.""" return self.service.validate(trans, history_id, id)
[docs] @router.put( "/api/histories/{history_id}/contents/{type}s/{id}", summary="Updates the values for the history content item with the given ``ID`` and path specified type.", operation_id="history_contents__update_typed", ) def update_typed( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, id: DecodedDatabaseIdField = HistoryItemIDPathParam, type: HistoryContentType = ContentTypePathParam, serialization_params: SerializationParams = Depends(query_serialization_params), payload: UpdateHistoryContentsPayload = Body(...), ) -> AnyHistoryContentItem: """Updates the values for the history content item with the given ``ID``.""" return self.service.update( trans, history_id, id, payload.dict(exclude_unset=True), serialization_params, contents_type=type )
[docs] @router.put( "/api/histories/{history_id}/contents/{id}", summary="Updates the values for the history content item with the given ``ID`` and query specified type. ``/api/histories/{history_id}/contents/{type}s/{id}`` should be used instead.", deprecated=True, operation_id="history_contents__update_legacy", ) def update_legacy( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, id: DecodedDatabaseIdField = HistoryItemIDPathParam, type: HistoryContentType = ContentTypeQueryParam(default=HistoryContentType.dataset), serialization_params: SerializationParams = Depends(query_serialization_params), payload: UpdateHistoryContentsPayload = Body(...), ) -> AnyHistoryContentItem: """Updates the values for the history content item with the given ``ID``.""" return self.service.update( trans, history_id, id, payload.dict(exclude_unset=True), serialization_params, contents_type=type )
[docs] @router.delete( "/api/histories/{history_id}/contents/{type}s/{id}", summary="Delete the history content with the given ``ID`` and path specified type.", responses=CONTENT_DELETE_RESPONSES, operation_id="history_contents__delete_typed", ) def delete_typed( self, response: Response, trans: ProvidesHistoryContext = DependsOnTrans, history_id: str = Path(..., description="History ID or any string."), id: DecodedDatabaseIdField = HistoryItemIDPathParam, type: HistoryContentType = ContentTypePathParam, serialization_params: SerializationParams = Depends(query_serialization_params), purge: Optional[bool] = PurgeQueryParam, recursive: Optional[bool] = RecursiveQueryParam, stop_job: Optional[bool] = StopJobQueryParam, payload: DeleteHistoryContentPayload = Body(None), ): """ Delete the history content with the given ``ID`` and path specified type. **Note**: Currently does not stop any active jobs for which this dataset is an output. """ return self._delete( response, trans, id, type, serialization_params, purge, recursive, stop_job, payload, )
[docs] @router.delete( "/api/histories/{history_id}/contents/{id}", summary="Delete the history dataset with the given ``ID``.", responses=CONTENT_DELETE_RESPONSES, operation_id="history_contents__delete_legacy", ) def delete_legacy( self, response: Response, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, id: DecodedDatabaseIdField = HistoryItemIDPathParam, type: HistoryContentType = ContentTypeQueryParam(default=HistoryContentType.dataset), serialization_params: SerializationParams = Depends(query_serialization_params), purge: Optional[bool] = PurgeQueryParam, recursive: Optional[bool] = RecursiveQueryParam, stop_job: Optional[bool] = StopJobQueryParam, payload: DeleteHistoryContentPayload = Body(None), ): """ Delete the history content with the given ``ID`` and query specified type (defaults to dataset). **Note**: Currently does not stop any active jobs for which this dataset is an output. """ return self._delete( response, trans, id, type, serialization_params, purge, recursive, stop_job, payload, )
def _delete( self, response: Response, trans: ProvidesHistoryContext, id: DecodedDatabaseIdField, type: HistoryContentType, serialization_params: SerializationParams, purge: Optional[bool], recursive: Optional[bool], stop_job: Optional[bool], payload: DeleteHistoryContentPayload, ): # TODO: should we just use the default payload and deprecate the query params? if payload is None: payload = DeleteHistoryContentPayload() payload.purge = payload.purge or purge is True payload.recursive = payload.recursive or recursive is True payload.stop_job = payload.stop_job or stop_job is True rval = self.service.delete( trans, id=id, serialization_params=serialization_params, contents_type=type, payload=payload, ) async_result = rval.pop("async_result", None) if async_result: response.status_code = status.HTTP_202_ACCEPTED return rval
[docs] @router.get( "/api/histories/{history_id}/contents/archive/{filename}.{format}", summary="Build and return a compressed archive of the selected history contents.", operation_id="history_contents__archive_named", ) def archive_named( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, filename: str = ArchiveFilenamePathParam, format: str = Path( description="Output format of the archive.", deprecated=True, # Looks like is not really used? ), dry_run: Optional[bool] = DryRunQueryParam, filter_query_params: FilterQueryParams = Depends(get_filter_query_params), ): """Build and return a compressed archive of the selected history contents. **Note**: this is a volatile endpoint and settings and behavior may change.""" archive = self.service.archive(trans, history_id, filter_query_params, filename, dry_run) if isinstance(archive, HistoryContentsArchiveDryRunResult): return archive return StreamingResponse(archive.response(), headers=archive.get_headers())
[docs] @router.get( "/api/histories/{history_id}/contents/archive", summary="Build and return a compressed archive of the selected history contents.", operation_id="history_contents__archive", ) def archive( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, filename: Optional[str] = ArchiveFilenameQueryParam, dry_run: Optional[bool] = DryRunQueryParam, filter_query_params: FilterQueryParams = Depends(get_filter_query_params), ): """Build and return a compressed archive of the selected history contents. **Note**: this is a volatile endpoint and settings and behavior may change.""" archive = self.service.archive(trans, history_id, filter_query_params, filename, dry_run) if isinstance(archive, HistoryContentsArchiveDryRunResult): return archive return StreamingResponse(archive.response(), headers=archive.get_headers())
[docs] @router.post("/api/histories/{history_id}/contents_from_store", summary="Create contents from store.") def create_from_store( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, serialization_params: SerializationParams = Depends(query_serialization_params), create_payload: CreateHistoryContentFromStore = Body(...), ) -> List[AnyHistoryContentItem]: """ Create history contents from model store. Input can be a tarfile created with build_objects script distributed with galaxy-data, from an exported history with files stripped out, or hand-crafted JSON dictionary. """ return self.service.create_from_store(trans, history_id, create_payload, serialization_params)
[docs] @router.post( "/api/histories/{history_id}/contents/datasets/{id}/materialize", summary="Materialize a deferred dataset into real, usable dataset.", ) def materialize_dataset( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, id: DecodedDatabaseIdField = HistoryItemIDPathParam, ) -> AsyncTaskResultSummary: materialize_request = MaterializeDatasetInstanceRequest.construct( history_id=history_id, source=DatasetSourceType.hda, content=id, ) rval = self.service.materialize(trans, materialize_request) return rval
[docs] @router.post( "/api/histories/{history_id}/materialize", summary="Materialize a deferred library or HDA dataset into real, usable dataset in specified history.", ) def materialize_to_history( self, trans: ProvidesHistoryContext = DependsOnTrans, history_id: DecodedDatabaseIdField = HistoryIDPathParam, materialize_api_payload: MaterializeDatasetInstanceAPIRequest = Body(...), ) -> AsyncTaskResultSummary: materialize_request: MaterializeDatasetInstanceRequest = MaterializeDatasetInstanceRequest.construct( history_id=history_id, **materialize_api_payload.dict() ) rval = self.service.materialize(trans, materialize_request) return rval