start introducing celery task manager into function job service layer

bisgaard-itis · bisgaard-itis · commit d65388276c7a · 2025-09-09T09:11:18.000+02:00
diff --git a/services/api-server/src/simcore_service_api_server/_service_function_jobs.py b/services/api-server/src/simcore_service_api_server/_service_function_jobs.py
@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import overload
+from typing import Final, overload
 
 import jsonschema
 from common_library.exclude import as_dict_exclude_none
@@ -38,21 +38,28 @@
 from models_library.rpc_pagination import PageLimitInt
 from models_library.users import UserID
 from pydantic import ValidationError
+from servicelib.celery.models import TaskUUID
+from servicelib.celery.task_manager import TaskManager
 from simcore_service_api_server._service_functions import FunctionService
 from simcore_service_api_server.services_rpc.storage import StorageService
 from sqlalchemy.ext.asyncio import AsyncEngine
 
 from ._service_jobs import JobService
+from .api.routes.tasks import _get_task_filter
 from .exceptions.function_errors import (
     FunctionJobCacheNotFoundError,
-    FunctionJobProjectMissingError,
 )
 from .models.api_resources import JobLinks
 from .models.domain.functions import PreRegisteredFunctionJobData
 from .models.schemas.jobs import JobInputs, JobPricingSpecification
 from .services_http.webserver import AuthSession
 from .services_rpc.wb_api_server import WbApiRpcClient
 
+_JOB_CREATION_TASK_STATUS_PREFIX: Final[str] = "JOB_CREATION_TASK_STATUS_"
+_JOB_CREATION_TASK_NOT_YET_SCHEDULED_STATUS: Final[str] = (
+    f"{_JOB_CREATION_TASK_STATUS_PREFIX}NOT_YET_SCHEDULED"
+)
+
 
 def join_inputs(
     default_inputs: FunctionInputs | None,
@@ -68,6 +75,23 @@ def join_inputs(
     return {**default_inputs, **function_inputs}
 
 
+async def _celery_task_status(
+    job_creation_task_id: TaskID | None,
+    task_manager: TaskManager,
+    user_id: UserID,
+    product_name: ProductName,
+) -> FunctionJobStatus:
+    if job_creation_task_id is None:
+        return FunctionJobStatus(status=_JOB_CREATION_TASK_NOT_YET_SCHEDULED_STATUS)
+    task_filter = _get_task_filter(user_id, product_name)
+    task_status = await task_manager.get_task_status(
+        task_uuid=TaskUUID(job_creation_task_id), task_filter=task_filter
+    )
+    return FunctionJobStatus(
+        status=f"{_JOB_CREATION_TASK_STATUS_PREFIX}{task_status.task_state}"
+    )
+
+
 @dataclass(frozen=True, kw_only=True)
 class FunctionJobService:
     user_id: UserID
@@ -159,7 +183,10 @@ async def validate_function_inputs(
         )
 
     async def inspect_function_job(
-        self, function: RegisteredFunction, function_job: RegisteredFunctionJob
+        self,
+        function: RegisteredFunction,
+        function_job: RegisteredFunctionJob,
+        task_manager: TaskManager,
     ) -> FunctionJobStatus:
         """Raises FunctionJobProjectNotRegisteredError if no project is associated with job"""
         stored_job_status = await self._web_rpc_client.get_function_job_status(
@@ -176,15 +203,25 @@ async def inspect_function_job(
             and function_job.function_class == FunctionClass.PROJECT
         ):
             if function_job.project_job_id is None:
-                raise FunctionJobProjectMissingError
+                return await _celery_task_status(
+                    job_creation_task_id=function_job.job_creation_task_id,
+                    task_manager=task_manager,
+                    user_id=self.user_id,
+                    product_name=self.product_name,
+                )
             job_status = await self._job_service.inspect_study_job(
                 job_id=function_job.project_job_id,
             )
         elif (function.function_class == FunctionClass.SOLVER) and (
             function_job.function_class == FunctionClass.SOLVER
         ):
             if function_job.solver_job_id is None:
-                raise FunctionJobProjectMissingError
+                return await _celery_task_status(
+                    job_creation_task_id=function_job.job_creation_task_id,
+                    task_manager=task_manager,
+                    user_id=self.user_id,
+                    product_name=self.product_name,
+                )
             job_status = await self._job_service.inspect_solver_job(
                 solver_key=function.solver_key,
                 version=function.solver_version,
@@ -225,6 +262,7 @@ async def get_cached_function_job(
         *,
         function: RegisteredFunction,
         job_inputs: JobInputs,
+        task_manager: TaskManager,
     ) -> RegisteredFunctionJob:
         """
         N.B. this function checks access rights
@@ -266,6 +304,7 @@ async def get_cached_function_job(
                 job_status = await self.inspect_function_job(
                     function=function,
                     function_job=cached_function_job,
+                    task_manager=task_manager,
                 )
                 if job_status.status == RunningState.SUCCESS:
                     return cached_function_job
@@ -476,18 +515,15 @@ async def function_job_outputs(
         product_name: ProductName,
         stored_job_outputs: FunctionOutputs | None,
         async_pg_engine: AsyncEngine,
+        task_manager: TaskManager,
     ) -> FunctionOutputs:
 
         if stored_job_outputs is not None:
             return stored_job_outputs
 
-        try:
-            job_status = await self.inspect_function_job(
-                function=function,
-                function_job=function_job,
-            )
-        except FunctionJobProjectMissingError:
-            return None
+        job_status = await self.inspect_function_job(
+            function=function, function_job=function_job, task_manager=task_manager
+        )
 
         if job_status.status != RunningState.SUCCESS:
             return None
diff --git a/services/api-server/src/simcore_service_api_server/api/routes/function_job_collections_routes.py b/services/api-server/src/simcore_service_api_server/api/routes/function_job_collections_routes.py
@@ -1,6 +1,6 @@
 from typing import Annotated, Final
 
-from fastapi import APIRouter, Depends, status
+from fastapi import APIRouter, Depends, FastAPI, status
 from fastapi_pagination.api import create_page
 from fastapi_pagination.bases import AbstractPage
 from models_library.api_schemas_webserver.functions import (
@@ -14,12 +14,14 @@
 from models_library.products import ProductName
 from models_library.users import UserID
 from servicelib.utils import limited_gather
-from simcore_service_api_server._service_function_jobs import FunctionJobService
 
+from ..._service_function_jobs import FunctionJobService
 from ...models.pagination import Page, PaginationParams
 from ...models.schemas.errors import ErrorGet
 from ...services_rpc.wb_api_server import WbApiRpcClient
+from ..dependencies.application import get_app
 from ..dependencies.authentication import get_current_user_id, get_product_name
+from ..dependencies.celery import get_task_manager
 from ..dependencies.functions import (
     get_function_from_functionjobid,
 )
@@ -254,6 +256,7 @@ async def function_job_collection_list_function_jobs_list(
     ),
 )
 async def function_job_collection_status(
+    app: Annotated[FastAPI, Depends(get_app)],
     function_job_collection_id: FunctionJobCollectionID,
     wb_api_rpc: Annotated[WbApiRpcClient, Depends(get_wb_api_rpc_client)],
     user_id: Annotated[UserID, Depends(get_current_user_id)],  # Updated type
@@ -262,6 +265,7 @@ async def function_job_collection_status(
         FunctionJobService, Depends(get_function_job_service)
     ],
 ) -> FunctionJobCollectionStatus:
+    task_manager = get_task_manager(app)
     function_job_collection = await get_function_job_collection(
         function_job_collection_id=function_job_collection_id,
         wb_api_rpc=wb_api_rpc,
@@ -284,6 +288,7 @@ async def function_job_collection_status(
                     user_id=user_id,
                     product_name=product_name,
                 ),
+                task_manager=task_manager,
             )
             for function_job_id in function_job_collection.job_ids
         ]
diff --git a/services/api-server/src/simcore_service_api_server/api/routes/function_jobs_routes.py b/services/api-server/src/simcore_service_api_server/api/routes/function_jobs_routes.py
@@ -1,7 +1,6 @@
 from logging import getLogger
 from typing import Annotated, Final
 
-from common_library.error_codes import create_error_code
 from fastapi import APIRouter, Depends, FastAPI, HTTPException, Query, status
 from fastapi_pagination.api import create_page
 from models_library.api_schemas_long_running_tasks.tasks import TaskGet
@@ -16,20 +15,16 @@
 from models_library.functions import RegisteredFunction
 from models_library.functions_errors import (
     UnsupportedFunctionClassError,
-    UnsupportedFunctionFunctionJobClassCombinationError,
 )
 from models_library.products import ProductName
 from models_library.projects_state import RunningState
 from models_library.users import UserID
-from servicelib.celery.models import TaskUUID
 from servicelib.fastapi.dependencies import get_app
-from servicelib.logging_errors import create_troubleshootting_log_kwargs
 from sqlalchemy.ext.asyncio import AsyncEngine
 
 from ..._service_function_jobs import FunctionJobService
 from ..._service_functions import FunctionService
 from ..._service_jobs import JobService
-from ...exceptions.function_errors import FunctionJobProjectMissingError
 from ...models.domain.functions import PageRegisteredFunctionJobWithorWithoutStatus
 from ...models.pagination import PaginationParams
 from ...models.schemas.errors import ErrorGet
@@ -55,7 +50,6 @@
     FMSG_CHANGELOG_NEW_IN_VERSION,
     create_route_description,
 )
-from .tasks import _get_task_filter
 
 _logger = getLogger(__name__)
 
@@ -66,7 +60,6 @@
 JOB_LIST_FILTER_PAGE_RELEASE_VERSION = "0.11.0"
 JOB_LOG_RELEASE_VERSION = "0.11.0"
 WITH_STATUS_RELEASE_VERSION = "0.13.0"
-_JOB_CREATION_TASK_STATUS_PREFIX: Final[str] = "JOB_CREATION_TASK_STATUS_"
 
 function_job_router = APIRouter()
 
@@ -280,43 +273,10 @@ async def function_job_status(
         FunctionJobService, Depends(get_function_job_service)
     ],
 ) -> FunctionJobStatus:
-    try:
-        return await function_job_service.inspect_function_job(
-            function=function, function_job=function_job
-        )
-    except FunctionJobProjectMissingError as exc:
-        if (
-            function.function_class == FunctionClass.PROJECT
-            and function_job.function_class == FunctionClass.PROJECT
-        ) or (
-            function.function_class == FunctionClass.SOLVER
-            and function_job.function_class == FunctionClass.SOLVER
-        ):
-            if task_id := function_job.job_creation_task_id:
-                task_manager = get_task_manager(app)
-                task_filter = _get_task_filter(user_id, product_name)
-                task_status = await task_manager.get_task_status(
-                    task_uuid=TaskUUID(task_id), task_filter=task_filter
-                )
-                return FunctionJobStatus(
-                    status=f"{_JOB_CREATION_TASK_STATUS_PREFIX}{task_status.task_state}"
-                )
-            user_error_msg = f"The creation of job {function_job.uid} failed"
-            support_id = create_error_code(Exception())
-            _logger.exception(
-                **create_troubleshootting_log_kwargs(
-                    user_error_msg,
-                    error=Exception(),
-                    error_code=support_id,
-                    tip="Initial call to run metamodeling function must have failed",
-                )
-            )
-            raise
-
-        raise UnsupportedFunctionFunctionJobClassCombinationError(
-            function_class=function.function_class,
-            function_job_class=function_job.function_class,
-        ) from exc
+    task_manager = get_task_manager(app)
+    return await function_job_service.inspect_function_job(
+        function=function, function_job=function_job, task_manager=task_manager
+    )
 
 
 async def get_function_from_functionjobid(
