Merge remote-tracking branch 'upstream/master' into pr-osparc-overwite-can-save-templates

Author: Andrei Neagu

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

@@ -85,3 +85,37 @@ class BatchGetEnvelope(BaseModel, Generic[ResourceT, IdentifierT]):
             description="List of identifiers for items that were not found",
         ),
     ] = DEFAULT_FACTORY
+
+
+class BatchCreateEnvelope(BaseModel, Generic[SchemaT]):
+    """Generic envelope model for batch-create operations.
+
+    This model represents the result of a strict batch create operation,
+    containing the list of created items. The operation is expected to be "strict"
+    in the sense that it either creates all requested items or fails entirely.
+    """
+
+    created_items: Annotated[
+        list[SchemaT],
+        Field(
+            min_length=1,
+            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

@@ -1,19 +1,21 @@
 import datetime
 from collections.abc import Mapping
 from enum import Enum
-from typing import Annotated, Any, Literal, TypeAlias
+from typing import Annotated, Any, Final, Literal, TypeAlias
 from uuid import UUID
 
 from models_library import projects
 from models_library.basic_regex import UUID_RE_BASE
 from models_library.basic_types import ConstrainedStr
+from models_library.batch_operations import BatchCreateEnvelope
 from models_library.groups import GroupID
 from models_library.products import ProductName
 from models_library.services_types import ServiceKey, ServiceVersion
 from models_library.users import UserID
 from models_library.utils.enums import StrAutoEnum
 from pydantic import BaseModel, ConfigDict, Field
 
+from .batch_operations import BatchGetEnvelope, BatchUpdateEnvelope
 from .projects import ProjectID
 from .utils.change_case import snake_to_camel
 
@@ -23,6 +25,7 @@
 FileID: TypeAlias = UUID
 
 InputTypes: TypeAlias = FileID | float | int | bool | str | list
+_MAX_LIST_LENGTH: Final[int] = 50
 
 
 class FunctionSchemaClass(str, Enum):
@@ -80,9 +83,10 @@ class FunctionClass(str, Enum):
 
 FunctionInputsList: TypeAlias = Annotated[
     list[FunctionInputs],
-    Field(max_length=50),
+    Field(max_length=_MAX_LIST_LENGTH),
 ]
 
+
 FunctionOutputs: TypeAlias = dict[str, Any] | None
 
 FunctionOutputsLogfile: TypeAlias = Any
@@ -238,6 +242,9 @@ class RegisteredPythonCodeFunctionJobPatch(BaseModel):
     ProjectFunctionJob | PythonCodeFunctionJob | SolverFunctionJob,
     Field(discriminator="function_class"),
 ]
+FunctionJobList: TypeAlias = Annotated[
+    list[FunctionJob], Field(max_length=_MAX_LIST_LENGTH)
+]
 
 
 class RegisteredFunctionJobBase(FunctionJobBase):
@@ -264,6 +271,21 @@ class RegisteredPythonCodeFunctionJob(PythonCodeFunctionJob, RegisteredFunctionJ
     Field(discriminator="function_class"),
 ]
 
+
+class BatchCreateRegisteredFunctionJobs(BatchCreateEnvelope[RegisteredFunctionJob]):
+    pass
+
+
+class BatchUpdateRegisteredFunctionJobs(BatchUpdateEnvelope[RegisteredFunctionJob]):
+    pass
+
+
+class BatchGetCachedRegisteredFunctionJobs(
+    BatchGetEnvelope[RegisteredFunctionJob, FunctionInputs]
+):
+    pass
+
+
 RegisteredFunctionJobPatch = Annotated[
     RegisteredProjectFunctionJobPatch
     | RegisteredPythonCodeFunctionJobPatch
@@ -272,6 +294,20 @@ class RegisteredPythonCodeFunctionJob(PythonCodeFunctionJob, RegisteredFunctionJ
 ]
 
 
+class FunctionJobPatchRequest(BaseModel):
+    uid: FunctionJobID
+    patch: RegisteredFunctionJobPatch
+
+
+FunctionJobPatchRequestList: TypeAlias = Annotated[
+    list[FunctionJobPatchRequest],
+    Field(
+        max_length=_MAX_LIST_LENGTH,
+        description="List of function job patch requests",
+    ),
+]
+
+
 class FunctionJobStatus(BaseModel):
     status: str
 
@@ -340,6 +376,20 @@ class RegisteredFunctionJobDB(FunctionJobDB):
     created: datetime.datetime
 
 
+class BatchGetCachedRegisteredFunctionJobsDB(
+    BatchGetEnvelope[RegisteredFunctionJobDB, FunctionInputs]
+):
+    pass
+
+
+class BatchCreateRegisteredFunctionJobsDB(BatchCreateEnvelope[RegisteredFunctionJobDB]):
+    pass
+
+
+class BatchUpdateRegisteredFunctionJobsDB(BatchUpdateEnvelope[RegisteredFunctionJobDB]):
+    pass
+
+
 class RegisteredFunctionJobWithStatusDB(FunctionJobDB):
     uuid: FunctionJobID
     created: datetime.datetime

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

@@ -170,3 +170,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 FunctionUnrecoverableError(FunctionBaseError):
+    msg_template = "Unrecoverable error."
+    status_code: int = 500

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

@@ -21,11 +21,11 @@
 from models_library.functions import (
     FunctionClass,
     FunctionGroupAccessRights,
+    FunctionInputsList,
     FunctionJobStatus,
     FunctionOutputs,
     FunctionUserAccessRights,
     FunctionUserApiAccessRights,
-    RegisteredFunctionJobPatch,
     RegisteredFunctionJobWithStatus,
 )
 from models_library.products import ProductName
@@ -359,29 +359,6 @@ async def register_function_job(
     )  # Validates the result as a RegisteredFunctionJob
 
 
-@log_decorator(_logger, level=logging.DEBUG)
-async def patch_registered_function_job(
-    rabbitmq_rpc_client: RabbitMQRPCClient,
-    *,
-    user_id: UserID,
-    product_name: ProductName,
-    function_job_uuid: FunctionJobID,
-    registered_function_job_patch: RegisteredFunctionJobPatch,
-) -> RegisteredFunctionJob:
-    result = await rabbitmq_rpc_client.request(
-        DEFAULT_WEBSERVER_RPC_NAMESPACE,
-        TypeAdapter(RPCMethodName).validate_python("patch_registered_function_job"),
-        user_id=user_id,
-        product_name=product_name,
-        function_job_uuid=function_job_uuid,
-        registered_function_job_patch=registered_function_job_patch,
-        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
-    )
-    return TypeAdapter(RegisteredFunctionJob).validate_python(
-        result
-    )  # Validates the result as a RegisteredFunctionJob
-
-
 @log_decorator(_logger, level=logging.DEBUG)
 async def get_function_job(
     rabbitmq_rpc_client: RabbitMQRPCClient,
@@ -512,20 +489,20 @@ async def find_cached_function_jobs(
     user_id: UserID,
     product_name: ProductName,
     function_id: FunctionID,
-    inputs: FunctionInputs,
-) -> list[RegisteredFunctionJob] | None:
+    inputs: FunctionInputsList,
+    status_filter: list[FunctionJobStatus] | None = None,
+) -> list[RegisteredFunctionJob | None]:
     result = await rabbitmq_rpc_client.request(
         DEFAULT_WEBSERVER_RPC_NAMESPACE,
         TypeAdapter(RPCMethodName).validate_python("find_cached_function_jobs"),
         function_id=function_id,
         inputs=inputs,
+        status_filter=status_filter,
         user_id=user_id,
         product_name=product_name,
         timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
-    if result is None:
-        return None
-    return TypeAdapter(list[RegisteredFunctionJob]).validate_python(result)
+    return TypeAdapter(list[RegisteredFunctionJob | None]).validate_python(result)
 
 
 @log_decorator(_logger, level=logging.DEBUG)

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

@@ -7,7 +7,6 @@
     FunctionID,
     FunctionInputs,
     FunctionInputSchema,
-    FunctionJob,
     FunctionJobCollection,
     FunctionJobCollectionID,
     FunctionJobCollectionsListFilters,
@@ -18,13 +17,19 @@
     RegisteredFunctionJobCollection,
 )
 from models_library.functions import (
+    BatchCreateRegisteredFunctionJobs,
+    BatchUpdateRegisteredFunctionJobs,
     FunctionClass,
     FunctionGroupAccessRights,
+    FunctionInputsList,
+    FunctionJob,
+    FunctionJobList,
+    FunctionJobPatchRequest,
+    FunctionJobPatchRequestList,
     FunctionJobStatus,
     FunctionOutputs,
     FunctionUserAccessRights,
     FunctionUserApiAccessRights,
-    RegisteredFunctionJobPatch,
     RegisteredFunctionJobWithStatus,
 )
 from models_library.products import ProductName
@@ -329,22 +334,54 @@ async def register_function_job(
             ),
         )
 
+    async def batch_register_function_jobs(
+        self,
+        *,
+        product_name: ProductName,
+        user_id: UserID,
+        function_jobs: FunctionJobList,
+    ) -> BatchCreateRegisteredFunctionJobs:
+        """Register a function job."""
+        return TypeAdapter(BatchCreateRegisteredFunctionJobs).validate_python(
+            await self._request(
+                "batch_register_function_jobs",
+                product_name=product_name,
+                user_id=user_id,
+                function_jobs=function_jobs,
+            ),
+        )
+
     async def patch_registered_function_job(
         self,
         *,
         product_name: ProductName,
         user_id: UserID,
-        function_job_uuid: FunctionJobID,
-        registered_function_job_patch: RegisteredFunctionJobPatch,
+        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_uuid=function_job_uuid,
-                registered_function_job_patch=registered_function_job_patch,
+                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 BatchUpdateRegisteredFunctionJobs.model_validate(
+            await self._request(
+                "batch_patch_registered_function_jobs",
+                product_name=product_name,
+                user_id=user_id,
+                function_job_patch_requests=function_job_patch_requests,
             ),
         )
 
@@ -462,16 +499,18 @@ async def find_cached_function_jobs(
         product_name: ProductName,
         user_id: UserID,
         function_id: FunctionID,
-        inputs: FunctionInputs,
-    ) -> list[RegisteredFunctionJob] | None:
+        inputs: FunctionInputsList,
+        cached_job_statuses: list[FunctionJobStatus] | None = None,
+    ) -> list[RegisteredFunctionJob | None]:
         """Find cached function jobs."""
-        return TypeAdapter(list[RegisteredFunctionJob] | None).validate_python(
+        return TypeAdapter(list[RegisteredFunctionJob | None]).validate_python(
             await self._request(
                 "find_cached_function_jobs",
                 product_name=product_name,
                 user_id=user_id,
                 function_id=function_id,
                 inputs=inputs,
+                cached_job_statuses=cached_job_statuses,
             ),
         )