Propagate data export endpoints to webserver (#7244)

Author: bisgaard-itis

api/specs/web-server/Makefile

@@ -16,3 +16,6 @@ install-dev install: _check_venv_active
 .PHONY: all
 all: _check_venv_active install
 	python openapi.py
+
+.PHONY: openapi-specs
+openapi-specs: all

api/specs/web-server/_storage.py

@@ -5,11 +5,10 @@
 
 
 from typing import TypeAlias
+from uuid import UUID
 
 from fastapi import APIRouter, Query, status
-from models_library.generics import Envelope
-from models_library.projects_nodes_io import LocationID
-from models_library.storage_schemas import (
+from models_library.api_schemas_storage.storage_schemas import (
     FileLocation,
     FileMetaDataGet,
     FileUploadCompleteFutureResponse,
@@ -19,6 +18,15 @@
     LinkType,
     PresignedLink,
 )
+from models_library.api_schemas_webserver.storage import (
+    DataExportPost,
+    StorageAsyncJobGet,
+    StorageAsyncJobResult,
+    StorageAsyncJobStatus,
+)
+from models_library.generics import Envelope
+from models_library.projects_nodes_io import LocationID
+from models_library.users import UserID
 from pydantic import AnyUrl, ByteSize
 from simcore_service_webserver._meta import API_VTAG
 from simcore_service_webserver.storage.schemas import DatasetMetaData, FileMetaData
@@ -167,3 +175,49 @@ async def is_completed_upload_file(
     location_id: LocationID, file_id: StorageFileIDStr, future_id: str
 ):
     """Returns state of upload completion"""
+
+
+# data export
+@router.post(
+    "/storage/locations/{location_id}/export-data",
+    response_model=Envelope[StorageAsyncJobGet],
+    name="export_data",
+    description="Export data",
+)
+async def export_data(data_export: DataExportPost, location_id: LocationID):
+    """Trigger data export. Returns async job id for getting status and results"""
+
+
+@router.get(
+    "/storage/async-jobs/{job_id}/status",
+    response_model=Envelope[StorageAsyncJobStatus],
+    name="get_async_job_status",
+)
+async def get_async_job_status(storage_async_job_get: StorageAsyncJobGet, job_id: UUID):
+    """Get async job status"""
+
+
+@router.post(
+    "/storage/async-jobs/{job_id}:abort",
+    name="abort_async_job",
+)
+async def abort_async_job(storage_async_job_get: StorageAsyncJobGet, job_id: UUID):
+    """aborts execution of an async job"""
+
+
+@router.get(
+    "/storage/async-jobs/{job_id}/result",
+    response_model=Envelope[StorageAsyncJobResult],
+    name="get_async_job_result",
+)
+async def get_async_job_result(storage_async_job_get: StorageAsyncJobGet, job_id: UUID):
+    """Get the result of the async job"""
+
+
+@router.get(
+    "/storage/async-jobs",
+    response_model=Envelope[list[StorageAsyncJobGet]],
+    name="get_async_jobs",
+)
+async def get_async_jobs(user_id: UserID):
+    """Retrunsa list of async jobs for the user"""

packages/aws-library/src/aws_library/s3/_client.py

@@ -13,9 +13,13 @@
 from boto3.s3.transfer import TransferConfig
 from botocore import exceptions as botocore_exc
 from botocore.client import Config
+from models_library.api_schemas_storage.storage_schemas import (
+    ETag,
+    S3BucketName,
+    UploadedPart,
+)
 from models_library.basic_types import SHA256Str
 from models_library.bytes_iters import BytesIter, DataSize
-from models_library.storage_schemas import ETag, S3BucketName, UploadedPart
 from pydantic import AnyUrl, ByteSize, TypeAdapter
 from servicelib.bytes_iters import DEFAULT_READ_CHUNK_SIZE, BytesStreamer
 from servicelib.logging_utils import log_catch, log_context

packages/aws-library/src/aws_library/s3/_models.py

@@ -2,8 +2,8 @@
 from dataclasses import dataclass
 from typing import TypeAlias
 
+from models_library.api_schemas_storage.storage_schemas import ETag
 from models_library.basic_types import SHA256Str
-from models_library.storage_schemas import ETag
 from pydantic import AnyUrl, BaseModel, ByteSize
 from types_aiobotocore_s3.type_defs import HeadObjectOutputTypeDef, ObjectTypeDef
 

packages/aws-library/tests/test_s3_client.py

@@ -38,8 +38,11 @@
 )
 from aws_library.s3._models import MultiPartUploadLinks
 from faker import Faker
+from models_library.api_schemas_storage.storage_schemas import (
+    S3BucketName,
+    UploadedPart,
+)
 from models_library.basic_types import SHA256Str
-from models_library.storage_schemas import S3BucketName, UploadedPart
 from moto.server import ThreadedMotoServer
 from pydantic import AnyUrl, ByteSize, TypeAdapter
 from pytest_benchmark.plugin import BenchmarkFixture

packages/models-library/src/models_library/api_schemas_rpc_data_export/__init__.py renamed to packages/models-library/src/models_library/api_schemas_rpc_async_jobs/__init__.py

@@ -0,0 +1,51 @@
+from datetime import datetime
+from typing import Any, TypeAlias
+from uuid import UUID
+
+from models_library.users import UserID
+from pydantic import BaseModel, model_validator
+from typing_extensions import Self
+
+from ..progress_bar import ProgressReport
+
+AsyncJobId: TypeAlias = UUID
+
+
+class AsyncJobStatus(BaseModel):
+    job_id: AsyncJobId
+    progress: ProgressReport
+    done: bool
+    started: datetime
+    stopped: datetime | None
+
+    @model_validator(mode="after")
+    def _check_consistency(self) -> Self:
+        is_done = self.done
+        is_stopped = self.stopped is not None
+
+        if is_done != is_stopped:
+            msg = f"Inconsistent data: {self.done=}, {self.stopped=}"
+            raise ValueError(msg)
+        return self
+
+
+class AsyncJobResult(BaseModel):
+    result: Any | None
+    error: Any | None
+
+
+class AsyncJobGet(BaseModel):
+    job_id: AsyncJobId
+    job_name: str
+
+
+class AsyncJobAbort(BaseModel):
+    result: bool
+    job_id: AsyncJobId
+
+
+class AsyncJobAccessData(BaseModel):
+    """Data for controlling access to an async job"""
+
+    user_id: UserID | None
+    product_name: str

packages/models-library/src/models_library/api_schemas_rpc_async_jobs/async_jobs.py

@@ -0,0 +1,13 @@
+from common_library.errors_classes import OsparcErrorMixin
+
+
+class BaseAsyncjobRpcError(OsparcErrorMixin, RuntimeError):
+    pass
+
+
+class StatusError(BaseAsyncjobRpcError):
+    msg_template: str = "Could not get status of job {job_id}"
+
+
+class ResultError(BaseAsyncjobRpcError):
+    msg_template: str = "Could not get results of job {job_id}"

packages/models-library/src/models_library/api_schemas_rpc_async_jobs/exceptions.py

@@ -0,0 +1,33 @@
+# pylint: disable=R6301
+from pathlib import Path
+
+from common_library.errors_classes import OsparcErrorMixin
+from models_library.projects_nodes_io import LocationID
+from models_library.users import UserID
+from pydantic import BaseModel, Field
+
+
+class DataExportTaskStartInput(BaseModel):
+    user_id: UserID
+    product_name: str
+    location_id: LocationID
+    paths: list[Path] = Field(..., min_length=1)
+
+
+### Exceptions
+
+
+class StorageRpcError(OsparcErrorMixin, RuntimeError):
+    pass
+
+
+class InvalidFileIdentifierError(StorageRpcError):
+    msg_template: str = "Could not find the file {file_id}"
+
+
+class AccessRightError(StorageRpcError):
+    msg_template: str = "User {user_id} does not have access to file {file_id}"
+
+
+class DataExportError(StorageRpcError):
+    msg_template: str = "Could not complete data export job with id {job_id}"