fix tasks endpoints

Author: giancarloromeo

services/web/server/src/simcore_service_webserver/storage/_rest.py

@@ -57,7 +57,7 @@
 from ..models import AuthenticatedRequestContext, WebServerOwnerMetadata
 from ..rabbitmq import get_rabbitmq_rpc_client
 from ..security.decorators import permission_required
-from ..tasks._controller._rest_exceptions import handle_export_data_exceptions
+from ..tasks._controller._rest_exceptions import handle_rest_requests_exceptions
 from .schemas import StorageFileIDStr
 from .settings import StorageSettings, get_plugin_settings
 
@@ -486,7 +486,7 @@ class _PathParams(BaseModel):
 )
 @login_required
 @permission_required("storage.files.*")
-@handle_export_data_exceptions
+@handle_rest_requests_exceptions
 async def export_data(request: web.Request) -> web.Response:
     class _PathParams(BaseModel):
         location_id: LocationID

services/web/server/src/simcore_service_webserver/tasks/_controller/_rest.py

@@ -1,8 +1,5 @@
-"""Handlers exposed by subsystem"""
-
 import logging
 from typing import Final
-from uuid import UUID
 
 from aiohttp import web
 from models_library.api_schemas_long_running_tasks.base import TaskProgress
@@ -11,11 +8,6 @@
     TaskResult,
     TaskStatus,
 )
-from models_library.api_schemas_rpc_async_jobs.async_jobs import (
-    AsyncJobId,
-)
-from models_library.api_schemas_storage import STORAGE_RPC_NAMESPACE
-from pydantic import BaseModel
 from servicelib.aiohttp import status
 from servicelib.aiohttp.long_running_tasks.server import (
     get_long_running_manager,
@@ -26,15 +18,15 @@
 from servicelib.aiohttp.rest_responses import create_data_response
 from servicelib.celery.models import OwnerMetadata
 from servicelib.long_running_tasks import lrt_api
-from servicelib.rabbitmq.rpc_interfaces.async_jobs import async_jobs
+from simcore_service_webserver.tasks._controller._rest_schemas import TaskPathParams
 
 from ..._meta import API_VTAG
+from ...celery import get_task_manager
 from ...login.decorators import login_required
 from ...long_running_tasks.plugin import webserver_request_context_decorator
 from ...models import AuthenticatedRequestContext, WebServerOwnerMetadata
-from ...rabbitmq import get_rabbitmq_rpc_client
-from ...security.decorators import permission_required
-from ._rest_exceptions import handle_export_data_exceptions
+from .. import _service
+from ._rest_exceptions import handle_rest_requests_exceptions
 
 log = logging.getLogger(__name__)
 
@@ -46,13 +38,12 @@
 
 @routes.get(
     _task_prefix,
-    name="get_async_jobs",
+    name="get_tasks",
 )
 @login_required
-@permission_required("storage.files.*")
-@handle_export_data_exceptions
+@handle_rest_requests_exceptions
 @webserver_request_context_decorator
-async def get_async_jobs(request: web.Request) -> web.Response:
+async def get_tasks(request: web.Request) -> web.Response:
     inprocess_long_running_manager = get_long_running_manager(request.app)
     inprocess_tracked_tasks = await lrt_api.list_tasks(
         inprocess_long_running_manager.rpc_client,
@@ -62,28 +53,26 @@ async def get_async_jobs(request: web.Request) -> web.Response:
 
     _req_ctx = AuthenticatedRequestContext.model_validate(request)
 
-    rabbitmq_rpc_client = get_rabbitmq_rpc_client(request.app)
-
-    user_async_jobs = await async_jobs.list_jobs(
-        rabbitmq_rpc_client=rabbitmq_rpc_client,
-        rpc_namespace=STORAGE_RPC_NAMESPACE,
+    tasks = await _service.list_tasks(
+        task_manager=get_task_manager(request.app),
         owner_metadata=OwnerMetadata.model_validate(
             WebServerOwnerMetadata(
                 user_id=_req_ctx.user_id,
                 product_name=_req_ctx.product_name,
             ).model_dump()
         ),
     )
+
     return create_data_response(
         [
             TaskGet(
-                task_id=f"{job.job_id}",
-                task_name=job.job_name,
-                status_href=f"{request.url.with_path(str(request.app.router['get_async_job_status'].url_for(task_id=str(job.job_id))))}",
-                abort_href=f"{request.url.with_path(str(request.app.router['cancel_async_job'].url_for(task_id=str(job.job_id))))}",
-                result_href=f"{request.url.with_path(str(request.app.router['get_async_job_result'].url_for(task_id=str(job.job_id))))}",
+                task_id=f"{task.job_id}",
+                task_name=task.job_name,
+                status_href=f"{request.url.with_path(str(request.app.router['get_async_job_status'].url_for(task_id=str(task.job_id))))}",
+                abort_href=f"{request.url.with_path(str(request.app.router['cancel_async_job'].url_for(task_id=str(task.job_id))))}",
+                result_href=f"{request.url.with_path(str(request.app.router['get_async_job_result'].url_for(task_id=str(task.job_id))))}",
             )
-            for job in user_async_jobs
+            for task in tasks
         ]
         + [
             TaskGet(
@@ -98,40 +87,35 @@ async def get_async_jobs(request: web.Request) -> web.Response:
     )
 
 
-class _StorageAsyncJobId(BaseModel):
-    task_id: AsyncJobId
-
-
 @routes.get(
     _task_prefix + "/{task_id}",
-    name="get_async_job_status",
+    name="get_task_status",
 )
 @login_required
-@handle_export_data_exceptions
-async def get_async_job_status(request: web.Request) -> web.Response:
+@handle_rest_requests_exceptions
+async def get_task_status(request: web.Request) -> web.Response:
 
     _req_ctx = AuthenticatedRequestContext.model_validate(request)
-    rabbitmq_rpc_client = get_rabbitmq_rpc_client(request.app)
+    _path_params = parse_request_path_parameters_as(TaskPathParams, request)
 
-    async_job_get = parse_request_path_parameters_as(_StorageAsyncJobId, request)
-    async_job_rpc_status = await async_jobs.status(
-        rabbitmq_rpc_client=rabbitmq_rpc_client,
-        rpc_namespace=STORAGE_RPC_NAMESPACE,
-        job_id=async_job_get.task_id,
+    task_status = await _service.get_task_status(
+        task_manager=get_task_manager(request.app),
         owner_metadata=OwnerMetadata.model_validate(
             WebServerOwnerMetadata(
                 user_id=_req_ctx.user_id,
                 product_name=_req_ctx.product_name,
             ).model_dump()
         ),
+        task_uuid=_path_params.task_id,
     )
-    _task_id = f"{async_job_rpc_status.job_id}"
+
+    _task_id = f"{task_status.job_id}"
     return create_data_response(
         TaskStatus(
             task_progress=TaskProgress(
-                task_id=_task_id, percent=async_job_rpc_status.progress.percent_value
+                task_id=_task_id, percent=task_status.progress.percent_value
             ),
-            done=async_job_rpc_status.done,
+            done=task_status.done,
             started=None,
         ),
         status=status.HTTP_200_OK,
@@ -140,61 +124,52 @@ async def get_async_job_status(request: web.Request) -> web.Response:
 
 @routes.delete(
     _task_prefix + "/{task_id}",
-    name="cancel_async_job",
+    name="cancel_task",
 )
 @login_required
-@permission_required("storage.files.*")
-@handle_export_data_exceptions
-async def cancel_async_job(request: web.Request) -> web.Response:
+@handle_rest_requests_exceptions
+async def cancel_task(request: web.Request) -> web.Response:
 
     _req_ctx = AuthenticatedRequestContext.model_validate(request)
+    _path_params = parse_request_path_parameters_as(TaskPathParams, request)
 
-    rabbitmq_rpc_client = get_rabbitmq_rpc_client(request.app)
-    async_job_get = parse_request_path_parameters_as(_StorageAsyncJobId, request)
-
-    await async_jobs.cancel(
-        rabbitmq_rpc_client=rabbitmq_rpc_client,
-        rpc_namespace=STORAGE_RPC_NAMESPACE,
-        job_id=async_job_get.task_id,
+    await _service.cancel_task(
+        task_manager=get_task_manager(request.app),
         owner_metadata=OwnerMetadata.model_validate(
             WebServerOwnerMetadata(
                 user_id=_req_ctx.user_id,
                 product_name=_req_ctx.product_name,
             ).model_dump()
         ),
+        task_uuid=_path_params.task_id,
     )
 
     return web.Response(status=status.HTTP_204_NO_CONTENT)
 
 
 @routes.get(
     _task_prefix + "/{task_id}/result",
-    name="get_async_job_result",
+    name="get_task_result",
 )
 @login_required
-@permission_required("storage.files.*")
-@handle_export_data_exceptions
-async def get_async_job_result(request: web.Request) -> web.Response:
-    class _PathParams(BaseModel):
-        task_id: UUID
+@handle_rest_requests_exceptions
+async def get_task_result(request: web.Request) -> web.Response:
 
     _req_ctx = AuthenticatedRequestContext.model_validate(request)
+    _path_params = parse_request_path_parameters_as(TaskPathParams, request)
 
-    rabbitmq_rpc_client = get_rabbitmq_rpc_client(request.app)
-    async_job_get = parse_request_path_parameters_as(_PathParams, request)
-    async_job_rpc_result = await async_jobs.result(
-        rabbitmq_rpc_client=rabbitmq_rpc_client,
-        rpc_namespace=STORAGE_RPC_NAMESPACE,
-        job_id=async_job_get.task_id,
+    task_result = await _service.get_task_result(
+        task_manager=get_task_manager(request.app),
         owner_metadata=OwnerMetadata.model_validate(
             WebServerOwnerMetadata(
                 user_id=_req_ctx.user_id,
                 product_name=_req_ctx.product_name,
             ).model_dump()
         ),
+        task_uuid=_path_params.task_id,
     )
 
     return create_data_response(
-        TaskResult(result=async_job_rpc_result.result, error=None),
+        TaskResult(result=task_result.result, error=None),
         status=status.HTTP_200_OK,
     )

services/web/server/src/simcore_service_webserver/tasks/_controller/_rest_exceptions.py

@@ -69,6 +69,6 @@
 }
 
 
-handle_export_data_exceptions = exception_handling_decorator(
+handle_rest_requests_exceptions = exception_handling_decorator(
     to_exceptions_handlers_map(_TO_HTTP_ERROR_MAP)
 )

services/web/server/src/simcore_service_webserver/tasks/_controller/_rest_schemas.py

@@ -0,0 +1,9 @@
+from uuid import UUID
+
+from pydantic import BaseModel
+from simcore_service_webserver.models import ConfigDict
+
+
+class TaskPathParams(BaseModel):
+    task_id: UUID
+    model_config = ConfigDict(extra="forbid", frozen=True)

services/web/server/src/simcore_service_webserver/tasks/_service.py

@@ -0,0 +1,120 @@
+import logging
+
+from celery_library.errors import (
+    TaskManagerError,
+    TaskNotFoundError,
+    TransferrableCeleryError,
+    decode_celery_transferrable_error,
+)
+from models_library.api_schemas_rpc_async_jobs.async_jobs import (
+    AsyncJobGet,
+    AsyncJobResult,
+    AsyncJobStatus,
+)
+from models_library.api_schemas_rpc_async_jobs.exceptions import (
+    JobError,
+    JobMissingError,
+    JobNotDoneError,
+    JobSchedulerError,
+)
+from servicelib.celery.models import OwnerMetadata, TaskState, TaskUUID
+from servicelib.celery.task_manager import TaskManager
+from servicelib.logging_utils import log_catch
+
+_logger = logging.getLogger(__name__)
+
+
+async def cancel_task(
+    task_manager: TaskManager,
+    owner_metadata: OwnerMetadata,
+    task_uuid: TaskUUID,
+):
+    try:
+        await task_manager.cancel_task(
+            owner_metadata=owner_metadata,
+            task_uuid=task_uuid,
+        )
+    except TaskNotFoundError as exc:
+        raise JobMissingError(job_id=task_uuid) from exc
+    except TaskManagerError as exc:
+        raise JobSchedulerError(exc=f"{exc}") from exc
+
+
+async def get_task_result(
+    task_manager: TaskManager,
+    owner_metadata: OwnerMetadata,
+    task_uuid: TaskUUID,
+) -> AsyncJobResult:
+    try:
+        _status = await task_manager.get_task_status(
+            owner_metadata=owner_metadata,
+            task_uuid=task_uuid,
+        )
+        if not _status.is_done:
+            raise JobNotDoneError(job_id=task_uuid)
+        _result = await task_manager.get_task_result(
+            owner_metadata=owner_metadata,
+            task_uuid=task_uuid,
+        )
+    except TaskNotFoundError as exc:
+        raise JobMissingError(job_id=task_uuid) from exc
+    except InternalError as exc:
+        raise JobSchedulerError(exc=f"{exc}") from exc
+
+    if _status.task_state == TaskState.FAILURE:
+        # fallback exception to report
+        exc_type = type(_result).__name__
+        exc_msg = f"{_result}"
+
+        # try to recover the original error
+        exception = None
+        with log_catch(_logger, reraise=False):
+            assert isinstance(_result, TransferrableCeleryError)  # nosec
+            exception = decode_celery_transferrable_error(_result)
+            exc_type = type(exception).__name__
+            exc_msg = f"{exception}"
+
+        if exception is None:
+            _logger.warning("Was not expecting '%s': '%s'", exc_type, exc_msg)
+
+        raise JobError(job_id=task_uuid, exc_type=exc_type, exc_msg=exc_msg)
+
+    return AsyncJobResult(result=_result)
+
+
+async def get_task_status(
+    task_manager: TaskManager,
+    owner_metadata: OwnerMetadata,
+    task_uuid: TaskUUID,
+) -> AsyncJobStatus:
+    try:
+        task_status = await task_manager.get_task_status(
+            owner_metadata=owner_metadata,
+            task_uuid=task_uuid,
+        )
+    except TaskNotFoundError as exc:
+        raise JobMissingError(job_id=task_uuid) from exc
+    except InternalError as exc:
+        raise JobSchedulerError(exc=f"{exc}") from exc
+
+    return AsyncJobStatus(
+        job_id=task_uuid,
+        progress=task_status.progress_report,
+        done=task_status.is_done,
+    )
+
+
+async def list_tasks(
+    task_manager: TaskManager,
+    owner_metadata: OwnerMetadata,
+) -> list[AsyncJobGet]:
+    try:
+        tasks = await task_manager.list_tasks(
+            owner_metadata=owner_metadata,
+        )
+    except InternalError as exc:
+        raise JobSchedulerError(exc=f"{exc}") from exc
+
+    return [
+        AsyncJobGet(job_id=task.uuid, job_name=task.metadata.name) for task in tasks
+    ]