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.cloud
"""
API operations on Cloud-based storages, such as Amazon Simple Storage Service (S3).
"""
import logging
from fastapi import (
Body,
Response,
status,
)
from pydantic import Required
from galaxy.managers.cloud import CloudManager
from galaxy.managers.context import ProvidesHistoryContext
from galaxy.managers.datasets import DatasetSerializer
from galaxy.schema.cloud import (
CloudDatasets,
CloudDatasetsResponse,
CloudObjects,
DatasetSummaryList,
StatusCode,
)
from galaxy.webapps.galaxy.api import (
depends,
Router,
)
from . import DependsOnTrans
log = logging.getLogger(__name__)
router = Router(tags=["cloud"])
[docs]@router.cbv
class FastAPICloudController:
cloud_manager: CloudManager = depends(CloudManager)
datasets_serializer: DatasetSerializer = depends(DatasetSerializer)
[docs] @router.get(
"/api/cloud/storage",
summary="Lists cloud-based buckets (e.g., S3 bucket, Azure blob) user has defined. Is not yet implemented",
deprecated=True,
)
def index(
self,
response: Response,
):
# TODO: This can be implemented leveraging PluggedMedia objects (part of the user-based object store project)
response.status_code = status.HTTP_501_NOT_IMPLEMENTED
return StatusCode(detail="Not yet implemented.", status=501)
[docs] @router.post(
"/api/cloud/storage/get",
summary="Gets given objects from a given cloud-based bucket to a Galaxy history.",
deprecated=True,
)
def get(
self,
payload: CloudObjects = Body(default=Required),
trans: ProvidesHistoryContext = DependsOnTrans,
) -> DatasetSummaryList:
datasets = self.cloud_manager.get(
trans=trans,
history_id=payload.history_id,
bucket_name=payload.bucket,
objects=payload.objects,
authz_id=payload.authz_id,
input_args=payload.input_args,
)
rtv = []
for dataset in datasets:
rtv.append(self.datasets_serializer.serialize_to_view(dataset, view="summary"))
return DatasetSummaryList.construct(__root__=rtv)
[docs] @router.post(
"/api/cloud/storage/send",
summary="Sends given dataset(s) in a given history to a given cloud-based bucket.",
deprecated=True,
)
def send(
self,
payload: CloudDatasets = Body(default=Required),
trans: ProvidesHistoryContext = DependsOnTrans,
) -> CloudDatasetsResponse:
log.info(
msg="Received api/send request for `{}` datasets using authnz with id `{}`, and history `{}`."
"".format(
"all the dataset in the given history" if not payload.dataset_ids else len(payload.dataset_ids),
payload.authz_id,
payload.history_id,
)
)
sent, failed = self.cloud_manager.send(
trans=trans,
history_id=payload.history_id,
bucket_name=payload.bucket,
authz_id=payload.authz_id,
dataset_ids=payload.dataset_ids,
overwrite_existing=payload.overwrite_existing,
)
return CloudDatasetsResponse(sent_dataset_labels=sent, failed_dataset_labels=failed, bucket_name=payload.bucket)