Introduce links for async jobs actions (hateos style) in webserver

api/specs/web-server/_long_running_tasks.py

@@ -4,13 +4,17 @@
 # pylint: disable=too-many-arguments
 
 
-from typing import Annotated
+from typing import Annotated, Any
 
 from fastapi import APIRouter, Depends, status
 from models_library.generics import Envelope
+from models_library.rest_error import EnvelopedError
 from servicelib.aiohttp.long_running_tasks._routes import _PathParam
 from servicelib.long_running_tasks._models import TaskGet, TaskStatus
 from simcore_service_webserver._meta import API_VTAG
+from simcore_service_webserver.tasks._exception_handlers import (
+    _TO_HTTP_ERROR_MAP as data_export_http_error_map,
+)
 
 router = APIRouter(
     prefix=f"/{API_VTAG}",
@@ -19,37 +23,52 @@
     ],
 )
 
+_data_export_responses: dict[int | str, dict[str, Any]] = {
+    i.status_code: {"model": EnvelopedError}
+    for i in data_export_http_error_map.values()
+}
+
 
 @router.get(
     "/tasks",
     response_model=Envelope[list[TaskGet]],
+    name="list_tasks",
+    description="Lists all long running tasks",
+    responses=_data_export_responses,
 )
-def list_tasks():
-    ...
+def list_tasks(): ...
 
 
 @router.get(
     "/tasks/{task_id}",
     response_model=Envelope[TaskStatus],
+    name="get_task_status",
+    description="Retrieves the status of a task",
+    responses=_data_export_responses,
 )
 def get_task_status(
     _path_params: Annotated[_PathParam, Depends()],
-):
-    ...
+): ...
 
 
 @router.delete(
     "/tasks/{task_id}",
+    name="cancel_and_delete_task",
+    description="Cancels and deletes a task",
+    responses=_data_export_responses,
     status_code=status.HTTP_204_NO_CONTENT,
 )
 def cancel_and_delete_task(
     _path_params: Annotated[_PathParam, Depends()],
-):
-    ...
+): ...
 
 
-@router.get("/tasks/{task_id}/result")
+@router.get(
+    "/tasks/{task_id}/result",
+    name="get_task_result",
+    description="Retrieves the result of a task",
+    responses=_data_export_responses,
+)
 def get_task_result(
     _path_params: Annotated[_PathParam, Depends()],
-):
-    ...
+): ...

api/specs/web-server/_storage.py

@@ -4,10 +4,12 @@
 # pylint: disable=too-many-arguments
 
 
-from typing import Annotated, TypeAlias
-from uuid import UUID
+from typing import Annotated, Any, TypeAlias
 
 from fastapi import APIRouter, Depends, Query, status
+from models_library.api_schemas_long_running_tasks.tasks import (
+    TaskGet,
+)
 from models_library.api_schemas_storage.storage_schemas import (
     FileLocation,
     FileMetaDataGet,
@@ -22,19 +24,19 @@
 from models_library.api_schemas_webserver.storage import (
     DataExportPost,
     ListPathsQueryParams,
-    StorageAsyncJobGet,
-    StorageAsyncJobResult,
-    StorageAsyncJobStatus,
     StorageLocationPathParams,
     StoragePathComputeSizeParams,
 )
 from models_library.generics import Envelope
 from models_library.projects_nodes_io import LocationID
-from models_library.users import UserID
+from models_library.rest_error import EnvelopedError
 from pydantic import AnyUrl, ByteSize
 from servicelib.fastapi.rest_pagination import CustomizedPathsCursorPage
 from simcore_service_webserver._meta import API_VTAG
 from simcore_service_webserver.storage.schemas import DatasetMetaData, FileMetaData
+from simcore_service_webserver.tasks._exception_handlers import (
+    _TO_HTTP_ERROR_MAP as data_export_http_error_map,
+)
 
 router = APIRouter(
     prefix=f"/{API_VTAG}",
@@ -71,7 +73,7 @@ async def list_storage_paths(
 
 @router.post(
     "/storage/locations/{location_id}/paths/{path}:size",
-    response_model=Envelope[StorageAsyncJobGet],
+    response_model=Envelope[TaskGet],
     status_code=status.HTTP_202_ACCEPTED,
 )
 async def compute_path_size(_path: Annotated[StoragePathComputeSizeParams, Depends()]):
@@ -205,46 +207,18 @@ async def is_completed_upload_file(
 
 
 # data export
+_data_export_responses: dict[int | str, dict[str, Any]] = {
+    i.status_code: {"model": EnvelopedError}
+    for i in data_export_http_error_map.values()
+}
+
+
 @router.post(
     "/storage/locations/{location_id}/export-data",
-    response_model=Envelope[StorageAsyncJobGet],
+    response_model=Envelope[TaskGet],
     name="export_data",
     description="Export data",
+    responses=_data_export_responses,
 )
 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(job_id: UUID):
-    """Get async job status"""
-
-
-@router.post(
-    "/storage/async-jobs/{job_id}:abort",
-    name="abort_async_job",
-)
-async def abort_async_job(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(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/models-library/src/models_library/api_schemas_long_running_tasks/tasks.py

@@ -10,7 +10,7 @@
 class TaskStatus(BaseModel):
     task_progress: TaskProgress
     done: bool
-    started: datetime
+    started: datetime | None
 
 
 class TaskResult(BaseModel):

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

@@ -16,8 +16,7 @@ class AsyncJobStatus(BaseModel):
 
 
 class AsyncJobResult(BaseModel):
-    result: Any | None
-    error: Any | None
+    result: Any
 
 
 class AsyncJobGet(BaseModel):

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

@@ -5,9 +5,27 @@ class BaseAsyncjobRpcError(OsparcErrorMixin, RuntimeError):
     pass
 
 
-class StatusError(BaseAsyncjobRpcError):
+class JobSchedulerError(BaseAsyncjobRpcError):
+    msg_template: str = "Celery exception: {exc}"
+
+
+class JobMissingError(BaseAsyncjobRpcError):
+    msg_template: str = "Job {job_id} does not exist"
+
+
+class JobStatusError(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}"
+class JobNotDoneError(BaseAsyncjobRpcError):
+    msg_template: str = "Job {job_id} not done"
+
+
+class JobAbortedError(BaseAsyncjobRpcError):
+    msg_template: str = "Job {job_id} aborted"
+
+
+class JobError(BaseAsyncjobRpcError):
+    msg_template: str = (
+        "Job {job_id} failed with exception type {exc_type} and message {exc_msg}"
+    )

packages/models-library/src/models_library/api_schemas_storage/data_export_async_jobs.py

@@ -17,17 +17,11 @@ class StorageRpcBaseError(OsparcErrorMixin, RuntimeError):
     pass
 
 
-class InvalidLocationIdError(StorageRpcBaseError):
-    msg_template: str = "Invalid location_id {location_id}"
-
-
 class InvalidFileIdentifierError(StorageRpcBaseError):
     msg_template: str = "Could not find the file {file_id}"
 
 
 class AccessRightError(StorageRpcBaseError):
-    msg_template: str = "User {user_id} does not have access to file {file_id} with location {location_id}"
-
-
-class DataExportError(StorageRpcBaseError):
-    msg_template: str = "Could not complete data export job with id {job_id}"
+    msg_template: str = (
+        "User {user_id} does not have access to file {file_id} with location {location_id}"
+    )

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

@@ -1,25 +1,18 @@
 from pathlib import Path
-from typing import Annotated, Any
+from typing import Annotated
 
 from pydantic import BaseModel, Field
 
-from ..api_schemas_rpc_async_jobs.async_jobs import (
-    AsyncJobGet,
-    AsyncJobId,
-    AsyncJobResult,
-    AsyncJobStatus,
-)
 from ..api_schemas_storage.data_export_async_jobs import DataExportTaskStartInput
 from ..api_schemas_storage.storage_schemas import (
     DEFAULT_NUMBER_OF_PATHS_PER_PAGE,
     MAX_NUMBER_OF_PATHS_PER_PAGE,
 )
-from ..progress_bar import ProgressReport
 from ..projects_nodes_io import LocationID, StorageFileID
 from ..rest_pagination import (
     CursorQueryParameters,
 )
-from ._base import InputSchema, OutputSchema
+from ._base import InputSchema
 
 
 class StorageLocationPathParams(BaseModel):
@@ -51,40 +44,3 @@ def to_rpc_schema(self, location_id: LocationID) -> DataExportTaskStartInput:
             file_and_folder_ids=self.paths,
             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
-
-    @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,
-        )
-
-
-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/service-library/src/servicelib/rabbitmq/rpc_interfaces/async_jobs/async_jobs.py

@@ -1,7 +1,6 @@
 from typing import Final
 
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
-    AsyncJobAbort,
     AsyncJobGet,
     AsyncJobId,
     AsyncJobNameData,
@@ -24,16 +23,14 @@ async def abort(
     rpc_namespace: RPCNamespace,
     job_id: AsyncJobId,
     job_id_data: AsyncJobNameData,
-) -> AsyncJobAbort:
-    result = await rabbitmq_rpc_client.request(
+) -> None:
+    await rabbitmq_rpc_client.request(
         rpc_namespace,
         _RPC_METHOD_NAME_ADAPTER.validate_python("abort"),
         job_id=job_id,
         job_id_data=job_id_data,
         timeout_s=_DEFAULT_TIMEOUT_S,
     )
-    assert isinstance(result, AsyncJobAbort)
-    return result
 
 
 async def get_status(