added required fields

Author: Andrei Neagu

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/generic_scheduler/_operation.py

@@ -1,8 +1,10 @@
 from abc import ABC, abstractmethod
+from datetime import timedelta
 from typing import Annotated, ClassVar, Final, TypeAlias, TypedDict
 
 from fastapi import FastAPI
 from pydantic import Field, NonNegativeInt, TypeAdapter, validate_call
+from servicelib.deferred_tasks import DeferredContext
 
 from ._errors import (
     OperationAlreadyRegisteredError,
@@ -17,13 +19,31 @@ class BaseStep(ABC):
     def get_step_name(cls) -> StepName:
         return cls.__name__
 
+    ### CREATE
+
     @classmethod
     @abstractmethod
     async def create(cls, app: FastAPI) -> None:
         """
         [mandatory] handler to be implemented with the code resposible for achieving a goal
+        NOTE: Ensure this is successful if:
+            - `create` is called multiple times and does not cause duplicate resources
         """
 
+    @classmethod
+    async def get_create_retries(cls, context: DeferredContext) -> int:
+        """[optional] amount of retires in case of creation"""
+        assert context  # nosec
+        return 0
+
+    @classmethod
+    async def get_create_timeout(cls, context: DeferredContext) -> timedelta:
+        """[optional] timeout between retires case of creation"""
+        assert context  # nosec
+        return timedelta(seconds=0)
+
+    ### DESTROY
+
     @classmethod
     async def destroy(cls, app: FastAPI) -> None:
         """
@@ -35,6 +55,18 @@ async def destroy(cls, app: FastAPI) -> None:
         """
         _ = app
 
+    @classmethod
+    async def get_destroy_retries(cls, context: DeferredContext) -> int:
+        """[optional] amount of retires in case of failure"""
+        assert context  # nosec
+        return 0
+
+    @classmethod
+    async def get_destroy_timeout(cls, context: DeferredContext) -> timedelta:
+        """[optional] timeout between retires in case of failure"""
+        assert context  # nosec
+        return timedelta(seconds=0)
+
 
 StepsSubGroup: TypeAlias = Annotated[tuple[type[BaseStep], ...], Field(min_length=1)]
 

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/generic_scheduler/_store.py

@@ -175,9 +175,10 @@ async def delete(self, *keys: _DeleteScheduleDataKeys) -> None:
 class _StepDict(TypedDict):
     status: NotRequired[StepStatus]
     deferred_task_uid: NotRequired[TaskUID]
+    error_traceback: NotRequired[str]
 
 
-_DeleteStepKeys = Literal["status", "deferred_task_uid"]
+_DeleteStepKeys = Literal["status", "deferred_task_uid", "error_traceback"]
 
 
 class StepStoreProxy:
@@ -211,6 +212,8 @@ def _get_hash_key(self) -> str:
     async def get(self, key: Literal["status"]) -> StepStatus: ...
     @overload
     async def get(self, key: Literal["deferred_task_uid"]) -> TaskUID: ...
+    @overload
+    async def get(self, key: Literal["error_traceback"]) -> str: ...
     async def get(self, key: str) -> Any:
         """raises KeyNotFoundInHashError if the key is not present in the hash"""
         hash_key = self._get_hash_key()
@@ -225,6 +228,8 @@ async def get(self, key: str) -> Any:
     async def set(self, key: Literal["status"], value: StepStatus) -> None: ...
     @overload
     async def set(self, key: Literal["deferred_task_uid"], value: TaskUID) -> None: ...
+    @overload
+    async def set(self, key: Literal["error_traceback"], value: str) -> None: ...
     async def set(self, key: str, value: Any) -> None:
         await self._store.set(self._get_hash_key(), key, value)
 

services/dynamic-scheduler/tests/unit/service_generic_scheduler/test__store.py

@@ -163,17 +163,16 @@ async def test_schedule_data_store_proxy_workflow(
 async def test_step_store_proxy_workflow(
     store: Store, schedule_id: ScheduleId, is_creating: bool
 ):
-    step_name = "MyStep"
     proxy = StepStoreProxy(
         store=store,
         schedule_id=schedule_id,
         operation_name="op1",
         step_group_name="sg1",
-        step_name=step_name,
+        step_name="step",
         is_creating=is_creating,
     )
     is_creating_str = "C" if is_creating else "D"
-    hash_key = f"SCH:{schedule_id}:STEPS:op1:sg1:{is_creating_str}:{step_name}"
+    hash_key = f"SCH:{schedule_id}:STEPS:op1:sg1:{is_creating_str}:step"
 
     # set
     await proxy.set("status", StepStatus.RUNNING)
@@ -190,12 +189,18 @@ async def test_step_store_proxy_workflow(
 
     # set multiple
     await proxy.set_multiple(
-        {"status": StepStatus.SUCCESS, "deferred_task_uid": TaskUID("mytask")}
+        {
+            "status": StepStatus.SUCCESS,
+            "deferred_task_uid": TaskUID("mytask"),
+            "error_traceback": "mock_traceback",
+        }
     )
     await _assert_keys(store, {hash_key})
-    await _assert_keys_in_hash(store, hash_key, {"status", "deferred_task_uid"})
+    await _assert_keys_in_hash(
+        store, hash_key, {"status", "deferred_task_uid", "error_traceback"}
+    )
 
     # remove all keys an even missing ones
-    await proxy.delete("status", "deferred_task_uid")
+    await proxy.delete("status", "deferred_task_uid", "error_traceback")
     await _assert_keys(store, set())
     await _assert_keys_in_hash(store, hash_key, set())