Propagate data export endpoints to webserver

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/tasks.py → packages/models-library/src/models_library/api_schemas_rpc_async_jobs/async_jobs.py

@@ -2,15 +2,18 @@
 from typing import Any, TypeAlias
 from uuid import UUID
 
-from pydantic import BaseModel, Field, PositiveFloat, model_validator
+from models_library.users import UserID
+from pydantic import BaseModel, model_validator
 from typing_extensions import Self
 
-TaskRpcId: TypeAlias = UUID
+from ..progress_bar import ProgressReport
 
+AsyncJobId: TypeAlias = UUID
 
-class TaskRpcStatus(BaseModel):
-    task_id: TaskRpcId
-    task_progress: PositiveFloat = Field(..., ge=0.0, le=1.0)
+
+class AsyncJobStatus(BaseModel):
+    job_id: AsyncJobId
+    progress: ProgressReport
     done: bool
     started: datetime
     stopped: datetime | None
@@ -26,11 +29,22 @@ def _check_consistency(self) -> Self:
         return self
 
 
-class TaskRpcResult(BaseModel):
+class AsyncJobResult(BaseModel):
     result: Any | None
     error: Any | None
 
 
-class TaskRpcGet(BaseModel):
-    task_id: TaskRpcId
-    task_name: str
+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

packages/models-library/src/models_library/api_schemas_rpc_async_jobs/exceptions.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_storage/data_export_async_jobs.py

@@ -0,0 +1,32 @@
+# 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
+    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}"

packages/models-library/src/models_library/storage_schemas.py → packages/models-library/src/models_library/api_schemas_storage/storage_schemas.py

@@ -8,6 +8,8 @@
 
 from datetime import datetime
 from enum import Enum
+
+# /data-export
 from typing import Annotated, Any, Literal, Self, TypeAlias
 from uuid import UUID
 
@@ -26,10 +28,10 @@
 )
 from pydantic.networks import AnyUrl
 
-from .basic_regex import DATCORE_DATASET_NAME_RE, S3_BUCKET_NAME_RE
-from .basic_types import SHA256Str
-from .generics import ListModel
-from .projects_nodes_io import (
+from ..basic_regex import DATCORE_DATASET_NAME_RE, S3_BUCKET_NAME_RE
+from ..basic_types import SHA256Str
+from ..generics import ListModel
+from ..projects_nodes_io import (
     LocationID,
     LocationName,
     NodeID,

packages/models-library/src/models_library/api_schemas_webserver/storage.py

@@ -0,0 +1,67 @@
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+from ..api_schemas_rpc_async_jobs.async_jobs import (
+    AsyncJobGet,
+    AsyncJobId,
+    AsyncJobResult,
+    AsyncJobStatus,
+)
+from ..api_schemas_storage.data_export_async_jobs import DataExportTaskStartInput
+from ..progress_bar import ProgressReport
+from ..projects_nodes_io import LocationID
+from ..users import UserID
+from ._base import InputSchema, OutputSchema
+
+
+class DataExportPost(InputSchema):
+    paths: list[Path]
+
+    def to_rpc_schema(
+        self, user_id: UserID, location_id: LocationID
+    ) -> DataExportTaskStartInput:
+        return DataExportTaskStartInput(
+            paths=self.paths, user_id=user_id, location_id=location_id
+        )
+
+
+class StorageAsyncJobGet(OutputSchema):
+    job_id: AsyncJobId
+
+    @classmethod
+    def from_rpc_schema(cls, async_job_rpc_get: AsyncJobGet) -> "StorageAsyncJobGet":
+        return StorageAsyncJobGet(job_id=async_job_rpc_get.job_id)
+
+
+class StorageAsyncJobStatus(OutputSchema):
+    job_id: AsyncJobId
+    progress: ProgressReport
+    done: bool
+    started: datetime
+    stopped: datetime | None
+
+    @classmethod
+    def from_rpc_schema(
+        cls, async_job_rpc_status: AsyncJobStatus
+    ) -> "StorageAsyncJobStatus":
+        return StorageAsyncJobStatus(
+            job_id=async_job_rpc_status.job_id,
+            progress=async_job_rpc_status.progress,
+            done=async_job_rpc_status.done,
+            started=async_job_rpc_status.started,
+            stopped=async_job_rpc_status.stopped,
+        )
+
+
+class StorageAsyncJobResult(OutputSchema):
+    result: Any | None
+    error: Any | None
+
+    @classmethod
+    def from_rpc_schema(
+        cls, async_job_rpc_result: AsyncJobResult
+    ) -> "StorageAsyncJobResult":
+        return StorageAsyncJobResult(
+            result=async_job_rpc_result.result, error=async_job_rpc_result.error
+        )

packages/pytest-simcore/src/pytest_simcore/helpers/s3.py

@@ -8,7 +8,11 @@
 import orjson
 from aws_library.s3 import MultiPartUploadLinks
 from fastapi import status
-from models_library.storage_schemas import ETag, FileUploadSchema, UploadedPart
+from models_library.api_schemas_storage.storage_schemas import (
+    ETag,
+    FileUploadSchema,
+    UploadedPart,
+)
 from pydantic import AnyUrl, ByteSize, TypeAdapter
 from servicelib.utils import limited_as_completed, logged_gather
 from types_aiobotocore_s3 import S3Client

packages/pytest-simcore/src/pytest_simcore/services_api_mocks_for_aiohttp_clients.py

@@ -12,10 +12,7 @@
 from aioresponses.core import CallbackResult
 from faker import Faker
 from models_library.api_schemas_directorv2.comp_tasks import ComputationGet
-from models_library.generics import Envelope
-from models_library.projects_pipeline import ComputationTask
-from models_library.projects_state import RunningState
-from models_library.storage_schemas import (
+from models_library.api_schemas_storage.storage_schemas import (
     FileMetaDataGet,
     FileUploadCompleteFutureResponse,
     FileUploadCompleteResponse,
@@ -25,6 +22,9 @@
     LinkType,
     PresignedLink,
 )
+from models_library.generics import Envelope
+from models_library.projects_pipeline import ComputationTask
+from models_library.projects_state import RunningState
 from models_library.utils.fastapi_encoders import jsonable_encoder
 from pydantic import AnyUrl, ByteSize, TypeAdapter
 from servicelib.aiohttp import status