python-statemachine/tests/examples/statechart_error_handling_machine.py at 9c190636fcfadb1a039ab31084d41aa245165670 · fgmacedo/python-statemachine · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
"""
Error handling -- Quest Recovery
=================================

This example demonstrates **error.execution** handling using ``StateChart``.

When ``error_on_execution=True`` (the ``StateChart`` default), runtime errors in
callbacks are caught and dispatched as ``error.execution`` events instead of
propagating as exceptions. This lets you define error-recovery transitions.

- The ``error_`` naming convention auto-registers both ``error_X`` and ``error.X``
  event names.
- Alternatively, use ``Event(transitions, id="error.execution")`` for explicit
  registration.
- Error data (the original exception, event, etc.) is available in handler kwargs.

"""

from statemachine import Event
from statemachine import State
from statemachine import StateChart


class QuestRecoveryMachine(StateChart):
    """A quest where actions can fail and the error handler routes to recovery.

    When ``on_enter_danger_zone`` raises, the ``error.execution`` event fires
    and transitions to the ``recovering`` state instead of crashing.
    """

    safe = State("Safe", initial=True)
    danger_zone = State("Danger Zone")
    recovering = State("Recovering")
    completed = State("Quest Complete", final=True)

    venture = safe.to(danger_zone)
    survive = danger_zone.to(completed)
    recover = recovering.to(safe)

    # Register error.execution handler using Event with explicit id
    error_execution = Event(
        safe.to(recovering) | danger_zone.to(recovering),
        id="error.execution",
    )

    def on_enter_danger_zone(self):
        # This simulates an unexpected error during a quest action
        raise RuntimeError("Ambush! Orcs attack!")

    def on_enter_recovering(self, error=None, **kwargs):
        if error:
            print(f"Error caught: {error}")
        print("Retreating to recover...")


# %%
# Error triggers recovery instead of crashing
# ----------------------------------------------
#
# When entering ``danger_zone`` raises a ``RuntimeError``, the error is caught
# and dispatched as ``error.execution``. The machine transitions to ``recovering``.

sm = QuestRecoveryMachine()
print(f"Start: {sorted(sm.configuration_values)}")
assert "safe" in sm.configuration_values

sm.send("venture")
print(f"After venture: {sorted(sm.configuration_values)}")
assert "recovering" in sm.configuration_values

# %%
# Recover and try again
# -----------------------

sm.send("recover")
print(f"After recovery: {sorted(sm.configuration_values)}")
assert "safe" in sm.configuration_values


# %%
# Comparison with error_on_execution=False (error propagation)
# --------------------------------------------------------------
#
# With ``error_on_execution=False``, the same error
# would propagate as an exception instead of being caught.


class QuestNoCatch(StateChart):
    error_on_execution = False

    safe = State("Safe", initial=True)
    danger_zone = State("Danger Zone")

    venture = safe.to(danger_zone)

    def on_enter_danger_zone(self):
        raise RuntimeError("Ambush! Orcs attack!")


sm2 = QuestNoCatch()
try:
    sm2.send("venture")
except RuntimeError as e:
    print(f"Exception propagated: {e}")