use patch batch endpoint

Author: bisgaard-itis

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

@@ -1,25 +1,23 @@
 from dataclasses import dataclass
-from typing import Annotated
 
 import jsonschema
 from common_library.exclude import as_dict_exclude_none
 from models_library.functions import (
+    BatchUpdateRegisteredFunctionJobs,
     FunctionClass,
     FunctionID,
     FunctionInputs,
     FunctionJobCollectionID,
     FunctionJobID,
     FunctionJobList,
+    FunctionJobPatchRequest,
+    FunctionJobPatchRequestList,
     FunctionSchemaClass,
     ProjectFunctionJob,
     RegisteredFunction,
     RegisteredFunctionJob,
     RegisteredProjectFunctionJobPatch,
-    RegisteredProjectFunctionJobPatchInput,
-    RegisteredProjectFunctionJobPatchInputList,
     RegisteredSolverFunctionJobPatch,
-    RegisteredSolverFunctionJobPatchInput,
-    RegisteredSolverFunctionJobPatchInputList,
     SolverFunctionJob,
 )
 from models_library.functions_errors import (
@@ -31,7 +29,7 @@
 from models_library.rest_pagination import PageMetaInfoLimitOffset, PageOffsetInt
 from models_library.rpc_pagination import PageLimitInt
 from models_library.users import UserID
-from pydantic import Field, TypeAdapter, ValidationError, validate_call
+from pydantic import TypeAdapter, ValidationError
 from simcore_service_api_server._service_functions import FunctionService
 from simcore_service_api_server.services_rpc.storage import StorageService
 
@@ -210,62 +208,51 @@ async def pre_register_function_jobs(
             for job, input_ in zip(jobs, job_inputs)
         ]
 
-    @validate_call
-    async def patch_registered_function_job(
+    async def batch_patch_registered_function_job(
         self,
         *,
         user_id: UserID,
         product_name: ProductName,
-        patches: Annotated[
-            list[FunctionJobPatch],
-            Field(max_length=50, min_length=1),
-        ],
-    ) -> list[RegisteredFunctionJob]:
-        patch_inputs: list[
-            RegisteredProjectFunctionJobPatchInput
-            | RegisteredSolverFunctionJobPatchInput
-        ] = []
-        for patch in patches:
+        function_job_patches: list[FunctionJobPatch],
+    ) -> BatchUpdateRegisteredFunctionJobs:
+        patch_inputs: FunctionJobPatchRequestList = []
+        for patch in function_job_patches:
             if patch.function_class == FunctionClass.PROJECT:
                 patch_inputs.append(
-                    RegisteredProjectFunctionJobPatchInput(
+                    FunctionJobPatchRequest(
                         uid=patch.function_job_id,
                         patch=RegisteredProjectFunctionJobPatch(
                             title=None,
                             description=None,
                             inputs=None,
                             outputs=None,
                             job_creation_task_id=patch.job_creation_task_id,
-                            project_job_id=patch.project_job_id,
+                            project_job_id=None,
                         ),
                     )
                 )
             elif patch.function_class == FunctionClass.SOLVER:
                 patch_inputs.append(
-                    RegisteredSolverFunctionJobPatchInput(
+                    FunctionJobPatchRequest(
                         uid=patch.function_job_id,
                         patch=RegisteredSolverFunctionJobPatch(
                             title=None,
                             description=None,
                             inputs=None,
                             outputs=None,
                             job_creation_task_id=patch.job_creation_task_id,
-                            solver_job_id=patch.solver_job_id,
+                            solver_job_id=None,
                         ),
                     )
                 )
             else:
                 raise UnsupportedFunctionClassError(
                     function_class=patch.function_class,
                 )
-
-        return await self._web_rpc_client.patch_registered_function_job(
+        return await self._web_rpc_client.batch_patch_registered_function_job(
             user_id=user_id,
             product_name=product_name,
-            registered_function_job_patch_inputs=TypeAdapter(
-                RegisteredProjectFunctionJobPatchInputList
-                | RegisteredSolverFunctionJobPatchInputList
-            ).validate_python(patch_inputs),
+            function_job_patch_requests=patch_inputs,
         )
 
     async def run_function(
@@ -294,20 +281,22 @@ async def run_function(
                 job_id=study_job.id,
                 pricing_spec=pricing_spec,
             )
-            registered_jobs = await self.patch_registered_function_job(
+            registered_job = await self._web_rpc_client.patch_registered_function_job(
                 user_id=self.user_id,
                 product_name=self.product_name,
-                patches=[
-                    FunctionJobPatch(
-                        function_class=FunctionClass.PROJECT,
-                        function_job_id=pre_registered_function_job_data.function_job_id,
+                function_job_patch_request=FunctionJobPatchRequest(
+                    uid=pre_registered_function_job_data.function_job_id,
+                    patch=RegisteredProjectFunctionJobPatch(
+                        title=None,
+                        description=None,
+                        inputs=None,
+                        outputs=None,
                         job_creation_task_id=None,
                         project_job_id=study_job.id,
-                    )
-                ],
+                    ),
+                ),
             )
-            assert len(registered_jobs) == 1
-            return registered_jobs[0]
+            return registered_job
 
         if function.function_class == FunctionClass.SOLVER:
             solver_job = await self._job_service.create_solver_job(
@@ -325,20 +314,22 @@ async def run_function(
                 job_id=solver_job.id,
                 pricing_spec=pricing_spec,
             )
-            registered_jobs = await self.patch_registered_function_job(
+            registered_job = await self._web_rpc_client.patch_registered_function_job(
                 user_id=self.user_id,
                 product_name=self.product_name,
-                patches=[
-                    FunctionJobPatch(
-                        function_class=FunctionClass.SOLVER,
-                        function_job_id=pre_registered_function_job_data.function_job_id,
+                function_job_patch_request=FunctionJobPatchRequest(
+                    uid=pre_registered_function_job_data.function_job_id,
+                    patch=RegisteredSolverFunctionJobPatch(
+                        title=None,
+                        description=None,
+                        inputs=None,
+                        outputs=None,
                         job_creation_task_id=None,
                         solver_job_id=solver_job.id,
-                    )
-                ],
+                    ),
+                ),
             )
-            assert len(registered_jobs) == 1
-            return registered_jobs[0]
+            return registered_job
 
         raise UnsupportedFunctionClassError(
             function_class=function.function_class,

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

@@ -371,10 +371,10 @@ async def create_function_job_creation_tasks(
             for pre_registered_function_job_data in pre_registered_function_job_data_list
         ]
 
-        patched_jobs = await self._function_job_service.patch_registered_function_job(
+        patched_jobs = await self._function_job_service.batch_patch_registered_function_job(
             user_id=user_identity.user_id,
             product_name=user_identity.product_name,
-            patches=[
+            function_job_patches=[
                 FunctionJobPatch(
                     function_class=function.function_class,
                     function_job_id=pre_registered_function_job_data.function_job_id,
@@ -387,6 +387,6 @@ async def create_function_job_creation_tasks(
                 )
             ],
         )
-        patched_jobs_iter = iter(patched_jobs)
+        patched_jobs_iter = iter(patched_jobs.updated_items)
         _ = lambda job: job if job is not None else next(patched_jobs_iter)
         return [_(job) for job in cached_jobs]

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

@@ -24,16 +24,17 @@
 from models_library.api_schemas_webserver.licensed_items import LicensedItemRpcGetPage
 from models_library.functions import (
     BatchCreateRegisteredFunctionJobs,
+    BatchUpdateRegisteredFunctionJobs,
     FunctionInputsList,
     FunctionJob,
     FunctionJobList,
+    FunctionJobPatchRequest,
+    FunctionJobPatchRequestList,
     FunctionJobStatus,
     FunctionOutputs,
     FunctionUserAccessRights,
     FunctionUserApiAccessRights,
     RegisteredFunctionJobWithStatus,
-    RegisteredProjectFunctionJobPatchInputList,
-    RegisteredSolverFunctionJobPatchInputList,
 )
 from models_library.licenses import LicensedItemID
 from models_library.products import ProductName
@@ -515,15 +516,25 @@ async def patch_registered_function_job(
         *,
         user_id: UserID,
         product_name: ProductName,
-        registered_function_job_patch_inputs: (
-            RegisteredProjectFunctionJobPatchInputList
-            | RegisteredSolverFunctionJobPatchInputList
-        ),
-    ) -> list[RegisteredFunctionJob]:
+        function_job_patch_request: FunctionJobPatchRequest,
+    ) -> RegisteredFunctionJob:
         return await self._rpc_client.functions.patch_registered_function_job(
             user_id=user_id,
             product_name=product_name,
-            registered_function_job_patch_inputs=registered_function_job_patch_inputs,
+            function_job_patch_request=function_job_patch_request,
+        )
+
+    async def batch_patch_registered_function_job(
+        self,
+        *,
+        product_name: ProductName,
+        user_id: UserID,
+        function_job_patch_requests: FunctionJobPatchRequestList,
+    ) -> BatchUpdateRegisteredFunctionJobs:
+        return await self._rpc_client.functions.batch_patch_registered_function_job(
+            product_name=product_name,
+            user_id=user_id,
+            function_job_patch_requests=function_job_patch_requests,
         )
 
     async def get_function_input_schema(