feat(experimental): Add Machine.model_override and `experimental.ut…

…ils.generate_base_model`

When `model_override` is set to True, Machine will assign only methods to a model that are already defined. This eases static type checking (#658) and enables tailored helper function assigment. Default value is False which prevents override of already defined model functions.

Changelog.md

@@ -5,6 +5,8 @@
 - Bug #610: Decorate models appropriately when `HierarchicalMachine` is passed to `add_state` (thanks @e0lithic)
 - Bug #647: Let `may_<trigger>` check all parallel states in processing order (thanks @spearsear)
 - 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.
 
 ## 0.9.1 (May 2024)
 

README.md

@@ -49,6 +49,7 @@ A lightweight, object-oriented state machine implementation in Python with many
   - [Alternative initialization patterns](#alternative-initialization-patterns)
   - [Logging](#logging)
   - [(Re-)Storing machine instances](#restoring)
+  - [Typing support](#typing-support)
   - [Extensions](#extensions)
     - [Diagrams](#diagrams)
     - [Hierarchical State Machine](#hsm)
@@ -1226,6 +1227,79 @@ m2.states.keys()
 >>> ['A', 'B', 'C']
 ```
 
+### <a name="typing-support"></a> Typing support
+
+As you probably noticed, `transitions` uses some of Python's dynamic features to give you handy ways to handle models. However, static type checkers don't like model attributes and methods not being known before runtime. Historically, `transitions` also didn't assign convenience methods already defined on models to prevent accidental overrides.
+
+But don't worry!  You can use the machine constructor parameter `model_override` to change how models are decorated. If you set `model_override=True`, `transitions` will only override already defined methods. This prevents new methods from showing up at runtime and also allows you to define which helper methods you want to use.
+
+```python
+from transitions import Machine
+
+# Dynamic assignment
+class Model:
+    pass
+
+model = Model()
+default_machine = Machine(model, states=["A", "B"], transitions=[["go", "A", "B"]], initial="A")
+print(model.__dict__.keys())  # all convenience functions have been assigned
+# >> dict_keys(['trigger', 'to_A', 'may_to_A', 'to_B', 'may_to_B', 'go', 'may_go', 'is_A', 'is_B', 'state'])
+assert model.is_A()  # Unresolved attribute reference 'is_A' for class 'Model'
+
+
+# Predefined assigment: We are just interested in calling our 'go' event and will trigger the other events by name
+class PredefinedModel:
+    # state (or another parameter if you set 'model_attribute') will be assigned anyway 
+    # because we need to keep track of the model's state
+    state: str
+
+    def go(self) -> bool:
+        raise RuntimeError("Should be overridden!")
+
+    def trigger(self, trigger_name: str) -> bool:
+        raise RuntimeError("Should be overridden!")
+
+
+model = PredefinedModel()
+override_machine = Machine(model, states=["A", "B"], transitions=[["go", "A", "B"]], initial="A", model_override=True)
+print(model.__dict__.keys())
+# >> dict_keys(['trigger', 'go', 'state'])
+model.trigger("to_B")
+assert model.state == "B"
+```
+
+If you want to use all the convenience functions and throw some callbacks into the mix, defining a model can get pretty complicated when you have a lot of states and transitions defined.
+The method `generate_base_model` in `transitions` can generate a base model from a machine configuration to help you out with that.
+
+```python
+from transitions.experimental.utils import generate_base_model
+simple_config = {
+    "states": ["A", "B"],
+    "transitions": [
+        ["go", "A", "B"],
+    ],
+    "initial": "A",
+    "before_state_change": "call_this",
+    "model_override": True,
+} 
+
+class_definition = generate_base_model(simple_config)
+with open("base_model.py", "w") as f:
+    f.write(class_definition)
+
+# ... in another file
+from transitions import Machine
+from base_model import BaseModel
+
+class Model(BaseModel):  #  call_this will be an abstract method in BaseModel
+
+    def call_this(self) -> None:
+        # do something  
+
+model = Model()
+machine = Machine(model, **simple_config)
+```
+
 ### <a name="extensions"></a> Extensions
 
 Even though the core of transitions is kept lightweight, there are a variety of MixIns to extend its functionality. Currently supported are:

tests/test_experimental.py

@@ -1,23 +1,127 @@
+from typing import TYPE_CHECKING
 from unittest import TestCase
+from types import ModuleType
 from unittest.mock import MagicMock
 
 from transitions import Machine
+from transitions.experimental.utils import generate_base_model
+from transitions.extensions import HierarchicalMachine
+
+from .utils import Stuff
+
+if TYPE_CHECKING:
+    from transitions.core import MachineConfig
+    from typing import Type
+
+
+def import_code(code: str, name: str) -> ModuleType:
+    module = ModuleType(name)
+    exec(code, module.__dict__)
+    return module
 
 
 class TestExperimental(TestCase):
 
     def setUp(self) -> None:
+        self.machine_cls = Machine  # type: Type[Machine]
 
+    def test_model_override(self):
 
         class Model:
 
+            def trigger(self, name: str) -> bool:
                 raise RuntimeError("Should be overridden")
 
+            def is_A(self) -> bool:
+                raise RuntimeError("Should be overridden")
 
             def is_C(self) -> bool:
+                raise RuntimeError("Should be overridden")
 
         model = Model()
+        machine = self.machine_cls(model, states=["A", "B"], initial="A", model_override=True)
         self.assertTrue(model.is_A())
+        with self.assertRaises(AttributeError):
+            model.to_B()  # type: ignore # Should not be assigned to model since its not declared
+        self.assertTrue(model.trigger("to_B"))
+        self.assertFalse(model.is_A())
+        with self.assertRaises(RuntimeError):
+            model.is_C()  # not overridden yet
         machine.add_state("C")
         self.assertFalse(model.is_C())  # now it is!
+        self.assertTrue(model.trigger("to_C"))
         self.assertTrue(model.is_C())
+
+    def test_generate_base_model(self):
+        simple_config = {
+            "states": ["A", "B"],
+            "transitions": [
+                ["go", "A", "B"],
+                ["back", "*", "A"]
+            ],
+            "initial": "A",
+            "model_override": True
+        }  # type: MachineConfig
+
+        mod = import_code(generate_base_model(simple_config), "base_module")
+        model = mod.BaseModel()
+        machine = self.machine_cls(model, **simple_config)
+        self.assertTrue(model.is_A())
+        self.assertTrue(model.go())
+        self.assertTrue(model.is_B())
+        self.assertTrue(model.back())
+        self.assertTrue(model.state == "A")
+        with self.assertRaises(AttributeError):
+            model.is_C()
+
+    def test_generate_base_model_callbacks(self):
+        simple_config = {
+            "states": ["A", "B"],
+            "transitions": [
+                ["go", "A", "B"],
+            ],
+            "initial": "A",
+            "model_override": True,
+            "before_state_change": "call_this"
+        }  # type: MachineConfig
+
+        mod = import_code(generate_base_model(simple_config), "base_module")
+        mock = MagicMock()
+
+        class Model(mod.BaseModel):  # type: ignore
+
+            @staticmethod
+            def call_this() -> None:
+                mock()
+
+        model = Model()
+        machine = self.machine_cls(model, **simple_config)
+        self.assertTrue(model.is_A())
+        self.assertTrue(model.go())
+        self.assertTrue(mock.called)
+
+    def test_generate_model_no_auto(self):
+        simple_config: MachineConfig = {
+            "states": ["A", "B"],
+            "auto_transitions": False,
+            "model_override": True,
+            "transitions": [
+                ["go", "A", "B"],
+                ["back", "*", "A"]
+            ],
+            "initial": "A"
+        }
+        mod = import_code(generate_base_model(simple_config), "base_module")
+        model = mod.BaseModel()
+        machine = self.machine_cls(model, **simple_config)
+        self.assertTrue(model.is_A())
+        self.assertTrue(model.go())
+        with self.assertRaises(AttributeError):
+            model.to_B()
+
+
+class TestHSMExperimental(TestExperimental):
+
+    def setUp(self):
+        self.machine_cls = HierarchicalMachine  # type: Type[HierarchicalMachine]
+        self.create_trigger_class()

transitions/core.py

@@ -511,8 +511,8 @@ def __init__(self, model=self_literal, states=None, initial='initial', transitio
                  send_event=False, auto_transitions=True,
                  ordered_transitions=False, ignore_invalid_triggers=None,
                  before_state_change=None, after_state_change=None, name=None,
-                 queued=False, prepare_event=None, finalize_event=None, model_attribute='state', on_exception=None,
-                 on_final=None, **kwargs):
+                 queued=False, prepare_event=None, finalize_event=None, model_attribute='state', model_override=False,
+                 on_exception=None, on_final=None, **kwargs):
         """
         Args:
             model (object or list): The object(s) whose states we want to manage. If set to `Machine.self_literal`
@@ -593,6 +593,7 @@ def __init__(self, model=self_literal, states=None, initial='initial', transitio
         self.on_final = on_final
         self.name = name + ": " if name is not None else ""
         self.model_attribute = model_attribute
+        self.model_override = model_override
 
         self.models = []
 
@@ -887,10 +888,10 @@ def _add_model_to_state(self, state, model):
 
     def _checked_assignment(self, model, name, func):
         bound_func = getattr(model, name, None)
-        if bound_func is None or getattr(bound_func, "expect_override", False):
+        if (bound_func is None) ^ self.model_override:
             setattr(model, name, func)
         else:
-            _LOGGER.warning("%sModel already contains an attribute '%s'. Skip binding.", self.name, name)
+            _LOGGER.warning("%sSkip binding of '%s' to model due to model override policy.", self.name, name)
 
     def _can_trigger(self, model, trigger, *args, **kwargs):
         state = self.get_model_state(model)

transitions/core.pyi

@@ -120,6 +120,7 @@ class Machine:
     ignore_invalid_triggers: Optional[bool]
     name: str
     model_attribute: str
+    model_override: bool
     models: List[Any]
     def __init__(self, model: Optional[ModelParameter] = ...,
                  states: Optional[Union[Sequence[StateConfig], Type[Enum]]] = ...,
@@ -130,7 +131,8 @@ class Machine:
                  before_state_change: CallbacksArg = ..., after_state_change: CallbacksArg = ...,
                  name: str = ..., queued: bool = ...,
                  prepare_event: CallbacksArg = ..., finalize_event: CallbacksArg = ...,
-                 model_attribute: str = ..., on_exception: CallbacksArg = ..., **kwargs: Dict[str, Any]) -> None: ...
+                 model_attribute: str = ..., model_override: bool = ...,
+                 on_exception: CallbacksArg = ..., on_final: CallbacksArg = ..., **kwargs: Dict[str, Any]) -> None: ...
     def add_model(self, model: ModelParameter,
                   initial: Optional[StateIdentifier] = ...) -> None: ...
     def remove_model(self, model: ModelParameter) -> None: ...

transitions/experimental/utils.py

@@ -0,0 +1,93 @@
+from collections import deque, defaultdict
+
+from transitions.core import listify
+from transitions.extensions.markup import HierarchicalMarkupMachine
+
+
+_placeholder_body = "raise RuntimeError('This should be overridden')"
+
+
+def generate_base_model(config):
+    m = HierarchicalMarkupMachine(**config)
+    triggers = set()
+    markup = m.markup
+    model_attribute = markup.get("model_attribute", "state")
+    trigger_block = ""
+    state_block = ""
+    callback_block = ""
+
+    callbacks = set(
+        [cb for cb in markup["prepare_event"]]
+        + [cb for cb in markup["before_state_change"]]
+        + [cb for cb in markup["after_state_change"]]
+        + [cb for cb in markup["on_exception"]]
+        + [cb for cb in markup["on_final"]]
+        + [cb for cb in markup["finalize_event"]]
+    )
+
+    for trans in markup["transitions"]:
+        triggers.add(trans["trigger"])
+
+    stack = [(markup["states"], markup["transitions"], "")]
+    has_nested_states = any("children" in state for state in markup["states"])
+    while stack:
+        states, transitions, prefix = stack.pop()
+        for state in states:
+            state_name = state["name"]
+
+            state_block += (
+                f"    def is_{prefix}{state_name}(self{', allow_substates=False' if has_nested_states else ''})"
+                f" -> bool: {_placeholder_body}\n"
+            )
+            if m.auto_transitions:
+                state_block += (
+                    f"    def to_{prefix}{state_name}(self) -> bool: {_placeholder_body}\n"
+                    f"    def may_to_{prefix}{state_name}(self) -> bool: {_placeholder_body}\n"
+                )
+
+            state_block += "\n"
+            for tran in transitions:
+                triggers.add(tran["trigger"])
+                new_set = set(
+                    [cb for cb in tran.get("prepare", [])]
+                    + [cb for cb in tran.get("conditions", [])]
+                    + [cb for cb in tran.get("unless", [])]
+                    + [cb for cb in tran.get("before", [])]
+                    + [cb for cb in tran.get("after", [])]
+                )
+                callbacks.update(new_set)
+
+            if "children" in state:
+                stack.append((state["children"], state.get("transitions", []), prefix + state_name + "_"))
+
+    for trigger_name in triggers:
+        trigger_block += (
+            f"    def {trigger_name}(self) -> bool: {_placeholder_body}\n"
+            f"    def may_{trigger_name}(self) -> bool: {_placeholder_body}\n"
+        )
+
+    extra_params = "event_data: EventData" if m.send_event else "*args: List[Any], **kwargs: Dict[str, Any]"
+    for callback_name in callbacks:
+        if isinstance(callback_name, str):
+            callback_block += (f"    @abstractmethod\n"
+                               f"    def {callback_name}(self, {extra_params}) -> Optional[bool]: ...\n")
+
+    template = f"""# autogenerated by transitions
+from abc import ABCMeta, abstractmethod
+from typing import Any, Callable, Dict, List, Optional, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from transitions.core import CallbacksArg, StateIdentifier, EventData
+
+
+class BaseModel(metaclass=ABCMeta):
+    {model_attribute}: "StateIdentifier" = ""
+    def trigger(self, name: str) -> bool: {_placeholder_body}
+
+{trigger_block}
+{state_block}\
+{callback_block}"""
+
+    return template
+
+

transitions/experimental/utils.pyi

@@ -0,0 +1,7 @@
+from typing import Union, Callable, List, Optional, Iterable, Type, ClassVar, Tuple, Dict, Any, DefaultDict, Deque
+from transitions.core import StateIdentifier, CallbacksArg, CallbackFunc, Machine, TransitionConfig, MachineConfig
+from transitions.extensions.markup import MarkupConfig
+
+_placeholder_body: str
+
+def generate_base_model(config: Union[MachineConfig, MarkupConfig]) -> str: ...