further improvements

Author: bisgaard-itis

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

@@ -6,7 +6,7 @@ class BaseAsyncjobRpcError(OsparcErrorMixin, RuntimeError):
 
 
 class JobSchedulerError(BaseAsyncjobRpcError):
-    msg_template: str = "Celery exception: {exc}"
+    msg_template: str = "Async job scheduler exception: {exc}"
 
 
 class JobMissingError(BaseAsyncjobRpcError):

services/api-server/src/simcore_service_api_server/api/dependencies/tasks.py

@@ -0,0 +1,13 @@
+from typing import Annotated
+
+from fastapi import Depends
+from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
+
+from ...services_rpc.async_jobs import AsyncJobClient
+from .rabbitmq import get_rabbitmq_rpc_client
+
+
+def get_async_jobs_client(
+    rabbitmq_rpc_client: Annotated[RabbitMQRPCClient, Depends(get_rabbitmq_rpc_client)],
+) -> AsyncJobClient:
+    return AsyncJobClient(_rabbitmq_rpc_client=rabbitmq_rpc_client)

services/api-server/src/simcore_service_api_server/api/routes/tasks.py

@@ -12,17 +12,15 @@
     AsyncJobId,
     AsyncJobNameData,
 )
-from models_library.api_schemas_storage import STORAGE_RPC_NAMESPACE
 from models_library.products import ProductName
 from models_library.users import UserID
 from servicelib.fastapi.dependencies import get_app
-from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
-from servicelib.rabbitmq.rpc_interfaces.async_jobs import async_jobs
 from simcore_service_api_server.models.schemas.tasks import ApiServerEnvelope
 
+from ...services_rpc.async_jobs import AsyncJobClient
 from ..dependencies.authentication import get_current_user_id
-from ..dependencies.rabbitmq import get_rabbitmq_rpc_client
 from ..dependencies.services import get_product_name
+from ..dependencies.tasks import get_async_jobs_client
 
 router = APIRouter()
 _logger = logging.getLogger(__name__)
@@ -37,11 +35,9 @@ async def get_async_jobs(
     app: Annotated[FastAPI, Depends(get_app)],
     user_id: Annotated[UserID, Depends(get_current_user_id)],
     product_name: Annotated[ProductName, Depends(get_product_name)],
-    rabbitmq_rpc_client: Annotated[RabbitMQRPCClient, Depends(get_rabbitmq_rpc_client)],
+    async_jobs: Annotated[AsyncJobClient, Depends(get_async_jobs_client)],
 ):
     user_async_jobs = await async_jobs.list_jobs(
-        rabbitmq_rpc_client=rabbitmq_rpc_client,
-        rpc_namespace=STORAGE_RPC_NAMESPACE,
         job_id_data=_get_job_id_data(user_id, product_name),
         filter_="",
     )
@@ -70,11 +66,9 @@ async def get_async_job_status(
     task_id: AsyncJobId,
     user_id: Annotated[UserID, Depends(get_current_user_id)],
     product_name: Annotated[ProductName, Depends(get_product_name)],
-    rabbitmq_rpc_client: Annotated[RabbitMQRPCClient, Depends(get_rabbitmq_rpc_client)],
+    async_jobs: Annotated[AsyncJobClient, Depends(get_async_jobs_client)],
 ):
     async_job_rpc_status = await async_jobs.status(
-        rabbitmq_rpc_client=rabbitmq_rpc_client,
-        rpc_namespace=STORAGE_RPC_NAMESPACE,
         job_id=task_id,
         job_id_data=_get_job_id_data(user_id, product_name),
     )
@@ -95,11 +89,9 @@ async def cancel_async_job(
     task_id: AsyncJobId,
     user_id: Annotated[UserID, Depends(get_current_user_id)],
     product_name: Annotated[ProductName, Depends(get_product_name)],
-    rabbitmq_rpc_client: Annotated[RabbitMQRPCClient, Depends(get_rabbitmq_rpc_client)],
+    async_jobs: Annotated[AsyncJobClient, Depends(get_async_jobs_client)],
 ):
     await async_jobs.cancel(
-        rabbitmq_rpc_client=rabbitmq_rpc_client,
-        rpc_namespace=STORAGE_RPC_NAMESPACE,
         job_id=task_id,
         job_id_data=_get_job_id_data(user_id, product_name),
     )
@@ -110,11 +102,9 @@ async def get_async_job_result(
     task_id: AsyncJobId,
     user_id: Annotated[UserID, Depends(get_current_user_id)],
     product_name: Annotated[ProductName, Depends(get_product_name)],
-    rabbitmq_rpc_client: Annotated[RabbitMQRPCClient, Depends(get_rabbitmq_rpc_client)],
+    async_jobs: Annotated[AsyncJobClient, Depends(get_async_jobs_client)],
 ):
     async_job_rpc_result = await async_jobs.result(
-        rabbitmq_rpc_client=rabbitmq_rpc_client,
-        rpc_namespace=STORAGE_RPC_NAMESPACE,
         job_id=task_id,
         job_id_data=_get_job_id_data(user_id, product_name),
     )

services/api-server/src/simcore_service_api_server/exceptions/handlers/_handlers_backend_errors.py

@@ -1,12 +1,30 @@
+import logging
+
+from common_library.error_codes import create_error_code
+from fastapi import status
+from servicelib.logging_errors import create_troubleshootting_log_kwargs
 from starlette.requests import Request
 from starlette.responses import JSONResponse
 
 from ...exceptions.backend_errors import BaseBackEndError
 from ._utils import create_error_json_response
 
+_logger = logging.getLogger(__name__)
+
 
 async def backend_error_handler(request: Request, exc: Exception) -> JSONResponse:
     assert request  # nosec
     assert isinstance(exc, BaseBackEndError)
-
-    return create_error_json_response(f"{exc}", status_code=exc.status_code)
+    user_error_msg = f"{exc}"
+    if not exc.status_code >= status.HTTP_500_INTERNAL_SERVER_ERROR:
+        oec = create_error_code(exc)
+        user_error_msg += f" [{oec}]"
+        _logger.exception(
+            **create_troubleshootting_log_kwargs(
+                user_error_msg,
+                error=exc,
+                error_code=oec,
+                tip="Unexpected error",
+            )
+        )
+    return create_error_json_response(user_error_msg, status_code=exc.status_code)

services/api-server/src/simcore_service_api_server/exceptions/task_errors.py

@@ -0,0 +1,37 @@
+from fastapi import status
+
+from .backend_errors import BaseBackEndError
+
+
+class TaskBaseError(BaseBackEndError):
+    pass
+
+
+class TaskSchedulerError(TaskBaseError):
+    msg_template: str = "TaskScheduler error"
+    status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
+
+
+class TaskMissingError(TaskBaseError):
+    msg_template: str = "Task {job_id} does not exist"
+    status_code = status.HTTP_404_NOT_FOUND
+
+
+class TaskStatusError(TaskBaseError):
+    msg_template: str = "Could not get status of task {job_id}"
+    status_code = status.HTTP_404_NOT_FOUND
+
+
+class TaskNotDoneError(TaskBaseError):
+    msg_template: str = "Task {job_id} not done"
+    status_code = status.HTTP_409_CONFLICT
+
+
+class TaskCancelledError(TaskBaseError):
+    msg_template: str = "Task {job_id} cancelled"
+    status_code = status.HTTP_409_CONFLICT
+
+
+class TaskError(TaskBaseError):
+    msg_template: str = "Task '{job_id}' failed"
+    status_code = status.HTTP_500_INTERNAL_SERVER_ERROR

services/api-server/src/simcore_service_api_server/services_rpc/async_jobs.py

@@ -1,7 +1,93 @@
+import functools
+from dataclasses import dataclass
+
+from models_library.api_schemas_rpc_async_jobs.async_jobs import (
+    AsyncJobGet,
+    AsyncJobId,
+    AsyncJobNameData,
+    AsyncJobResult,
+    AsyncJobStatus,
+)
+from models_library.api_schemas_rpc_async_jobs.exceptions import (
+    JobAbortedError,
+    JobError,
+    JobMissingError,
+    JobNotDoneError,
+    JobSchedulerError,
+    JobStatusError,
+)
+from models_library.api_schemas_storage import STORAGE_RPC_NAMESPACE
+from servicelib.long_running_tasks.errors import TaskCancelledError
 from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
+from servicelib.rabbitmq.rpc_interfaces.async_jobs import async_jobs
+from simcore_service_api_server.exceptions.task_errors import (
+    TaskError,
+    TaskMissingError,
+    TaskNotDoneError,
+    TaskSchedulerError,
+    TaskStatusError,
+)
+
+from ..exceptions.service_errors_utils import service_exception_mapper
+
+_exception_mapper = functools.partial(
+    service_exception_mapper, service_name="Async jobs"
+)
+
+_exception_map = {
+    JobSchedulerError: TaskSchedulerError,
+    JobMissingError: TaskMissingError,
+    JobStatusError: TaskStatusError,
+    JobNotDoneError: TaskNotDoneError,
+    JobAbortedError: TaskCancelledError,
+    JobError: TaskError,
+}
 
 
+@dataclass
 class AsyncJobClient:
+    _rabbitmq_rpc_client: RabbitMQRPCClient
+
+    @_exception_mapper(rpc_exception_map=_exception_map)
+    async def cancel(
+        self, *, job_id: AsyncJobId, job_id_data: AsyncJobNameData
+    ) -> None:
+        return await async_jobs.cancel(
+            self._rabbitmq_rpc_client,
+            rpc_namespace=STORAGE_RPC_NAMESPACE,
+            job_id=job_id,
+            job_id_data=job_id_data,
+        )
+
+    @_exception_mapper(rpc_exception_map=_exception_map)
+    async def status(
+        self, *, job_id: AsyncJobId, job_id_data: AsyncJobNameData
+    ) -> AsyncJobStatus:
+        return await async_jobs.status(
+            self._rabbitmq_rpc_client,
+            rpc_namespace=STORAGE_RPC_NAMESPACE,
+            job_id=job_id,
+            job_id_data=job_id_data,
+        )
+
+    @_exception_mapper(rpc_exception_map=_exception_map)
+    async def result(
+        self, *, job_id: AsyncJobId, job_id_data: AsyncJobNameData
+    ) -> AsyncJobResult:
+        return await async_jobs.result(
+            self._rabbitmq_rpc_client,
+            rpc_namespace=STORAGE_RPC_NAMESPACE,
+            job_id=job_id,
+            job_id_data=job_id_data,
+        )
 
-    def __init__(self, rabbitmq_rpc_client: RabbitMQRPCClient):
-        self._rabbitmq_rpc_client = rabbitmq_rpc_client
+    @_exception_mapper(rpc_exception_map=_exception_map)
+    async def list_jobs(
+        self, *, filter_: str, job_id_data: AsyncJobNameData
+    ) -> list[AsyncJobGet]:
+        return await async_jobs.list_jobs(
+            self._rabbitmq_rpc_client,
+            rpc_namespace=STORAGE_RPC_NAMESPACE,
+            filter_=filter_,
+            job_id_data=job_id_data,
+        )