add mehod for patching a registered function job to function job service

Author: bisgaard-itis

packages/models-library/src/models_library/functions_errors.py

@@ -163,3 +163,8 @@ class FunctionJobCollectionsExecuteApiAccessDeniedError(FunctionBaseError):
 class FunctionJobPatchModelIncompatibleError(FunctionBaseError):
     msg_template = "Incompatible patch model for Function '{function_id}' in product '{product_name}'."
     status_code: int = 422
+
+
+class FunctionJobCacheNotFoundError(FunctionBaseError):
+    msg_template: str = "No cached function job found."
+    status_code: int = 404  # Not Found

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

@@ -1,4 +1,5 @@
 from dataclasses import dataclass
+from typing import NamedTuple, overload
 
 import jsonschema
 from common_library.exclude import as_dict_exclude_none
@@ -16,11 +17,16 @@
     RegisteredFunction,
     RegisteredFunctionJob,
     RegisteredFunctionJobCollection,
+    RegisteredProjectFunctionJobPatch,
+    RegisteredSolverFunctionJobPatch,
     SolverFunctionJob,
+    SolverJobID,
+    TaskID,
 )
 from models_library.functions_errors import (
     FunctionExecuteAccessDeniedError,
     FunctionInputsValidationError,
+    FunctionJobCacheNotFoundError,
     FunctionsExecuteApiAccessDeniedError,
     UnsupportedFunctionClassError,
     UnsupportedFunctionFunctionJobClassCombinationError,
@@ -43,6 +49,11 @@
 from .services_rpc.wb_api_server import WbApiRpcClient
 
 
+class RegisteredFunctionJobData(NamedTuple):
+    function_job_id: FunctionJobID
+    job_inputs: JobInputs
+
+
 def join_inputs(
     default_inputs: FunctionInputs | None,
     function_inputs: FunctionInputs | None,
@@ -162,12 +173,29 @@ async def inspect_function_job(
             job_status=new_job_status,
         )
 
-    async def run_function_pre_check(
+    async def create_function_job_inputs(
         self,
         *,
         function: RegisteredFunction,
         function_inputs: FunctionInputs,
     ) -> JobInputs:
+        joined_inputs = join_inputs(
+            function.default_inputs,
+            function_inputs,
+        )
+        return JobInputs(
+            values=joined_inputs or {},
+        )
+
+    async def get_cached_function_job(
+        self,
+        *,
+        function: RegisteredFunction,
+        function_inputs: FunctionInputs,
+        job_inputs: JobInputs,
+    ) -> RegisteredFunctionJob:
+        """raises FunctionJobCacheNotFoundError if no cached job is found"""
+
         user_api_access_rights = (
             await self._web_rpc_client.get_functions_user_api_access_rights(
                 user_id=self.user_id, product_name=self.product_name
@@ -190,21 +218,137 @@ async def run_function_pre_check(
                 function_id=function.uid,
             )
 
-        joined_inputs = join_inputs(
-            function.default_inputs,
-            function_inputs,
-        )
+        if cached_function_jobs := await self._web_rpc_client.find_cached_function_jobs(
+            function_id=function.uid,
+            inputs=job_inputs.values,
+            user_id=self.user_id,
+            product_name=self.product_name,
+        ):
+            for cached_function_job in cached_function_jobs:
+                job_status = await self.inspect_function_job(
+                    function=function,
+                    function_job=cached_function_job,
+                )
+                if job_status.status == RunningState.SUCCESS:
+                    return cached_function_job
+
+        raise FunctionJobCacheNotFoundError()
+
+    async def create_registered_function_job(
+        self,
+        *,
+        function: RegisteredFunction,
+        function_inputs: FunctionInputs,
+        job_inputs: JobInputs,
+    ) -> FunctionJobID:
 
         if function.input_schema is not None:
             is_valid, validation_str = await self.validate_function_inputs(
                 function_id=function.uid,
-                inputs=joined_inputs,
+                inputs=job_inputs.values,
             )
             if not is_valid:
                 raise FunctionInputsValidationError(error=validation_str)
 
-        return JobInputs(
-            values=joined_inputs or {},
+        if function.function_class == FunctionClass.PROJECT:
+            job = await self._web_rpc_client.register_function_job(
+                function_job=ProjectFunctionJob(
+                    function_uid=function.uid,
+                    title=f"Function job of function {function.uid}",
+                    description=function.description,
+                    inputs=job_inputs.values,
+                    outputs=None,
+                    project_job_id=None,
+                    job_creation_task_id=None,
+                ),
+                user_id=self.user_id,
+                product_name=self.product_name,
+            )
+
+        elif function.function_class == FunctionClass.SOLVER:
+            job = await self._web_rpc_client.register_function_job(
+                function_job=SolverFunctionJob(
+                    function_uid=function.uid,
+                    title=f"Function job of function {function.uid}",
+                    description=function.description,
+                    inputs=job_inputs.values,
+                    outputs=None,
+                    solver_job_id=None,
+                    job_creation_task_id=None,
+                ),
+                user_id=self.user_id,
+                product_name=self.product_name,
+            )
+        else:
+            raise UnsupportedFunctionClassError(
+                function_class=function.function_class,
+            )
+
+        return job.uid
+
+    @overload
+    async def patch_registered_function_job(
+        self,
+        *,
+        user_id: UserID,
+        product_name: ProductName,
+        function_job_id: FunctionJobID,
+        function_class: FunctionClass,
+        job_creation_task_id: TaskID | None,
+        project_job_id: ProjectID | None,
+    ) -> RegisteredFunctionJob: ...
+
+    @overload
+    async def patch_registered_function_job(
+        self,
+        *,
+        user_id: UserID,
+        product_name: ProductName,
+        function_job_id: FunctionJobID,
+        function_class: FunctionClass,
+        job_creation_task_id: TaskID | None,
+        solver_job_id: SolverJobID | None,
+    ) -> RegisteredFunctionJob: ...
+
+    async def patch_registered_function_job(
+        self,
+        *,
+        user_id: UserID,
+        product_name: ProductName,
+        function_job_id: FunctionJobID,
+        function_class: FunctionClass,
+        job_creation_task_id: TaskID | None,
+        project_job_id: ProjectID | None = None,
+        solver_job_id: SolverJobID | None = None,
+    ) -> RegisteredFunctionJob:
+        # Only allow one of project_job_id or solver_job_id depending on function_class
+        if function_class == FunctionClass.PROJECT:
+            patch = RegisteredProjectFunctionJobPatch(
+                title=None,
+                description=None,
+                inputs=None,
+                outputs=None,
+                job_creation_task_id=job_creation_task_id,
+                project_job_id=project_job_id,
+            )
+        elif function_class == FunctionClass.SOLVER:
+            patch = RegisteredSolverFunctionJobPatch(
+                title=None,
+                description=None,
+                inputs=None,
+                outputs=None,
+                job_creation_task_id=job_creation_task_id,
+                solver_job_id=solver_job_id,
+            )
+        else:
+            raise UnsupportedFunctionClassError(
+                function_class=function_class,
+            )
+        return await self._web_rpc_client.patch_registered_function_job(
+            user_id=user_id,
+            product_name=product_name,
+            function_job_id=function_job_id,
+            registered_function_job_patch=patch,
         )
 
     async def run_function(
@@ -217,20 +361,7 @@ async def run_function(
         x_simcore_parent_project_uuid: NodeID | None,
         x_simcore_parent_node_id: NodeID | None,
     ) -> RegisteredFunctionJob:
-
-        if cached_function_jobs := await self._web_rpc_client.find_cached_function_jobs(
-            function_id=function.uid,
-            inputs=job_inputs.values,
-            user_id=self.user_id,
-            product_name=self.product_name,
-        ):
-            for cached_function_job in cached_function_jobs:
-                job_status = await self.inspect_function_job(
-                    function=function,
-                    function_job=cached_function_job,
-                )
-                if job_status.status == RunningState.SUCCESS:
-                    return cached_function_job
+        """N.B. this function does not check access rights. Use get_cached_function_job for that"""
 
         if function.function_class == FunctionClass.PROJECT:
             study_job = await self._job_service.create_studies_job(
@@ -306,7 +437,7 @@ async def map_function(
     ) -> RegisteredFunctionJobCollection:
 
         job_inputs = [
-            await self.run_function_pre_check(
+            await self.create_registered_function_job(
                 function=function,
                 function_inputs=inputs,
             )
@@ -335,3 +466,25 @@ async def map_function(
             user_id=self.user_id,
             product_name=self.product_name,
         )
+        function_jobs = [
+            await self.run_function(
+                function=function,
+                job_inputs=inputs,
+                pricing_spec=pricing_spec,
+                job_links=job_links,
+                x_simcore_parent_project_uuid=x_simcore_parent_project_uuid,
+                x_simcore_parent_node_id=x_simcore_parent_node_id,
+            )
+            for inputs in job_inputs
+        ]
+
+        function_job_collection_description = f"Function job collection of map of function {function.uid} with {len(function_inputs_list)} inputs"
+        return await self._web_rpc_client.register_function_job_collection(
+            function_job_collection=FunctionJobCollection(
+                title="Function job collection of function map",
+                description=function_job_collection_description,
+                job_ids=[function_job.uid for function_job in function_jobs],
+            ),
+            user_id=self.user_id,
+            product_name=self.product_name,
+        )

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

@@ -17,13 +17,13 @@
     RegisteredFunctionJobCollection,
 )
 from models_library.api_schemas_rpc_async_jobs.async_jobs import AsyncJobFilter
+from models_library.functions_errors import FunctionJobCacheNotFoundError
 from models_library.products import ProductName
 from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
 from models_library.users import UserID
 from servicelib.celery.models import TaskFilter, TaskMetadata, TasksQueue
 from servicelib.fastapi.dependencies import get_reverse_url_mapper
-from servicelib.long_running_tasks.models import TaskGet
 
 from ..._service_function_jobs import FunctionJobService
 from ..._service_functions import FunctionService
@@ -314,7 +314,7 @@ async def validate_function_inputs(
 
 @function_router.post(
     "/{function_id:uuid}:run",
-    response_model=TaskGet,
+    response_model=RegisteredFunctionJob,
     responses={**_COMMON_FUNCTION_ERROR_RESPONSES},
     description=create_route_description(
         base="Run function",
@@ -333,7 +333,7 @@ async def run_function(  # noqa: PLR0913
     ],
     x_simcore_parent_project_uuid: Annotated[ProjectID | Literal["null"], Header()],
     x_simcore_parent_node_id: Annotated[NodeID | Literal["null"], Header()],
-) -> TaskGet:
+) -> RegisteredFunctionJob:
     task_manager = get_task_manager(request.app)
     parent_project_uuid = (
         x_simcore_parent_project_uuid
@@ -348,10 +348,28 @@ async def run_function(  # noqa: PLR0913
     pricing_spec = JobPricingSpecification.create_from_headers(request.headers)
     job_links = await function_service.get_function_job_links(to_run_function, url_for)
 
-    job_inputs = await function_job_service.run_function_pre_check(
+    job_inputs = await function_job_service.create_function_job_inputs(
         function=to_run_function, function_inputs=function_inputs
     )
+    try:
+        # checks access rights
+        return await function_job_service.get_cached_function_job(
+            function=to_run_function,
+            function_inputs=function_inputs,
+            job_inputs=job_inputs,
+        )
+    except FunctionJobCacheNotFoundError:
+        pass
+
+    pre_registered_function_job_id = (
+        await function_job_service.create_registered_function_job(
+            function=to_run_function,
+            function_inputs=function_inputs,
+            job_inputs=job_inputs,
+        )
+    )
 
+    # run function in celery task
     job_filter = AsyncJobFilter(
         user_id=user_identity.user_id,
         product_name=user_identity.product_name,
@@ -376,12 +394,14 @@ async def run_function(  # noqa: PLR0913
         x_simcore_parent_node_id=parent_node_id,
     )
 
-    return TaskGet(
-        task_id=f"{task_uuid}",
-        task_name=task_name,
-        status_href=url_for("get_task_status", task_id=task_uuid),
-        result_href=url_for("get_task_result", task_id=task_uuid),
-        abort_href=url_for("cancel_task", task_id=task_uuid),
+    return await function_job_service.patch_registered_function_job(
+        user_id=user_identity.user_id,
+        product_name=user_identity.product_name,
+        function_job_id=pre_registered_function_job_id,
+        registered_function_job_patch=RegisteredFunctionJobPatch(
+            status=RunningState.RUNNING,
+            task_id=task_uuid,
+        ),
     )
 
 

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

@@ -28,6 +28,7 @@
     FunctionOutputs,
     FunctionUserAccessRights,
     FunctionUserApiAccessRights,
+    RegisteredFunctionJobPatch,
 )
 from models_library.licenses import LicensedItemID
 from models_library.products import ProductName
@@ -489,6 +490,22 @@ async def register_function_job(
             function_job=function_job,
         )
 
+    async def patch_registered_function_job(
+        self,
+        *,
+        user_id: UserID,
+        product_name: ProductName,
+        function_job_id: FunctionJobID,
+        registered_function_job_patch: RegisteredFunctionJobPatch,
+    ) -> RegisteredFunctionJob:
+        return await functions_rpc_interface.patch_registered_function_job(
+            self._client,
+            user_id=user_id,
+            product_name=product_name,
+            function_job_uuid=function_job_id,
+            registered_function_job_patch=registered_function_job_patch,
+        )
+
     async def get_function_input_schema(
         self, *, user_id: UserID, product_name: ProductName, function_id: FunctionID
     ) -> FunctionInputSchema: