Merge branch 'master' into is8159/fix-redis-client-lifecycle

Author: giancarloromeo

packages/dask-task-models-library/src/dask_task_models_library/plugins/task_life_cycle_worker_plugin.py

@@ -34,6 +34,7 @@ def transition(
     ):
         _logger.info("Task '%s' transition from %s to %s", key, start, finish)
         assert self._worker  # nosec
+        assert isinstance(self._worker, Worker)  # nosec
         self._worker.log_event(
             TASK_LIFE_CYCLE_EVENT.format(key=key),
             TaskLifeCycleState.from_worker_task_state(

packages/pytest-simcore/src/pytest_simcore/pydantic_models.py

@@ -97,7 +97,6 @@ def _is_model_cls(obj) -> bool:
     assert inspect.ismodule(module)
 
     for model_name, model_cls in inspect.getmembers(module, _is_model_cls):
-
         yield from iter_model_examples_in_class(model_cls, model_name)
 
 
@@ -172,7 +171,7 @@ def model_cls_examples(model_cls: type[BaseModel]) -> dict[str, dict[str, Any]]:
     """
     warnings.warn(
         "The 'model_cls_examples' fixture is deprecated and will be removed in a future version. "
-        "Please use 'iter_model_example_in_class' or 'iter_model_examples_in_module' as an alternative.",
+        "Please use 'iter_model_examples_in_class' or 'iter_model_examples_in_module' as an alternative.",
         DeprecationWarning,
         stacklevel=2,
     )

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

@@ -1,4 +1,5 @@
 import logging
+from typing import Literal
 
 from models_library.api_schemas_webserver import WEBSERVER_RPC_NAMESPACE
 from models_library.api_schemas_webserver.functions import (
@@ -18,6 +19,7 @@
 )
 from models_library.functions import (
     FunctionClass,
+    FunctionGroupAccessRights,
     FunctionJobStatus,
     FunctionOutputs,
     FunctionUserAccessRights,
@@ -418,6 +420,7 @@ async def update_function_job_status(
     product_name: ProductName,
     function_job_id: FunctionJobID,
     job_status: FunctionJobStatus,
+    check_write_permissions: bool = True,
 ) -> FunctionJobStatus:
     result = await rabbitmq_rpc_client.request(
         WEBSERVER_RPC_NAMESPACE,
@@ -426,6 +429,7 @@ async def update_function_job_status(
         job_status=job_status,
         user_id=user_id,
         product_name=product_name,
+        check_write_permissions=check_write_permissions,
     )
     return TypeAdapter(FunctionJobStatus).validate_python(result)
 
@@ -438,6 +442,7 @@ async def update_function_job_outputs(
     product_name: ProductName,
     function_job_id: FunctionJobID,
     outputs: FunctionOutputs,
+    check_write_permissions: bool = True,
 ) -> FunctionOutputs:
     result = await rabbitmq_rpc_client.request(
         WEBSERVER_RPC_NAMESPACE,
@@ -446,6 +451,7 @@ async def update_function_job_outputs(
         outputs=outputs,
         user_id=user_id,
         product_name=product_name,
+        check_write_permissions=check_write_permissions,
     )
     return TypeAdapter(FunctionOutputs).validate_python(result)
 
@@ -578,3 +584,37 @@ async def get_functions_user_api_access_rights(
         product_name=product_name,
     )
     return TypeAdapter(FunctionUserApiAccessRights).validate_python(result)
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def set_group_permissions(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    user_id: UserID,
+    product_name: ProductName,
+    object_type: Literal["function", "function_job", "function_job_collection"],
+    object_ids: list[FunctionID | FunctionJobID | FunctionJobCollectionID],
+    permission_group_id: int,
+    read: bool | None = None,
+    write: bool | None = None,
+    execute: bool | None = None,
+) -> list[
+    tuple[
+        FunctionID | FunctionJobID | FunctionJobCollectionID, FunctionGroupAccessRights
+    ]
+]:
+    result = await rabbitmq_rpc_client.request(
+        WEBSERVER_RPC_NAMESPACE,
+        TypeAdapter(RPCMethodName).validate_python("set_group_permissions"),
+        user_id=user_id,
+        product_name=product_name,
+        object_type=object_type,
+        object_ids=object_ids,
+        permission_group_id=permission_group_id,
+        read=read,
+        write=write,
+        execute=execute,
+    )
+    return TypeAdapter(
+        list[tuple[FunctionID | FunctionJobID, FunctionGroupAccessRights]]
+    ).validate_python(result)

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

@@ -203,6 +203,7 @@ async def inspect_function_job(
             user_id=self.user_id,
             product_name=self.product_name,
             job_status=new_job_status,
+            check_write_permissions=False,
         )
 
     async def create_function_job_inputs(  # pylint: disable=no-self-use
@@ -529,4 +530,5 @@ async def function_job_outputs(
             user_id=user_id,
             product_name=product_name,
             outputs=new_outputs,
+            check_write_permissions=False,
         )

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

@@ -474,7 +474,7 @@ async def _run_single_function(function_inputs: FunctionInputs) -> FunctionJobID
             for function_inputs in function_inputs_list
         ],
         reraise=False,
-        limit=10,
+        limit=1,
     )
 
     # Check if any tasks raised exceptions and raise the first one found

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

@@ -587,13 +587,15 @@ async def update_function_job_status(
         user_id: UserID,
         product_name: ProductName,
         job_status: FunctionJobStatus,
+        check_write_permissions: bool = True,
     ) -> FunctionJobStatus:
         return await functions_rpc_interface.update_function_job_status(
             self._client,
             function_job_id=function_job_id,
             user_id=user_id,
             product_name=product_name,
             job_status=job_status,
+            check_write_permissions=check_write_permissions,
         )
 
     async def update_function_job_outputs(
@@ -603,13 +605,15 @@ async def update_function_job_outputs(
         user_id: UserID,
         product_name: ProductName,
         outputs: FunctionOutputs,
+        check_write_permissions: bool = True,
     ) -> FunctionOutputs:
         return await functions_rpc_interface.update_function_job_outputs(
             self._client,
             function_job_id=function_job_id,
             user_id=user_id,
             product_name=product_name,
             outputs=outputs,
+            check_write_permissions=check_write_permissions,
         )
 
     async def find_cached_function_jobs(

services/director-v2/src/simcore_service_director_v2/core/settings.py

@@ -99,6 +99,14 @@ class ComputationalBackendSettings(BaseCustomSettings):
         ),
     ] = datetime.timedelta(minutes=10)
 
+    COMPUTATIONAL_BACKEND_MAX_WAITING_FOR_RETRIEVING_RESULTS: Annotated[
+        datetime.timedelta,
+        Field(
+            description="maximum time the computational scheduler waits until retrieving results from the computational backend is failed"
+            "(default to seconds, or see https://pydantic-docs.helpmanual.io/usage/types/#datetime-types for string formatting)."
+        ),
+    ] = datetime.timedelta(minutes=10)
+
     @cached_property
     def default_cluster(self) -> BaseCluster:
         return BaseCluster(

services/director-v2/src/simcore_service_director_v2/models/comp_run_snapshot_tasks.py

@@ -1,12 +1,21 @@
+from contextlib import suppress
 from datetime import datetime
 from typing import Annotated, Any
 
 from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
 from models_library.projects_state import RunningState
 from models_library.resource_tracker import HardwareInfo
-from pydantic import BaseModel, BeforeValidator, ConfigDict, PositiveInt
+from pydantic import (
+    BaseModel,
+    BeforeValidator,
+    ConfigDict,
+    PositiveInt,
+    field_validator,
+)
+from simcore_postgres_database.models.comp_pipeline import StateType
 
+from ..utils.db import DB_TO_RUNNING_STATE
 from .comp_tasks import BaseCompTaskAtDB, Image
 
 
@@ -100,3 +109,15 @@ class CompRunSnapshotTaskDBGet(BaseModel):
     started_at: datetime | None
     ended_at: datetime | None
     iteration: PositiveInt
+
+    @field_validator("state", mode="before")
+    @classmethod
+    def convert_result_from_state_type_enum_if_needed(cls, v):
+        if isinstance(v, str):
+            # try to convert to a StateType, if it fails the validations will continue
+            # and pydantic will try to convert it to a RunninState later on
+            with suppress(ValueError):
+                v = StateType(v)
+        if isinstance(v, StateType):
+            return RunningState(DB_TO_RUNNING_STATE[StateType(v)])
+        return v