refacrtored core modules

Author: Andrei Neagu

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

@@ -0,0 +1,19 @@
+from common_library.errors_classes import OsparcErrorMixin
+
+
+class BaseGenericSchedulerError(OsparcErrorMixin, Exception):
+    """base exception for this module"""
+
+
+class KeyNotFoundInHashError(BaseGenericSchedulerError):
+    msg_template: str = "Key '{key}' not found in hash '{hash_key}'"
+
+
+class OperationAlreadyRegisteredError(BaseGenericSchedulerError):
+    msg_template: str = "Operation '{operation_name}' already registered"
+
+
+class OperationNotFoundError(BaseGenericSchedulerError):
+    msg_template: str = (
+        "Operation '{operation_name}' was not found, registerd_operations='{registerd_operations}'"
+    )

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

@@ -0,0 +1,157 @@
+from abc import ABC, abstractmethod
+from typing import Annotated, ClassVar, Final, TypeAlias
+
+from fastapi import FastAPI
+from pydantic import Field, NonNegativeInt, TypeAdapter, validate_call
+
+from ._errors import OperationAlreadyRegisteredError, OperationNotFoundError
+from ._models import OperationName, StepGroupName, StepName
+
+
+class BaseStep(ABC):
+    @classmethod
+    def get_step_name(cls) -> StepName:
+        return cls.__name__
+
+    @classmethod
+    @abstractmethod
+    async def create(cls, app: FastAPI) -> None:
+        """
+        [mandatory] handler to be implemented with the code resposible for achieving a goal
+        """
+
+    @classmethod
+    async def destroy(cls, app: FastAPI) -> None:
+        """
+        [optional] handler resposible for celanup of resources created above.
+        NOTE: Ensure this is successful if:
+            - `create` is not executed
+            - `create` is executed partially
+            - `destory` is called multiple times
+        """
+        _ = app
+
+
+StepsSubGroup: TypeAlias = Annotated[tuple[type[BaseStep], ...], Field(min_length=1)]
+
+
+class BaseStepGroup(ABC):
+    def __init__(self, *, repeat_steps: bool) -> None:
+        """
+        if repeat_steps is True, the steps in this group will be repeated forever
+        """
+        self.repeat_steps = repeat_steps
+
+    @abstractmethod
+    def get_step_group_name(self, *, index: NonNegativeInt) -> StepGroupName:
+        """returns the name of this step group"""
+
+    @abstractmethod
+    def get_steps_names(self) -> list[StepName]:
+        """return sorted list of StepName"""
+
+    @abstractmethod
+    def get_step_subgroup_to_run(self) -> StepsSubGroup:
+        """returns subgroups of steps to run"""
+
+
+class SingleStepGroup(BaseStepGroup):
+    def __init__(self, step: type[BaseStep], *, repeat_steps: bool = False) -> None:
+        self._step: type[BaseStep] = step
+        super().__init__(repeat_steps=repeat_steps)
+
+    def get_step_group_name(self, *, index: NonNegativeInt) -> StepGroupName:
+        return f"{index}S{'R' if self.repeat_steps else ''}"
+
+    def get_steps_names(self) -> list[StepName]:
+        return [self._step.get_step_name()]
+
+    def get_step_subgroup_to_run(self) -> StepsSubGroup:
+        return TypeAdapter(StepsSubGroup).validate_python((self._step,))
+
+
+_MIN_PARALLEL_STEPS: Final[int] = 2
+
+
+class ParallelStepGroup(BaseStepGroup):
+    def __init__(self, *steps: type[BaseStep], repeat_steps: bool = False) -> None:
+
+        self._steps: list[type[BaseStep]] = list(steps)
+
+        super().__init__(repeat_steps=repeat_steps)
+
+    @property
+    def steps(self) -> list[type[BaseStep]]:
+        return self._steps
+
+    def get_step_group_name(self, *, index: NonNegativeInt) -> StepGroupName:
+        return f"{index}P{'R' if self.repeat_steps else ''}"
+
+    def get_steps_names(self) -> list[StepName]:
+        return sorted(x.get_step_name() for x in self._steps)
+
+    def get_step_subgroup_to_run(self) -> StepsSubGroup:
+        return TypeAdapter(StepsSubGroup).validate_python(tuple(self._steps))
+
+
+Operation: TypeAlias = Annotated[list[BaseStepGroup], Field(min_length=1)]
+
+
+@validate_call(config={"arbitrary_types_allowed": True})
+def _validate_operation(operation: Operation) -> None:
+    detected_steps_names: set[StepName] = set()
+
+    for k, step_group in enumerate(operation):
+        if isinstance(step_group, ParallelStepGroup):
+            if len(step_group.steps) < _MIN_PARALLEL_STEPS:
+                msg = (
+                    f"{ParallelStepGroup.__name__} needs at least {_MIN_PARALLEL_STEPS} "
+                    f"steps. TIP: use {SingleStepGroup.__name__} instead."
+                )
+                raise ValueError(msg)
+
+        if k < len(operation) - 1 and step_group.repeat_steps is True:
+            msg = f"Only the last step group can have repeat_steps=True. Error at index {k=}"
+            raise ValueError(msg)
+
+        for step in step_group.get_step_subgroup_to_run():
+            step_name = step.get_step_name()
+
+            if step_name in detected_steps_names:
+                msg = f"Step {step_name=} is already used in this operation {detected_steps_names=}"
+                raise ValueError(msg)
+
+            detected_steps_names.add(step_name)
+
+
+class OperationRegistry:
+    _OPERATIONS: ClassVar[dict[str, Operation]] = {}
+
+    @classmethod
+    def register(cls, operation_name: OperationName, operation: Operation) -> None:
+        _validate_operation(operation)
+
+        if operation_name in cls._OPERATIONS:
+            raise OperationAlreadyRegisteredError(operation_name=operation_name)
+
+        cls._OPERATIONS[operation_name] = operation
+
+    @classmethod
+    def get(cls, operation_name: OperationName) -> Operation:
+        if operation_name not in cls._OPERATIONS:
+            raise OperationNotFoundError(
+                operation_name=operation_name,
+                registerd_operations=list(cls._OPERATIONS.keys()),
+            )
+
+        return cls._OPERATIONS[operation_name]
+
+    @classmethod
+    def unregister(cls, operation_name: OperationName) -> None:
+        if operation_name not in cls._OPERATIONS:
+            raise OperationNotFoundError(
+                operation_name=operation_name,
+                registerd_operations=list(cls._OPERATIONS.keys()),
+            )
+
+        del cls._OPERATIONS[operation_name]

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

@@ -8,7 +8,14 @@
 from servicelib.redis._utils import handle_redis_returns_union_types
 from settings_library.redis import RedisDatabase, RedisSettings
 
-from ._models import OperationName, ScheduleId, StepGroup, StepName
+from ._errors import KeyNotFoundInHashError
+from ._models import (
+    OperationContext,
+    OperationName,
+    ScheduleId,
+    StepGroupName,
+    StepName,
+)
 
 _SCHEDULE_NAMESPACE: Final[str] = "SCH"
 _STEPS_KEY: Final[str] = "STEPS"
@@ -26,7 +33,7 @@ def _get_step_hash_key(
     *,
     schedule_id: ScheduleId,
     operation_name: OperationName,
-    group: StepGroup,
+    group: StepGroupName,
     step_name: StepName,
 ) -> str:
     # SCHEDULE_NAMESPACE:SCHEDULE_ID:STEPS:OPERATION_NAME:GROUP_INDEX:STEP_NAME:KEY
@@ -102,11 +109,14 @@ async def remove(self, hash_key: str) -> None:
 
 class _UpdateScheduleDataDict(TypedDict):
     operation_name: NotRequired[OperationName]
+    operation_context: NotRequired[OperationContext]
     group_index: NotRequired[NonNegativeInt]
     is_creating: NotRequired[bool]
 
 
-_DeleteScheduleDataKeys = Literal["operation_name", "group_index", "is_creating"]
+_DeleteScheduleDataKeys = Literal[
+    "operation_name", "operation_context", "group_index", "is_creating"
+]
 
 
 class ScheduleDataStoreProxy:
@@ -118,19 +128,33 @@ def _get_hash_key(self) -> str:
         return _get_scheduler_data_hash_key(schedule_id=self._schedule_id)
 
     @overload
-    async def get(self, key: Literal["operation_name"]) -> str: ...
+    async def get(self, key: Literal["operation_name"]) -> OperationName: ...
     @overload
-    async def get(self, key: Literal["group_index"]) -> int: ...
+    async def get(self, key: Literal["operation_context"]) -> OperationContext: ...
+    @overload
+    async def get(self, key: Literal["group_index"]) -> NonNegativeInt: ...
     @overload
     async def get(self, key: Literal["is_creating"]) -> bool: ...
     async def get(self, key: str) -> Any:
-        (result,) = await self._store.get(self._get_hash_key(), key)
+        """raises KeyNotFoundInHashError if the key is not present in the hash"""
+        hash_key = self._get_hash_key()
+        (result,) = await self._store.get(hash_key, key)
+        if result is None:
+            raise KeyNotFoundInHashError(
+                schedule_id=self._schedule_id, hash_key=hash_key
+            )
         return result
 
     @overload
-    async def set(self, key: Literal["operation_name"], value: str) -> None: ...
+    async def set(
+        self, key: Literal["operation_name"], value: OperationName
+    ) -> None: ...
+    @overload
+    async def set(
+        self, key: Literal["operation_context"], value: OperationContext
+    ) -> None: ...
     @overload
-    async def set(self, key: Literal["group_index"], value: int) -> None: ...
+    async def set(self, key: Literal["group_index"], value: NonNegativeInt) -> None: ...
     @overload
     async def set(self, key: Literal["is_creating"], *, value: bool) -> None: ...
     async def set(self, key: str, value: Any) -> None:
@@ -150,7 +174,7 @@ def __init__(
         store: Store,
         schedule_id: ScheduleId,
         operation_name: OperationName,
-        group: StepGroup,
+        group: StepGroupName,
         step_name: StepName,
     ) -> None:
         self._store = store

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

@@ -0,0 +1,123 @@
+# pylint: disable=protected-access
+
+import pytest
+from simcore_service_dynamic_scheduler.services.generic_scheduler._errors import (
+    OperationAlreadyRegisteredError,
+    OperationNotFoundError,
+)
+from simcore_service_dynamic_scheduler.services.generic_scheduler._operation import (
+    BaseStep,
+    Operation,
+    OperationRegistry,
+    ParallelStepGroup,
+    SingleStepGroup,
+    _validate_operation,
+)
+
+
+class BS1(BaseStep):
+    pass
+
+
+class BS2(BaseStep):
+    pass
+
+
+class BS3(BaseStep):
+    pass
+
+
+@pytest.mark.parametrize(
+    "operation",
+    [
+        [
+            SingleStepGroup(BS1),
+            ParallelStepGroup(BS2, BS3),
+        ],
+        [
+            SingleStepGroup(BS1),
+        ],
+        [
+            SingleStepGroup(BS1),
+            SingleStepGroup(BS2),
+        ],
+        [
+            SingleStepGroup(BS2),
+            ParallelStepGroup(BS1, BS3, repeat_steps=True),
+        ],
+        [
+            ParallelStepGroup(BS1, BS3),
+            SingleStepGroup(BS2, repeat_steps=True),
+        ],
+        [
+            SingleStepGroup(BS1, repeat_steps=True),
+        ],
+        [
+            ParallelStepGroup(BS1, BS3, repeat_steps=True),
+        ],
+    ],
+)
+def test_validate_operation_passes(operation: Operation):
+    _validate_operation(operation)
+
+
+@pytest.mark.parametrize(
+    "operation, match",
+    [
+        ([], "List should have at least 1 item after validation"),
+        (
+            [
+                SingleStepGroup(BS1, repeat_steps=True),
+                SingleStepGroup(BS2),
+            ],
+            "Only the last step group can have repeat_steps=True",
+        ),
+        (
+            [
+                SingleStepGroup(BS1),
+                SingleStepGroup(BS1),
+            ],
+            f"step_name='{BS1.__name__}' is already used in this operation",
+        ),
+        (
+            [
+                ParallelStepGroup(BS2, BS2),
+            ],
+            f"step_name='{BS2.__name__}' is already used in this operation",
+        ),
+        (
+            [
+                ParallelStepGroup(BS1),
+            ],
+            f"{ParallelStepGroup.__name__} needs at least 2 steps",
+        ),
+    ],
+)
+def test_validate_operations_fails(operation: Operation, match: str):
+    with pytest.raises(ValueError, match=match):
+        _validate_operation(operation)
+
+
+def test_operation_registry_workflow():
+    operation: Operation = [SingleStepGroup(BS1)]
+    OperationRegistry.register("op1", operation)
+    assert len(OperationRegistry._OPERATIONS) == 1
+
+    assert OperationRegistry.get("op1") == operation
+
+    OperationRegistry.unregister("op1")
+    assert len(OperationRegistry._OPERATIONS) == 0
+
+
+def test_operation_registry_register_twice_fails():
+    operation: Operation = [SingleStepGroup(BS1)]
+    OperationRegistry.register("op1", operation)
+
+    with pytest.raises(OperationAlreadyRegisteredError):
+        OperationRegistry.register("op1", operation)
+
+    with pytest.raises(OperationNotFoundError):
+        OperationRegistry.get("non_existing")
+
+    with pytest.raises(OperationNotFoundError):
+        OperationRegistry.unregister("non_existing")