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

@@ -10,6 +10,9 @@
 from models_library.generics import Envelope
 from models_library.projects_nodes_io import LocationID
 from models_library.storage_schemas import (
+    AsyncJobGet,
+    AsyncJobStatus,
+    DataExportPost,
     FileLocation,
     FileMetaDataGet,
     FileUploadCompleteFutureResponse,
@@ -167,3 +170,23 @@ 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/export-data",
+    response_model=Envelope[AsyncJobGet],
+    name="storage_export_data",
+    description="Export data",
+)
+async def export_data(data_export: DataExportPost):
+    """Trigger data export. Returns async job id for getting status and results"""
+
+
+@router.get(
+    "/storage/async-jobs/status",
+    response_model=Envelope[AsyncJobStatus],
+    name="storage_async_job_status",
+)
+async def get_async_job_status(task_id: AsyncJobGet):
+    """Get async job status"""

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

@@ -5,11 +5,11 @@
 from pydantic import BaseModel, Field, PositiveFloat, model_validator
 from typing_extensions import Self
 
-TaskRpcId: TypeAlias = UUID
+AsyncJobRpcId: TypeAlias = UUID
 
 
-class TaskRpcStatus(BaseModel):
-    task_id: TaskRpcId
+class AsyncJobRpcStatus(BaseModel):
+    job_id: AsyncJobRpcId
     task_progress: PositiveFloat = Field(..., ge=0.0, le=1.0)
     done: bool
     started: datetime
@@ -26,11 +26,16 @@ def _check_consistency(self) -> Self:
         return self
 
 
-class TaskRpcResult(BaseModel):
+class AsyncJobRpcResult(BaseModel):
     result: Any | None
     error: Any | None
 
 
-class TaskRpcGet(BaseModel):
-    task_id: TaskRpcId
+class AsyncJobRpcGet(BaseModel):
+    job_id: AsyncJobRpcId
     task_name: str
+
+
+class AsyncJobRpcAbort(BaseModel):
+    result: bool
+    job_id: AsyncJobRpcId

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 StorageRpcErrors(OsparcErrorMixin, RuntimeError):
+    pass
+
+
+class InvalidFileIdentifierError(StorageRpcErrors):
+    msg_template: str = "Could not find the file {file_id}"
+
+
+class AccessRightError(StorageRpcErrors):
+    msg_template: str = "User {user_id} does not have access to file {file_id}"
+
+
+class DataExportError(StorageRpcErrors):
+    msg_template: str = "Could not complete data export job with id {job_id}"

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

@@ -8,16 +8,28 @@
 
 from datetime import datetime
 from enum import Enum
+
+# /data-export
+from pathlib import Path
 from typing import Annotated, Any, Literal, Self, TypeAlias
 from uuid import UUID
 
+from models_library.api_schemas_rpc_async_jobs.async_jobs import (
+    AsyncJobRpcGet,
+    AsyncJobRpcResult,
+    AsyncJobRpcStatus,
+)
+from models_library.api_schemas_storage.data_export_async_jobs import (
+    DataExportTaskStartInput,
+)
 from models_library.projects import ProjectID
 from models_library.users import UserID
 from pydantic import (
     BaseModel,
     ByteSize,
     ConfigDict,
     Field,
+    PositiveFloat,
     PositiveInt,
     RootModel,
     StringConstraints,
@@ -368,3 +380,55 @@ def ensure_consistent_entries(self: Self) -> Self:
 
 class SoftCopyBody(BaseModel):
     link_id: SimcoreS3FileID
+
+
+class DataExportPost(BaseModel):
+    paths: list[Path]
+
+    def to_storage_model(
+        self, user_id: UserID, location_id: LocationID
+    ) -> DataExportTaskStartInput:
+        return DataExportTaskStartInput(
+            paths=self.paths, user_id=user_id, location_id=location_id
+        )
+
+
+class AsyncJobGet(BaseModel):
+    job_id: UUID
+
+    @classmethod
+    def from_async_job_rpc_get(cls, async_job_rpc_get: AsyncJobRpcGet) -> "AsyncJobGet":
+        return AsyncJobGet(job_id=async_job_rpc_get.job_id)
+
+
+class AsyncJobStatus(BaseModel):
+    job_id: UUID
+    task_progress: PositiveFloat = Field(..., ge=0.0, le=1.0)
+    done: bool
+    started: datetime
+    stopped: datetime | None
+
+    @classmethod
+    def from_async_job_rpc_status(
+        cls, async_job_rpc_status: AsyncJobRpcStatus
+    ) -> "AsyncJobStatus":
+        return AsyncJobStatus(
+            job_id=async_job_rpc_status.job_id,
+            task_progress=async_job_rpc_status.task_progress,
+            done=async_job_rpc_status.done,
+            started=async_job_rpc_status.started,
+            stopped=async_job_rpc_status.stopped,
+        )
+
+
+class AsyncJobResult(BaseModel):
+    result: Any | None
+    error: Any | None
+
+    @classmethod
+    def from_async_job_rpc_result(
+        cls, async_job_rpc_result: AsyncJobRpcResult
+    ) -> "AsyncJobResult":
+        return AsyncJobResult(
+            result=async_job_rpc_result.result, error=async_job_rpc_result.error
+        )

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/async_jobs/async_jobs.py

@@ -0,0 +1,64 @@
+from typing import Final
+
+from models_library.api_schemas_rpc_async_jobs.async_jobs import (
+    AsyncJobRpcAbort,
+    AsyncJobRpcId,
+    AsyncJobRpcResult,
+    AsyncJobRpcStatus,
+)
+from models_library.rabbitmq_basic_types import RPCMethodName, RPCNamespace
+from pydantic import NonNegativeInt, TypeAdapter
+
+from ... import RabbitMQRPCClient
+
+_DEFAULT_TIMEOUT_S: Final[NonNegativeInt] = 30
+
+_RPC_METHOD_NAME_ADAPTER = TypeAdapter(RPCMethodName)
+
+
+async def abort(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    rpc_namespace: RPCNamespace,
+    job_id: AsyncJobRpcId
+) -> AsyncJobRpcAbort:
+    result = await rabbitmq_rpc_client.request(
+        rpc_namespace,
+        _RPC_METHOD_NAME_ADAPTER.validate_python("abort"),
+        job_id=job_id,
+        timeout_s=_DEFAULT_TIMEOUT_S,
+    )
+    assert isinstance(result, AsyncJobRpcAbort)
+    return result
+
+
+async def get_status(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    rpc_namespace: RPCNamespace,
+    job_id: AsyncJobRpcId
+) -> AsyncJobRpcStatus:
+    result = await rabbitmq_rpc_client.request(
+        rpc_namespace,
+        _RPC_METHOD_NAME_ADAPTER.validate_python("get_status"),
+        job_id=job_id,
+        timeout_s=_DEFAULT_TIMEOUT_S,
+    )
+    assert isinstance(result, AsyncJobRpcStatus)
+    return result
+
+
+async def get_result(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    rpc_namespace: RPCNamespace,
+    job_id: AsyncJobRpcId
+) -> AsyncJobRpcResult:
+    result = await rabbitmq_rpc_client.request(
+        rpc_namespace,
+        _RPC_METHOD_NAME_ADAPTER.validate_python("get_result"),
+        job_id=job_id,
+        timeout_s=_DEFAULT_TIMEOUT_S,
+    )
+    assert isinstance(result, AsyncJobRpcResult)
+    return result

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/storage/data_export.py

@@ -1,14 +1,8 @@
 from typing import Final
 
-from models_library.api_schemas_rpc_data_export.tasks import (
-    TaskRpcGet,
-    TaskRpcId,
-    TaskRpcResult,
-    TaskRpcStatus,
-)
+from models_library.api_schemas_rpc_async_jobs.async_jobs import AsyncJobRpcGet
 from models_library.api_schemas_storage import STORAGE_RPC_NAMESPACE
-from models_library.api_schemas_storage.data_export_tasks import (
-    DataExportTaskAbortOutput,
+from models_library.api_schemas_storage.data_export_async_jobs import (
     DataExportTaskStartInput,
 )
 from models_library.rabbitmq_basic_types import RPCMethodName
@@ -23,51 +17,12 @@
 
 async def start_data_export(
     rabbitmq_rpc_client: RabbitMQRPCClient, *, paths: DataExportTaskStartInput
-) -> TaskRpcGet:
+) -> AsyncJobRpcGet:
     result = await rabbitmq_rpc_client.request(
         STORAGE_RPC_NAMESPACE,
         _RPC_METHOD_NAME_ADAPTER.validate_python("start_data_export"),
         paths=paths,
         timeout_s=_DEFAULT_TIMEOUT_S,
     )
-    assert isinstance(result, TaskRpcGet)
-    return result
-
-
-async def abort_data_export(
-    rabbitmq_rpc_client: RabbitMQRPCClient, *, task_id: TaskRpcId
-) -> DataExportTaskAbortOutput:
-    result = await rabbitmq_rpc_client.request(
-        STORAGE_RPC_NAMESPACE,
-        _RPC_METHOD_NAME_ADAPTER.validate_python("abort_data_export"),
-        task_id=task_id,
-        timeout_s=_DEFAULT_TIMEOUT_S,
-    )
-    assert isinstance(result, DataExportTaskAbortOutput)
-    return result
-
-
-async def get_data_export_status(
-    rabbitmq_rpc_client: RabbitMQRPCClient, *, task_id: TaskRpcId
-) -> TaskRpcStatus:
-    result = await rabbitmq_rpc_client.request(
-        STORAGE_RPC_NAMESPACE,
-        _RPC_METHOD_NAME_ADAPTER.validate_python("get_data_export_status"),
-        task_id=task_id,
-        timeout_s=_DEFAULT_TIMEOUT_S,
-    )
-    assert isinstance(result, TaskRpcStatus)
-    return result
-
-
-async def get_data_export_result(
-    rabbitmq_rpc_client: RabbitMQRPCClient, *, task_id: TaskRpcId
-) -> TaskRpcResult:
-    result = await rabbitmq_rpc_client.request(
-        STORAGE_RPC_NAMESPACE,
-        _RPC_METHOD_NAME_ADAPTER.validate_python("get_data_export_result"),
-        task_id=task_id,
-        timeout_s=_DEFAULT_TIMEOUT_S,
-    )
-    assert isinstance(result, TaskRpcResult)
+    assert isinstance(result, AsyncJobRpcGet)
     return result