chore: Improved enum support; Allow binding eventes to objects (#454)

Author: fgmacedo

docs/mixins.md

@@ -44,6 +44,7 @@ class.
 ...     state_machine_name = '__main__.CampaignMachineWithKeys'
 ...     state_machine_attr = 'sm'
 ...     state_field_name = 'workflow_step'
+...     bind_events_as_methods = True
 ...
 ...     workflow_step = 1
 ...
@@ -65,7 +66,11 @@ True
 >>> model.sm.current_state == model.sm.draft
 True
 
->>> model.sm.cancel()
+>>> model.produce()  # `bind_events_as_methods = True` adds triggers to events in the mixin instance
+>>> model.workflow_step
+2
+
+>>> model.sm.cancel()  # You can still call the SM directly
 
 >>> model.workflow_step
 4

docs/releases/2.3.2.md

@@ -41,6 +41,17 @@ Paulista Avenue after: green--(cycle)-->yellow
 See {ref}`listeners` for more details.
 ```
 
+### Binding event triggers to external objects
+
+Now it's possible to bind events to external objets. One expected use case is in conjunction with the {ref}`Mixins` models,
+that wrap state machines internally. This way you don't need to expose the state machine.
+
+
+```{seealso}
+See {ref}`sphx_glr_auto_examples_user_machine.py` for an example binding event triggers with a state machine.
+```
+
+
 ## Bugfixes in 2.3.2
 
 - Fixes [#446](https://github.com/fgmacedo/python-statemachine/issues/446): Regression that broke sync callbacks

statemachine/mixins.py

@@ -7,24 +7,30 @@ class MachineMixin:
     ``StateMachine``.
     """
 
-    state_field_name = "state"  # type: str
+    state_field_name: str = "state"
     """The model's state field name that will hold the state value."""
 
-    state_machine_name = None  # type: str
+    state_machine_name: "str | None" = None
     """A fully qualified name of the class, where it can be imported."""
 
-    state_machine_attr = "statemachine"  # type: str
+    state_machine_attr: str = "statemachine"
     """Name of the model's attribute that will hold the machine instance."""
 
+    bind_events_as_methods: bool = False
+    """If ``True`` the state machine events triggers will be bound to the model as methods."""
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         if not self.state_machine_name:
             raise ValueError(
                 _("{!r} is not a valid state machine name.").format(self.state_machine_name)
             )
         machine_cls = registry.get_machine_cls(self.state_machine_name)
+        sm = machine_cls(self, state_field=self.state_field_name)
         setattr(
             self,
             self.state_machine_attr,
-            machine_cls(self, state_field=self.state_field_name),
+            sm,
         )
+        if self.bind_events_as_methods:
+            sm.bind_events_to(self)

statemachine/statemachine.py

@@ -141,6 +141,22 @@ def _get_initial_state(self):
         except KeyError as err:
             raise InvalidStateValue(current_state_value) from err
 
+    def bind_events_to(self, *targets):
+        """Bind the state machine events to the target objects."""
+
+        for event in self.events:
+            trigger = getattr(self, event.name)
+            for target in targets:
+                if hasattr(target, event.name):
+                    warnings.warn(
+                        f"Attribute {event.name!r} already exists on {target!r}. "
+                        f"Skipping binding.",
+                        UserWarning,
+                        stacklevel=2,
+                    )
+                    continue
+                setattr(target, event.name, trigger)
+
     async def activate_initial_state(self):
         """
         Activate the initial state.
@@ -173,9 +189,6 @@ async def activate_initial_state(self):
             )
             await self._activate(event_data)
 
-    async def _ensure_is_initialized(self):
-        await self.activate_initial_state()
-
     def _add_listener(self, listeners: "Listeners"):
         register = partial(listeners.resolve, registry=self._callbacks_registry)
         for visited in iterate_states_and_transitions(self.states):
@@ -295,7 +308,7 @@ def allowed_events(self):
 
     async def _trigger(self, trigger_data: TriggerData):
         event_data = None
-        await self._ensure_is_initialized()
+        await self.activate_initial_state()
 
         state = self.current_state
         for transition in state.transitions:

statemachine/states.py

@@ -54,6 +54,7 @@ def __eq__(self, other):
         return list(self) == list(other)
 
     def __getattr__(self, name: str):
+        name = name.lower()
         if name in self._states:
             return self._states[name]
         raise AttributeError(f"{name} not found in {self.__class__.__name__}")
@@ -80,7 +81,7 @@ def items(self):
         return self._states.items()
 
     @classmethod
-    def from_enum(cls, enum_type: EnumType, initial: Enum, final=None):
+    def from_enum(cls, enum_type: EnumType, initial, final=None):
         """
         Creates a new instance of the ``States`` class from an enumeration.
 
@@ -124,7 +125,7 @@ def from_enum(cls, enum_type: EnumType, initial: Enum, final=None):
         True
 
         >>> sm.current_state_value
-        2
+        <Status.completed: 2>
 
         Args:
             enum_type: An enumeration containing the states of the machine.
@@ -137,7 +138,7 @@ def from_enum(cls, enum_type: EnumType, initial: Enum, final=None):
         final_set = set(ensure_iterable(final))
         return cls(
             {
-                e.name: State(value=e.value, initial=e is initial, final=e in final_set)
+                e.name.lower(): State(value=e, initial=e is initial, final=e in final_set)
                 for e in enum_type
             }
         )

tests/django_project/workflow/models.py

@@ -6,10 +6,18 @@
 User = get_user_model()
 
 
+class WorkflowSteps(models.TextChoices):
+    DRAFT = "draft"
+    PUBLISHED = "published"
+
+
 class Workflow(models.Model, MachineMixin):
     state_machine_name = "workflow.statemachines.WorfklowStateMachine"
     state_machine_attr = "wf"
+    bind_events_as_methods = True
 
-    state = models.CharField(max_length=30, blank=True, null=True)
+    state = models.CharField(
+        max_length=30, choices=WorkflowSteps.choices, default=WorkflowSteps.DRAFT
+    )
     user = models.ForeignKey(User, on_delete=models.CASCADE, null=True)
     is_active = models.BooleanField(default=False)

tests/django_project/workflow/statemachines.py

@@ -1,13 +1,14 @@
-from statemachine import State
 from statemachine import StateMachine
+from statemachine.states import States
+
+from .models import WorkflowSteps
 
 
 class WorfklowStateMachine(StateMachine):
-    draft = State(initial=True)
-    published = State(final=True)
+    _ = States.from_enum(WorkflowSteps, initial=WorkflowSteps.DRAFT, final=WorkflowSteps.PUBLISHED)
 
-    publish = draft.to(published, cond="is_active")
-    notify_user = draft.to.itself(internal=True, cond="has_user")
+    publish = _.draft.to(_.published, cond="is_active")
+    notify_user = _.draft.to.itself(internal=True, cond="has_user")
 
     def has_user(self):
         return bool(self.model.user)

tests/django_project/workflow/tests.py

@@ -1,6 +1,7 @@
 import pytest
 
 from statemachine.exceptions import TransitionNotAllowed
+from workflow.models import WorkflowSteps
 from workflow.statemachines import WorfklowStateMachine
 
 pytestmark = [
@@ -59,3 +60,12 @@ def test_async_with_db_operation(self, one, User, Workflow):
 
         wf = WorfklowStateMachine(one)
         wf.send("notify_user")
+
+    def test_should_publish(self, one):
+        one.is_active = True
+        one.publish()
+        one.save()
+
+        assert one.state == "published"
+        assert one.wf.current_state_value == "published"
+        assert one.wf.current_state_value == WorkflowSteps.PUBLISHED

tests/examples/enum_campaign_machine.py

@@ -54,4 +54,4 @@ class CampaignMachine(StateMachine):
 assert sm.producing.is_active is True
 assert sm.closed.is_active is False
 assert sm.current_state == sm.producing
-assert sm.current_state_value == CampaignStatus.producing.value
+assert sm.current_state_value == CampaignStatus.producing

tests/examples/user_machine.py

@@ -0,0 +1,130 @@
+"""
+User workflow machine
+=====================
+
+This machine binds the events to the User model, the StateMachine is wrapped internally
+in the `User` class.
+
+Demonstrates that multiple state machines can be used in the same model.
+
+And that logic can be reused with listeners.
+
+"""
+
+from dataclasses import dataclass
+from enum import Enum
+
+from statemachine import State
+from statemachine import StateMachine
+from statemachine.states import States
+
+
+class UserStatus(str, Enum):
+    signup_incomplete = "SIGNUP_INCOMPLETE"
+    signup_complete = "SIGNUP_COMPLETE"
+    signup_rejected = "SIGNUP_REJECTED"
+    operational_enabled = "OPERATIONAL_ENABLED"
+    operational_disabled = "OPERATIONAL_DISABLED"
+    operational_rescinded = "OPERATIONAL_RESCINDED"
+
+
+class UserExperience(str, Enum):
+    basic = "BASIC"
+    premium = "PREMIUM"
+
+
+@dataclass
+class User:
+    name: str
+    email: str
+    status: UserStatus = UserStatus.signup_incomplete
+    experience: UserExperience = UserExperience.basic
+
+    verified: bool = False
+
+    def __post_init__(self):
+        self._status_sm = UserStatusMachine(
+            self, state_field="status", listeners=[MachineChangeListenter()]
+        )
+        self._status_sm.bind_events_to(self)
+
+        self._experience_sm = UserExperienceMachine(
+            self, state_field="experience", listeners=[MachineChangeListenter()]
+        )
+        self._experience_sm.bind_events_to(self)
+
+
+class MachineChangeListenter:
+    def before_transition(self, event: str, state: State):
+        print(f"Before {event} in {state}")
+
+    def on_enter_state(self, state: State, event: str):
+        print(f"Entering {state} from {event}")
+
+
+class UserStatusMachine(StateMachine):
+    _states = States.from_enum(
+        UserStatus,
+        initial=UserStatus.signup_incomplete,
+        final=[
+            UserStatus.operational_rescinded,
+            UserStatus.signup_rejected,
+        ],
+    )
+
+    signup = _states.signup_incomplete.to(_states.signup_complete)
+    reject = _states.signup_rejected.from_(
+        _states.signup_incomplete,
+        _states.signup_complete,
+    )
+    enable = _states.signup_complete.to(_states.operational_enabled)
+    disable = _states.operational_enabled.to(_states.operational_disabled)
+    rescind = _states.operational_rescinded.from_(
+        _states.operational_enabled,
+        _states.operational_disabled,
+    )
+
+    def on_signup(self, token: str):
+        if token == "":
+            raise ValueError("Token is required")
+        self.model.verified = True
+
+
+class UserExperienceMachine(StateMachine):
+    _states = States.from_enum(
+        UserExperience,
+        initial=UserExperience.basic,
+    )
+
+    upgrade = _states.basic.to(_states.premium)
+    downgrade = _states.premium.to(_states.basic)
+
+
+# %%
+# Executing
+
+
+def main():  # type: ignore[attr-defined]
+    # By binding the events to the User model, the events can be fired directly from the model
+    user = User(name="Frodo", email="frodo@lor.com")
+
+    try:
+        # Trying to signup with an empty token should raise an exception
+        user.signup("")
+    except Exception as e:
+        print(e)
+
+    assert user.verified is False
+
+    user.signup("1234")
+
+    assert user.status == UserStatus.signup_complete
+    assert user.verified is True
+
+    print(user.experience)
+    user.upgrade()
+    print(user.experience)
+
+
+if __name__ == "__main__":
+    main()