refactor patch method

Author: bisgaard-itis

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

@@ -102,3 +102,20 @@ class BatchCreateEnvelope(BaseModel, Generic[SchemaT]):
             description="List of successfully created items",
         ),
     ]
+
+
+class BatchUpdateEnvelope(BaseModel, Generic[SchemaT]):
+    """Generic envelope model for batch-update operations.
+
+    This model represents the result of a strict batch update operation,
+    containing the list of updated items. The operation is expected to be "strict"
+    in the sense that it either updates all requested items or fails entirely. See https://google.aip.dev/234
+    """
+
+    updated_items: Annotated[
+        list[SchemaT],
+        Field(
+            min_length=1,
+            description="List of successfully updated items",
+        ),
+    ]

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

@@ -15,6 +15,7 @@
 from models_library.utils.enums import StrAutoEnum
 from pydantic import BaseModel, ConfigDict, Field
 
+from .batch_operations import BatchUpdateEnvelope
 from .projects import ProjectID
 from .utils.change_case import snake_to_camel
 
@@ -273,8 +274,10 @@ class RegisteredPythonCodeFunctionJob(PythonCodeFunctionJob, RegisteredFunctionJ
 
 
 class BatchCreateRegisteredFunctionJobs(BatchCreateEnvelope[RegisteredFunctionJob]):
-    """Envelope model for batch registering function jobs"""
+    pass
+
 
+class BatchUpdateRegisteredFunctionJobs(BatchUpdateEnvelope[RegisteredFunctionJob]):
     pass
 
 
@@ -291,6 +294,16 @@ class FunctionJobPatchRequest(BaseModel):
     patch: RegisteredFunctionJobPatch
 
 
+FunctionJobPatchRequestList: TypeAlias = Annotated[
+    list[FunctionJobPatchRequest],
+    Field(
+        max_length=_MAX_LIST_LENGTH,
+        min_length=_MIN_LIST_LENGTH,
+        description="List of function job patch requests",
+    ),
+]
+
+
 class FunctionJobStatus(BaseModel):
     status: str
 
@@ -363,6 +376,10 @@ class BatchCreateRegisteredFunctionJobsDB(BatchCreateEnvelope[RegisteredFunction
     pass
 
 
+class BatchUpdateRegisteredFunctionJobsDB(BatchUpdateEnvelope[RegisteredFunctionJobDB]):
+    pass
+
+
 class RegisteredFunctionJobWithStatusDB(FunctionJobDB):
     uuid: FunctionJobID
     created: datetime.datetime

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/webserver/v1/functions.py

@@ -18,18 +18,19 @@
 )
 from models_library.functions import (
     BatchCreateRegisteredFunctionJobs,
+    BatchUpdateRegisteredFunctionJobs,
     FunctionClass,
     FunctionGroupAccessRights,
     FunctionInputsList,
     FunctionJob,
     FunctionJobList,
+    FunctionJobPatchRequest,
+    FunctionJobPatchRequestList,
     FunctionJobStatus,
     FunctionOutputs,
     FunctionUserAccessRights,
     FunctionUserApiAccessRights,
     RegisteredFunctionJobWithStatus,
-    RegisteredProjectFunctionJobPatchInputList,
-    RegisteredSolverFunctionJobPatchInputList,
 )
 from models_library.products import ProductName
 from models_library.rabbitmq_basic_types import RPCNamespace
@@ -355,18 +356,32 @@ async def patch_registered_function_job(
         *,
         product_name: ProductName,
         user_id: UserID,
-        registered_function_job_patch_inputs: (
-            RegisteredProjectFunctionJobPatchInputList
-            | RegisteredSolverFunctionJobPatchInputList
-        ),
-    ) -> list[RegisteredFunctionJob]:
+        function_job_patch_request: FunctionJobPatchRequest,
+    ) -> RegisteredFunctionJob:
+        """Patch a registered function job."""
+        return TypeAdapter(RegisteredFunctionJob).validate_python(
+            await self._request(
+                "patch_registered_function_job",
+                product_name=product_name,
+                user_id=user_id,
+                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:
         """Patch a registered function job."""
-        return TypeAdapter(list[RegisteredFunctionJob]).validate_python(
+        return BatchUpdateRegisteredFunctionJobs.model_validate(
             await self._request(
                 "patch_registered_function_job",
                 product_name=product_name,
                 user_id=user_id,
-                registered_function_job_patch_inputs=registered_function_job_patch_inputs,
+                function_job_patch_requests=function_job_patch_requests,
             ),
         )
 

services/web/server/src/simcore_service_webserver/functions/_controller/_functions_rpc.py

@@ -3,6 +3,7 @@
 from aiohttp import web
 from models_library.functions import (
     BatchCreateRegisteredFunctionJobs,
+    BatchUpdateRegisteredFunctionJobs,
     Function,
     FunctionAccessRights,
     FunctionClass,
@@ -16,6 +17,8 @@
     FunctionJobCollectionsListFilters,
     FunctionJobID,
     FunctionJobList,
+    FunctionJobPatchRequest,
+    FunctionJobPatchRequestList,
     FunctionJobStatus,
     FunctionOutputs,
     FunctionOutputSchema,
@@ -25,8 +28,6 @@
     RegisteredFunctionJob,
     RegisteredFunctionJobCollection,
     RegisteredFunctionJobWithStatus,
-    RegisteredProjectFunctionJobPatchInputList,
-    RegisteredSolverFunctionJobPatchInputList,
 )
 from models_library.functions_errors import (
     FunctionIDNotFoundError,
@@ -133,17 +134,37 @@ 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 _functions_service.patch_registered_function_job(
         app=app,
         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,
+    )
+
+
+@router.expose(
+    reraise_if_error_type=(
+        UnsupportedFunctionJobClassError,
+        FunctionJobsWriteApiAccessDeniedError,
+        FunctionJobPatchModelIncompatibleError,
+    )
+)
+async def batch_patch_registered_function_jobs(
+    app: web.Application,
+    *,
+    user_id: UserID,
+    product_name: ProductName,
+    function_job_patch_requests: FunctionJobPatchRequestList,
+) -> BatchUpdateRegisteredFunctionJobs:
+
+    return await _functions_service.batch_patch_registered_function_jobs(
+        app=app,
+        user_id=user_id,
+        product_name=product_name,
+        function_job_patch_requests=function_job_patch_requests,
     )
 
 

services/web/server/src/simcore_service_webserver/functions/_function_jobs_repository.py

@@ -7,21 +7,21 @@
 from aiohttp import web
 from models_library.functions import (
     BatchCreateRegisteredFunctionJobsDB,
+    BatchUpdateRegisteredFunctionJobsDB,
     FunctionClass,
     FunctionClassSpecificData,
     FunctionID,
     FunctionInputsList,
     FunctionJobCollectionID,
     FunctionJobDB,
     FunctionJobID,
+    FunctionJobPatchRequest,
     FunctionJobStatus,
     FunctionOutputs,
     FunctionsApiAccessRights,
     RegisteredFunctionJobDB,
     RegisteredFunctionJobPatch,
     RegisteredFunctionJobWithStatusDB,
-    RegisteredProjectFunctionJobPatchInputList,
-    RegisteredSolverFunctionJobPatchInputList,
 )
 from models_library.functions_errors import (
     FunctionJobIDNotFoundError,
@@ -130,24 +130,14 @@ async def create_function_jobs(  # noqa: PLR0913
     return BatchCreateRegisteredFunctionJobsDB(created_items=created_jobs)
 
 
-async def patch_function_job(
+async def patch_function_jobs(
     app: web.Application,
     connection: AsyncConnection | None = None,
     *,
     user_id: UserID,
     product_name: ProductName,
-    registered_function_job_patch_inputs: (
-        RegisteredProjectFunctionJobPatchInputList
-        | RegisteredSolverFunctionJobPatchInputList
-    ),
-) -> list[RegisteredFunctionJobDB]:
-
-    # check only a single function class is used
-    TypeAdapter(
-        RegisteredProjectFunctionJobPatchInputList
-        | RegisteredSolverFunctionJobPatchInputList
-    ).validate_python(registered_function_job_patch_inputs)
-    used_function_class = registered_function_job_patch_inputs[0].patch.function_class
+    function_job_patch_requests: list[FunctionJobPatchRequest],
+) -> BatchUpdateRegisteredFunctionJobsDB:
 
     async with transaction_context(get_asyncpg_engine(app), connection) as transaction:
         await check_user_api_access_rights(
@@ -160,41 +150,41 @@ async def patch_function_job(
             ],
         )
         updated_jobs = []
-        for patch_input in registered_function_job_patch_inputs:
+        for patch_request in function_job_patch_requests:
 
             job = await get_function_job(
                 app,
                 connection=transaction,
                 user_id=user_id,
                 product_name=product_name,
-                function_job_id=patch_input.uid,
+                function_job_id=patch_request.uid,
             )
-            if job.function_class != used_function_class:
+            if job.function_class != patch_request.patch.function_class:
                 raise FunctionJobPatchModelIncompatibleError(
                     function_id=job.function_uuid, product_name=product_name
                 )
 
             class_specific_data = _update_class_specific_data(
-                class_specific_data=job.class_specific_data, patch=patch_input.patch
+                class_specific_data=job.class_specific_data, patch=patch_request.patch
             )
             update_values = {
-                "inputs": patch_input.patch.inputs,
-                "outputs": patch_input.patch.outputs,
+                "inputs": patch_request.patch.inputs,
+                "outputs": patch_request.patch.outputs,
                 "class_specific_data": class_specific_data,
-                "title": patch_input.patch.title,
-                "description": patch_input.patch.description,
+                "title": patch_request.patch.title,
+                "description": patch_request.patch.description,
                 "status": "created",
             }
 
             result = await transaction.execute(
                 function_jobs_table.update()
-                .where(function_jobs_table.c.uuid == f"{patch_input.uid}")
+                .where(function_jobs_table.c.uuid == f"{patch_request.uid}")
                 .values(**{k: v for k, v in update_values.items() if v is not None})
                 .returning(*_FUNCTION_JOBS_TABLE_COLS)
             )
             updated_jobs.append(RegisteredFunctionJobDB.model_validate(result.one()))
 
-        return updated_jobs
+        return BatchUpdateRegisteredFunctionJobsDB(updated_items=updated_jobs)
 
 
 async def list_function_jobs_with_status(

services/web/server/src/simcore_service_webserver/functions/_functions_service.py

@@ -4,6 +4,7 @@
 from models_library.basic_types import IDStr
 from models_library.functions import (
     BatchCreateRegisteredFunctionJobs,
+    BatchUpdateRegisteredFunctionJobs,
     Function,
     FunctionClass,
     FunctionClassSpecificData,
@@ -20,6 +21,8 @@
     FunctionJobDB,
     FunctionJobID,
     FunctionJobList,
+    FunctionJobPatchRequest,
+    FunctionJobPatchRequestList,
     FunctionJobStatus,
     FunctionOutputs,
     FunctionOutputSchema,
@@ -35,11 +38,9 @@
     RegisteredFunctionJobWithStatusDB,
     RegisteredProjectFunction,
     RegisteredProjectFunctionJob,
-    RegisteredProjectFunctionJobPatchInputList,
     RegisteredProjectFunctionJobWithStatus,
     RegisteredSolverFunction,
     RegisteredSolverFunctionJob,
-    RegisteredSolverFunctionJobPatchInputList,
     RegisteredSolverFunctionJobWithStatus,
 )
 from models_library.functions_errors import (
@@ -134,19 +135,39 @@ 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:
+
+    result = await _function_jobs_repository.patch_function_jobs(
+        app=app,
+        user_id=user_id,
+        product_name=product_name,
+        function_job_patch_requests=[function_job_patch_request],
+    )
+    assert len(result.updated_items) == 1  # nosec
+    return _decode_functionjob(result.updated_items[0])
+
 
-    result = await _function_jobs_repository.patch_function_job(
+async def batch_patch_registered_function_jobs(
+    app: web.Application,
+    *,
+    user_id: UserID,
+    product_name: ProductName,
+    function_job_patch_requests: FunctionJobPatchRequestList,
+) -> BatchUpdateRegisteredFunctionJobs:
+    TypeAdapter(FunctionJobPatchRequestList).validate_python(
+        function_job_patch_requests
+    )
+
+    result = await _function_jobs_repository.patch_function_jobs(
         app=app,
         user_id=user_id,
         product_name=product_name,
-        registered_function_job_patch_inputs=registered_function_job_patch_inputs,
+        function_job_patch_requests=function_job_patch_requests,
+    )
+    return BatchUpdateRegisteredFunctionJobs(
+        updated_items=[_decode_functionjob(job) for job in result.updated_items]
     )
-    return [_decode_functionjob(job) for job in result]
 
 
 async def register_function_job_collection(