feat(experimental): add transitions.experimental.utils.{add_transitio…

…ns, event, with_model_definitions, transition}

Add helper functions to define transitions on a model class for better type checking (#658)

Changelog.md

@@ -7,6 +7,7 @@
 - Bug: `HSM.is_state` works with parallel states now
 - Experimental features: 
   + Add `model_override` to Machine constructor to determine the mode of operation. With `model_override=Fale` (default), `transitions` will not override already defined methods on a model just as it did before. For workflows relying on typing, `model_override=True` will override methods already defined on the model and only those (!). This allows to control which convenience methods shall be assigned to the model and keeps the statically 'assumed' model in sync with its runtime counterpart. Since defining each and every method manually is rather tiresome, `transitions.experimental.utils.generate_base_model` features a way to convert a machine configuration into a `BaseClass` with all convenience functions and callbacks.
+  + Add `transitions.experimental.utils.{add_transitions, event, with_model_definitions, transition}` to define trigger methods in a class model for more convenient type checking. `add_transitions` can be used as a function decorator and is stackable. `event` returns a placeholder object for attribute assigment. `add_transitions` and `event` have the same signature and support transition definition like machine constructors. The function `transition` can used for better typing and returns a dictionary that can be passed to the utility functions but also to a machine constructor. `add_transitions` and `event` require a machine decorated with `with_model_definitions`. Decorating a machine `with_model_definitions` implies `model_override=True`.
 
 ## 0.9.1 (May 2024)
 

tests/test_experimental.py

@@ -5,6 +5,7 @@
 
 from transitions import Machine
 from transitions.experimental.utils import generate_base_model
+from transitions.experimental.utils import add_transitions, transition, event, with_model_definitions
 from transitions.extensions import HierarchicalMachine
 
 from .utils import Stuff
@@ -24,6 +25,14 @@ class TestExperimental(TestCase):
 
     def setUp(self) -> None:
         self.machine_cls = Machine  # type: Type[Machine]
+        self.create_trigger_class()
+
+    def create_trigger_class(self):
+        @with_model_definitions
+        class TriggerMachine(self.machine_cls):  # type: ignore
+            pass
+
+        self.trigger_machine = TriggerMachine
 
     def test_model_override(self):
 
@@ -119,6 +128,113 @@ def test_generate_model_no_auto(self):
         with self.assertRaises(AttributeError):
             model.to_B()
 
+    def test_decorator(self):
+
+        class Model:
+
+            state: str = ""
+
+            def is_B(self) -> bool:
+                return False
+
+            @add_transitions(transition(source="A", dest="B"))
+            @add_transitions([["A", "B"], "C"])
+            def go(self) -> bool:
+                raise RuntimeError("Should be overridden!")
+
+        model = Model()
+        machine = self.trigger_machine(model, states=["A", "B", "C"], initial="A")
+        self.assertEqual("A", model.state)
+        self.assertTrue(machine.is_state("A", model))
+        self.assertTrue(model.go())
+        with self.assertRaises(AttributeError):
+            model.is_A()  # type: ignore
+        self.assertEqual("B", model.state)
+        self.assertTrue(model.is_B())
+        self.assertTrue(model.go())
+        self.assertFalse(model.is_B())
+        self.assertEqual("C", model.state)
+
+    def test_decorator_complex(self):
+
+        class Model:
+
+            state: str = ""
+
+            def check_param(self, param: bool) -> bool:
+                return param
+
+            @add_transitions(transition(source="A", dest="B"),
+                             transition(source="B", dest="C", unless=Stuff.this_passes),
+                             transition(source="B", dest="A", conditions=Stuff.this_passes, unless=Stuff.this_fails))
+            def go(self) -> bool:
+                raise RuntimeError("Should be overridden")
+
+            @add_transitions({"source": "A", "dest": "B", "conditions": "check_param"})
+            def event(self, param) -> bool:
+                raise RuntimeError("Should be overridden")
+
+        model = Model()
+        machine = self.trigger_machine(model, states=["A", "B"], initial="A")
+        self.assertTrue(model.go())
+        self.assertTrue(model.state == "B")
+        self.assertTrue(model.go())
+        self.assertTrue(model.state == "A")
+        self.assertFalse(model.event(param=False))
+        self.assertTrue(model.state == "A")
+        self.assertTrue(model.event(param=True))
+        self.assertTrue(model.state == "B")
+
+    def test_event_definition(self):
+
+        class Model:
+
+            state: str = ""
+
+            def is_B(self) -> bool:
+                return False
+
+            go = event(transition(source="A", dest="B"), [["A", "B"], "C"])
+
+        model = Model()
+        machine = self.trigger_machine(model, states=["A", "B", "C"], initial="A")
+        self.assertEqual("A", model.state)
+        self.assertTrue(machine.is_state("A", model))
+        self.assertTrue(model.go())
+        with self.assertRaises(AttributeError):
+            model.is_A()  # type: ignore
+        self.assertEqual("B", model.state)
+        self.assertTrue(model.is_B())
+        self.assertTrue(model.go())
+        self.assertFalse(model.is_B())
+        self.assertEqual("C", model.state)
+
+    def test_event_definition_complex(self):
+
+        class Model:
+
+            state: str = ""
+
+            go = event(transition(source="A", dest="B"),
+                       transition(source="B", dest="C", unless=Stuff.this_passes),
+                       transition(source="B", dest="A", conditions=Stuff.this_passes, unless=Stuff.this_fails))
+
+            event = event({"source": "A", "dest": "B", "conditions": "check_param"})
+
+            def check_param(self, param: bool) -> bool:
+                return param
+
+        model = Model()
+        machine = self.trigger_machine(model, states=["A", "B"], initial="A")
+        self.assertTrue(model.go())
+        self.assertTrue(model.state == "B")
+        self.assertTrue(model.go())
+        self.assertTrue(model.state == "A")
+        self.assertFalse(model.event(param=False))
+        self.assertTrue(model.state == "A")
+        self.assertTrue(model.event(param=True))
+        self.assertTrue(model.state == "B")
+
 
 class TestHSMExperimental(TestExperimental):
 

transitions/core.pyi

@@ -1,8 +1,7 @@
 from logging import Logger
-from functools import partial
 from typing import (
     Any, Optional, Callable, Sequence, Union, Iterable, List, Dict, DefaultDict,
-    Type, Deque, OrderedDict, Tuple, Literal, Collection
+    Type, Deque, OrderedDict, Tuple, Literal, Collection, TypedDict
 )
 
 # Enums are supported for Python 3.4+ and Python 2.7 with enum34 package installed
@@ -17,6 +16,27 @@ CallbacksArg = Optional[Union[Callback, CallbackList]]
 ModelState = Union[str, Enum, List["ModelState"]]
 ModelParameter = Union[Union[Literal['self'], Any], List[Union[Literal['self'], Any]]]
 
+
+class MachineConfig(TypedDict, total=False):
+    states: List[StateIdentifier]
+    transitions: List[TransitionConfig]
+    initial: str
+    auto_transitions: bool
+    ordered_transitions: bool
+    send_event: bool
+    ignore_invalid_triggers: bool
+    before_state_change: CallbacksArg
+    after_state_change: CallbacksArg
+    name: str
+    queued: bool
+    prepare_event: CallbacksArg
+    finalize_event: CallbacksArg
+    model_override: bool
+    model_attribute: str
+    on_exception: CallbacksArg
+    on_final: CallbacksArg
+
+
 def listify(obj: Union[None, List[Any], Tuple[Any], EnumMeta, Any]) -> Union[List[Any], Tuple[Any], EnumMeta]: ...
 
 def _prep_ordered_arg(desired_length: int, arguments: CallbacksArg) -> CallbackList: ...

transitions/experimental/utils.py

@@ -91,3 +91,57 @@ def trigger(self, name: str) -> bool: {_placeholder_body}
     return template
 
 
+def with_model_definitions(cls):
+    add_model = getattr(cls, "add_model")
+
+    def add_model_override(self, model, initial=None):
+        self.model_override = True
+        for model in listify(model):
+            model = self if model == "self" else model
+            for name, specs in TriggerPlaceholder.definitions.get(model.__class__).items():
+                for spec in specs:
+                    if isinstance(spec, list):
+                        self.add_transition(name, *spec)
+                    elif isinstance(spec, dict):
+                        self.add_transition(name, **spec)
+                    else:
+                        raise ValueError("Cannot add {} for event {} to machine", spec, name)
+        add_model(self, model, initial)
+
+    setattr(cls, 'add_model', add_model_override)
+    return cls
+
+
+class TriggerPlaceholder:
+    definitions = defaultdict(lambda: defaultdict(list))
+
+    def __init__(self, configs):
+        self.configs = deque(configs)
+
+    def __set_name__(self, owner, name):
+        for config in self.configs:
+            TriggerPlaceholder.definitions[owner][name].append(config)
+
+    def __call__(self, *args, **kwargs):
+        raise RuntimeError("Trigger was not initialized correctly!")
+
+
+def event(*configs):
+    return TriggerPlaceholder(configs)
+
+
+def add_transitions(*configs):
+    def _outer(trigger_func):
+        if isinstance(trigger_func, TriggerPlaceholder):
+            for config in reversed(configs):
+                trigger_func.configs.appendleft(config)
+        else:
+            trigger_func = TriggerPlaceholder(configs)
+        return trigger_func
+
+    return _outer
+
+
+def transition(source, dest=None, conditions=None, unless=None, before=None, after=None, prepare=None):
+    return {"source": source, "dest": dest, "conditions": conditions, "unless": unless, "before": before,
+            "after": after, "prepare": prepare}

transitions/experimental/utils.pyi

@@ -5,3 +5,23 @@ from transitions.extensions.markup import MarkupConfig
 _placeholder_body: str
 
 def generate_base_model(config: Union[MachineConfig, MarkupConfig]) -> str: ...
+
+def with_model_definitions(cls: Type[Machine]) -> Type[Machine]: ...
+
+def add_transitions(*configs: TransitionConfig) -> Callable[[CallbackFunc], CallbackFunc]: ...
+def event(*configs: TransitionConfig) -> Callable[..., Optional[bool]]: ...
+
+def transition(source: Union[StateIdentifier, List[StateIdentifier]],
+                dest: Optional[StateIdentifier] = ...,
+                conditions: CallbacksArg = ..., unless: CallbacksArg = ...,
+                before: CallbacksArg = ..., after: CallbacksArg = ...,
+                prepare: CallbacksArg = ...) -> TransitionConfig: ...
+
+class TriggerPlaceholder:
+    definitions: ClassVar[DefaultDict[type, DefaultDict[str, List[TransitionConfig]]]]
+    configs: Deque[TransitionConfig]
+    def __init__(self, configs: Iterable[TransitionConfig]) -> None: ...
+
+    def __set_name__(self, owner: type, name: str) -> None: ...
+
+    def __call__(self, *args: Tuple[Any], **kwargs: Dict[str, Any]) -> Optional[bool]: ...