move inspect function job to function_job_service

Author: bisgaard-itis

services/api-server/src/simcore_service_api_server/_service_function_jobs.py

@@ -2,23 +2,33 @@
 
 from common_library.exclude import as_dict_exclude_none
 from models_library.functions import (
+    FunctionClass,
     FunctionID,
     FunctionJobCollectionID,
     FunctionJobID,
+    FunctionJobStatus,
+    RegisteredFunction,
     RegisteredFunctionJob,
 )
+from models_library.functions_errors import (
+    UnsupportedFunctionFunctionJobClassCombinationError,
+)
 from models_library.products import ProductName
+from models_library.projects_state import RunningState
 from models_library.rest_pagination import PageMetaInfoLimitOffset, PageOffsetInt
 from models_library.rpc_pagination import PageLimitInt
 from models_library.users import UserID
-from simcore_service_api_server.services_rpc.wb_api_server import WbApiRpcClient
+
+from ._service_jobs import JobService
+from .services_rpc.wb_api_server import WbApiRpcClient
 
 
 @dataclass(frozen=True, kw_only=True)
 class FunctionJobService:
     user_id: UserID
     product_name: ProductName
     _web_rpc_client: WbApiRpcClient
+    _job_service: JobService
 
     async def list_function_jobs(
         self,
@@ -43,3 +53,46 @@ async def list_function_jobs(
             filter_by_function_job_collection_id=filter_by_function_job_collection_id,
             **pagination_kwargs,
         )
+
+    async def inspect_function_job(
+        self, function: RegisteredFunction, function_job: RegisteredFunctionJob
+    ) -> FunctionJobStatus:
+
+        stored_job_status = await self._web_rpc_client.get_function_job_status(
+            function_job_id=function_job.uid,
+            user_id=self.user_id,
+            product_name=self.product_name,
+        )
+
+        if stored_job_status.status in (RunningState.SUCCESS, RunningState.FAILED):
+            return stored_job_status
+
+        if (
+            function.function_class == FunctionClass.PROJECT
+            and function_job.function_class == FunctionClass.PROJECT
+        ):
+            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
+        ):
+            job_status = await self._job_service.inspect_solver_job(
+                solver_key=function.solver_key,
+                version=function.solver_version,
+                job_id=function_job.solver_job_id,
+            )
+        else:
+            raise UnsupportedFunctionFunctionJobClassCombinationError(
+                function_class=function.function_class,
+                function_job_class=function_job.function_class,
+            )
+
+        new_job_status = FunctionJobStatus(status=job_status.state)
+
+        return await self._web_rpc_client.update_function_job_status(
+            function_job_id=function_job.uid,
+            user_id=self.user_id,
+            product_name=self.product_name,
+            job_status=new_job_status,
+        )

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

@@ -150,11 +150,13 @@ def get_function_service(
 
 def get_function_job_service(
     web_rpc_api: Annotated[WbApiRpcClient, Depends(get_wb_api_rpc_client)],
+    job_service: Annotated[JobService, Depends(get_job_service)],
     user_id: Annotated[UserID, Depends(get_current_user_id)],
     product_name: Annotated[ProductName, Depends(get_product_name)],
 ) -> FunctionJobService:
     return FunctionJobService(
         _web_rpc_client=web_rpc_api,
+        _job_service=job_service,
         user_id=user_id,
         product_name=product_name,
     )

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

@@ -16,31 +16,26 @@
 from servicelib.utils import limited_gather
 from simcore_service_api_server._service_function_jobs import FunctionJobService
 
-from ..._service_jobs import JobService
 from ...models.pagination import Page, PaginationParams
 from ...models.schemas.errors import ErrorGet
-from ...services_http.director_v2 import DirectorV2Api
 from ...services_rpc.wb_api_server import WbApiRpcClient
 from ..dependencies.authentication import get_current_user_id, get_product_name
 from ..dependencies.functions import (
     get_function_from_functionjobid,
-    get_stored_job_status,
 )
 from ..dependencies.models_schemas_function_filters import (
     get_function_job_collections_filters,
 )
 from ..dependencies.services import (
-    get_api_client,
     get_function_job_service,
-    get_job_service,
 )
 from ..dependencies.webserver_rpc import get_wb_api_rpc_client
 from ._constants import (
     FMSG_CHANGELOG_ADDED_IN_VERSION,
     FMSG_CHANGELOG_NEW_IN_VERSION,
     create_route_description,
 )
-from .function_jobs_routes import function_job_status, get_function_job
+from .function_jobs_routes import get_function_job
 
 # pylint: disable=too-many-arguments
 
@@ -261,10 +256,11 @@ async def function_job_collection_list_function_jobs_list(
 async def function_job_collection_status(
     function_job_collection_id: FunctionJobCollectionID,
     wb_api_rpc: Annotated[WbApiRpcClient, Depends(get_wb_api_rpc_client)],
-    director2_api: Annotated[DirectorV2Api, Depends(get_api_client(DirectorV2Api))],
     user_id: Annotated[UserID, Depends(get_current_user_id)],  # Updated type
     product_name: Annotated[ProductName, Depends(get_product_name)],
-    job_service: Annotated[JobService, Depends(get_job_service)],
+    function_job_service: Annotated[
+        FunctionJobService, Depends(get_function_job_service)
+    ],
 ) -> FunctionJobCollectionStatus:
     function_job_collection = await get_function_job_collection(
         function_job_collection_id=function_job_collection_id,
@@ -275,7 +271,7 @@ async def function_job_collection_status(
 
     job_statuses = await limited_gather(
         *[
-            function_job_status(
+            function_job_service.inspect_function_job(
                 function_job=await get_function_job(
                     function_job_id=function_job_id,
                     wb_api_rpc=wb_api_rpc,
@@ -288,17 +284,6 @@ async def function_job_collection_status(
                     user_id=user_id,
                     product_name=product_name,
                 ),
-                stored_job_status=await get_stored_job_status(
-                    function_job_id=function_job_id,
-                    wb_api_rpc=wb_api_rpc,
-                    user_id=user_id,
-                    product_name=product_name,
-                ),
-                wb_api_rpc=wb_api_rpc,
-                director2_api=director2_api,
-                user_id=user_id,
-                product_name=product_name,
-                job_service=job_service,
             )
             for function_job_id in function_job_collection.job_ids
         ]

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

@@ -15,10 +15,8 @@
 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.fastapi.dependencies import get_app
 from simcore_service_api_server.models.schemas.functions_filters import (
@@ -30,7 +28,6 @@
 from ..._service_jobs import JobService
 from ...models.pagination import Page, PaginationParams
 from ...models.schemas.errors import ErrorGet
-from ...services_http.director_v2 import DirectorV2Api
 from ...services_http.storage import StorageApi
 from ...services_http.webserver import AuthSession
 from ...services_rpc.wb_api_server import WbApiRpcClient
@@ -40,7 +37,6 @@
     get_function_from_functionjob,
     get_function_job_dependency,
     get_stored_job_outputs,
-    get_stored_job_status,
 )
 from ..dependencies.models_schemas_function_filters import get_function_jobs_filters
 from ..dependencies.services import (
@@ -50,7 +46,7 @@
 )
 from ..dependencies.webserver_http import get_webserver_session
 from ..dependencies.webserver_rpc import get_wb_api_rpc_client
-from . import solvers_jobs, solvers_jobs_read, studies_jobs
+from . import solvers_jobs_read, studies_jobs
 from ._constants import (
     FMSG_CHANGELOG_ADDED_IN_VERSION,
     FMSG_CHANGELOG_NEW_IN_VERSION,
@@ -204,48 +200,13 @@ async def function_job_status(
         RegisteredFunctionJob, Depends(get_function_job_dependency)
     ],
     function: Annotated[RegisteredFunction, Depends(get_function_from_functionjob)],
-    stored_job_status: Annotated[FunctionJobStatus, Depends(get_stored_job_status)],
-    director2_api: Annotated[DirectorV2Api, Depends(get_api_client(DirectorV2Api))],
-    job_service: Annotated[JobService, Depends(get_job_service)],
-    user_id: Annotated[UserID, Depends(get_current_user_id)],
-    product_name: Annotated[ProductName, Depends(get_product_name)],
-    wb_api_rpc: Annotated[WbApiRpcClient, Depends(get_wb_api_rpc_client)],
+    function_job_service: Annotated[
+        FunctionJobService, Depends(get_function_job_service)
+    ],
 ) -> FunctionJobStatus:
 
-    if stored_job_status.status in (RunningState.SUCCESS, RunningState.FAILED):
-        return stored_job_status
-
-    if (
-        function.function_class == FunctionClass.PROJECT
-        and function_job.function_class == FunctionClass.PROJECT
-    ):
-        job_status = await studies_jobs.inspect_study_job(
-            study_id=function.project_id,
-            job_id=function_job.project_job_id,
-            job_service=job_service,
-        )
-    elif (function.function_class == FunctionClass.SOLVER) and (
-        function_job.function_class == FunctionClass.SOLVER
-    ):
-        job_status = await solvers_jobs.inspect_job(
-            solver_key=function.solver_key,
-            version=function.solver_version,
-            job_id=function_job.solver_job_id,
-            job_service=job_service,
-        )
-    else:
-        raise UnsupportedFunctionFunctionJobClassCombinationError(
-            function_class=function.function_class,
-            function_job_class=function_job.function_class,
-        )
-
-    new_job_status = FunctionJobStatus(status=job_status.state)
-
-    return await wb_api_rpc.update_function_job_status(
-        function_job_id=function_job.uid,
-        user_id=user_id,
-        product_name=product_name,
-        job_status=new_job_status,
+    return await function_job_service.inspect_function_job(
+        function=function, function_job=function_job
     )
 
 

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

@@ -40,23 +40,16 @@
 from ..._service_function_jobs import FunctionJobService
 from ..._service_functions import FunctionService
 from ..._service_jobs import JobService
-from ..._service_solvers import SolverService
 from ...models.pagination import Page, PaginationParams
 from ...models.schemas.errors import ErrorGet
 from ...models.schemas.jobs import JobInputs
-from ...services_http.director_v2 import DirectorV2Api
-from ...services_http.webserver import AuthSession
 from ...services_rpc.wb_api_server import WbApiRpcClient
 from ..dependencies.authentication import get_current_user_id, get_product_name
-from ..dependencies.functions import get_stored_job_status
 from ..dependencies.services import (
-    get_api_client,
     get_function_job_service,
     get_function_service,
     get_job_service,
-    get_solver_service,
 )
-from ..dependencies.webserver_http import get_webserver_session
 from ..dependencies.webserver_rpc import get_wb_api_rpc_client
 from . import solvers_jobs, studies_jobs
 from ._constants import (
@@ -372,13 +365,13 @@ async def run_function(  # noqa: PLR0913
     function_id: FunctionID,
     to_run_function: Annotated[RegisteredFunction, Depends(get_function)],
     wb_api_rpc: Annotated[WbApiRpcClient, Depends(get_wb_api_rpc_client)],
-    webserver_api: Annotated[AuthSession, Depends(get_webserver_session)],
     url_for: Annotated[Callable, Depends(get_reverse_url_mapper)],
-    director2_api: Annotated[DirectorV2Api, Depends(get_api_client(DirectorV2Api))],
     function_inputs: FunctionInputs,
     user_id: Annotated[UserID, Depends(get_current_user_id)],
     product_name: Annotated[str, Depends(get_product_name)],
-    solver_service: Annotated[SolverService, Depends(get_solver_service)],
+    function_jobs_service: Annotated[
+        FunctionJobService, Depends(get_function_job_service)
+    ],
     job_service: Annotated[JobService, Depends(get_job_service)],
     x_simcore_parent_project_uuid: Annotated[ProjectID | Literal["null"], Header()],
     x_simcore_parent_node_id: Annotated[NodeID | Literal["null"], Header()],
@@ -414,8 +407,6 @@ async def run_function(  # noqa: PLR0913
             function_id=function_id,
         )
 
-    from .function_jobs_routes import function_job_status
-
     joined_inputs = _join_inputs(
         to_run_function.default_inputs,
         function_inputs,
@@ -439,20 +430,9 @@ async def run_function(  # noqa: PLR0913
         product_name=product_name,
     ):
         for cached_function_job in cached_function_jobs:
-            job_status = await function_job_status(
+            job_status = await function_jobs_service.inspect_function_job(
                 function=to_run_function,
                 function_job=cached_function_job,
-                stored_job_status=await get_stored_job_status(
-                    function_job_id=cached_function_job.uid,
-                    user_id=user_id,
-                    product_name=product_name,
-                    wb_api_rpc=wb_api_rpc,
-                ),
-                wb_api_rpc=wb_api_rpc,
-                user_id=user_id,
-                director2_api=director2_api,
-                product_name=product_name,
-                job_service=job_service,
             )
             if job_status.status == RunningState.SUCCESS:
                 return cached_function_job
@@ -565,13 +545,13 @@ async def map_function(  # noqa: PLR0913
     to_run_function: Annotated[RegisteredFunction, Depends(get_function)],
     function_inputs_list: FunctionInputsList,
     wb_api_rpc: Annotated[WbApiRpcClient, Depends(get_wb_api_rpc_client)],
-    webserver_api: Annotated[AuthSession, Depends(get_webserver_session)],
     url_for: Annotated[Callable, Depends(get_reverse_url_mapper)],
-    director2_api: Annotated[DirectorV2Api, Depends(get_api_client(DirectorV2Api))],
     user_id: Annotated[UserID, Depends(get_current_user_id)],
     product_name: Annotated[str, Depends(get_product_name)],
-    solver_service: Annotated[SolverService, Depends(get_solver_service)],
     job_service: Annotated[JobService, Depends(get_job_service)],
+    function_jobs_service: Annotated[
+        FunctionJobService, Depends(get_function_job_service)
+    ],
     x_simcore_parent_project_uuid: Annotated[ProjectID | Literal["null"], Header()],
     x_simcore_parent_node_id: Annotated[NodeID | Literal["null"], Header()],
 ) -> RegisteredFunctionJobCollection:
@@ -583,12 +563,10 @@ async def map_function(  # noqa: PLR0913
             function_inputs=function_inputs,
             product_name=product_name,
             user_id=user_id,
-            webserver_api=webserver_api,
             url_for=url_for,
-            director2_api=director2_api,
             request=request,
-            solver_service=solver_service,
             job_service=job_service,
+            function_jobs_service=function_jobs_service,
             x_simcore_parent_project_uuid=x_simcore_parent_project_uuid,
             x_simcore_parent_node_id=x_simcore_parent_node_id,
         )