python-statemachine

The invoke mechanism allows a state to spawn external work (such as API calls, file I/O, or child state machines) when it is entered. This work runs for the duration of the state and is automatically cancelled when the state is exited, following SCXML <invoke> semantics.

Execution Model

Invoke handlers run outside the main state machine processing loop:

• Sync engine: Each handler runs in a daemon thread.
• Async engine:
  • Sync handlers: Run in a thread executor (loop.run_in_executor) wrapped in an asyncio.Task to prevent blocking the event loop.
  • Coroutine functions and IInvoke handlers with async def run(): These are awaited directly on the event loop, making them ideal for non-blocking async I/O (e.g., aiohttp).

Lifecycle Events

• Completion: When a handler finishes, a done.invoke.<state>.<id> event is sent to the machine. The return value of the handler is passed as the data keyword argument to callbacks on the target state.
• Error: If a handler raises an exception, an error.execution event is sent.
• Cancellation: If the state is exited before completion, the invocation is cancelled. ctx.cancelled is set and on_cancel() is called on IInvoke handlers.

To break down a complex state into sub-steps, use State.Compound. When a compound state is entered, it automatically activates its own initial child state. If a child state is marked as final=True, reaching it can trigger automatic transitions out of the parent state. This allows you to nest logic without changing the top-level API used to send events.

from statemachine import StateChart, State

class CoffeeOrder(StateChart):
    pending = State(initial=True)

    class preparing(State.Compound):
        """Drink preparation with internal steps."""
        grinding = State(initial=True)
        brewing = State()
        serving = State(final=True)

        grind = grinding.to(brewing)
        brew = brewing.to(serving)

    picked_up = State(final=True)

    start = pending.to(preparing)
    done_state_preparing = preparing.to(picked_up)

order = CoffeeOrder()
order.send("start")
# order.configuration_values will contain {"preparing", "grinding"}

To run multiple callables concurrently and wait for all of them to complete before transitioning, use statemachine.invoke.invoke_group.

Behavior:

• The transition only fires after every callable in the group finishes.
• The data passed to the transition is a list of results in the same order as the input callables.
• If any callable raises an exception, the remaining ones are cancelled and an error.execution event is sent.
• If the owning state is exited before all callables finish, the group is cancelled.

Use this when you need a batch of operations to complete before moving to the next state.

from statemachine.invoke import invoke_group

class BatchLoader(StateChart):
    loading = State(
        initial=True,
        invoke=invoke_group(
            lambda: file_a.read_text(),
            lambda: file_b.read_text(),
        ),
    )
    ready = State(final=True)
    done_invoke_loading = loading.to(ready)

    def on_enter_ready(self, data=None, **kwargs):
        self.results = data

sm = BatchLoader()

Validator exceptions always propagate to the caller. They are not caught by the state machine engine, even if catch_errors_as_events=True is set.

This is because validators operate during the transition-selection phase. Because the transition is rejected before it even begins, validator exceptions:

• Are not converted to error.execution events.
• Do not trigger error.execution transitions.
• Must be handled by the caller using standard try/except blocks.

In contrast, errors occurring during the execution phase (e.g., inside an action method) are caught by the engine and can be routed through error transitions.

class GuardedWithErrorHandler(StateChart):
    idle = State(initial=True)
    active = State()
    error_state = State(final=True)

    # Validator error: propagates to caller, state remains 'idle'
    start = idle.to(active, validators="check_input")
    
    # Action error: caught by engine, state moves to 'error_state'
    do_work = active.to.itself(on="risky_action")
    error_execution = active.to(error_state)

    def check_input(self, value=None, **kwargs):
        if value is None:
            raise ValueError("Input required")

    def risky_action(self, **kwargs):
        raise RuntimeError("Boom")

sm = GuardedWithErrorHandler()

# 1. Validator rejection (direct exception)
try:
    sm.send("start")
except ValueError as e:
    print(e)  # "Input required"

# 2. Action error (routed via error_execution transition)
sm.send("start", value="ok")
sm.send("do_work")
# sm is now in 'error_state'

In hierarchical (compound) or parallel states, a transition may cross multiple levels. The engine exits and enters each level individually following the SCXML specification:

• Exit: Runs from the innermost (deepest child) state up to the ancestor being left. Children exit before their parents.
• Enter: Runs from the outermost (highest ancestor) state down to the target leaf. Parents enter before their children.

Use state-specific callbacks like on_exit_<state> or on_enter_<state> to target specific levels. Generic on_exit_state() and on_enter_state() callbacks also fire for each state in the set; for these, the state parameter is bound to the individual state being processed.

>>> from statemachine import State, StateChart

>>> class HierarchicalExample(StateChart):
...     class parent_a(State.Compound):
...         child_a = State(initial=True)
...     class parent_b(State.Compound):
...         child_b = State(initial=True, final=True)
...     cross = parent_a.to(parent_b)
...
...     def on_exit_child_a(self):
...         print("  exit  child_a")
...     def on_exit_parent_a(self):
...         print("  exit  parent_a")
...     def on_enter_parent_b(self):
...         print("  enter parent_b")
...     def on_enter_child_b(self):
...         print("  enter child_b")

>>> sm = HierarchicalExample()
>>> sm.send("cross")
  exit  child_a
  exit  parent_a
  enter parent_b
  enter child_b

In StateChart, exceptions raised during actions are caught by the engine and dispatched as error.execution internal events. This allows the machine to react to failures (e.g., transitioning to an error state or retrying) instead of crashing.

By default, StateChart is SCXML-compliant and uses the catch_errors_as_events class attribute set to True. If you set catch_errors_as_events = False, exceptions will propagate directly to the caller instead of being converted into events.

from statemachine import State, StateChart

class ResilientChart(StateChart):
    operational = State(initial=True)
    broken = State(final=True)

    do_work = operational.to(operational, on="risky_action")
    error_execution = operational.to(broken)

    def risky_action(self):
        raise RuntimeError("something went wrong")

sm = ResilientChart()
sm.send("do_work")
# The machine transitions to 'broken' because the error was caught and dispatched

python-statemachine performs structural validation at two stages:

1. Class definition time: Checks for structural issues like unreachable states or disconnected components. If validation fails, the class is not created and raises InvalidDefinition.
2. Instance creation time: Checks for callback resolution, boolean expression parsing, and other runtime-dependent logic.

Common validations include: exactly one initial state, no transitions from final states, unreachable states, trap states, and callback resolution.

from statemachine import State, StateChart
from statemachine.exceptions import InvalidDefinition

try:
    class Bad(StateChart):
        red = State(initial=True)
        green = State()
        hazard = State()
        cycle = red.to(green) | green.to(red)
        blink = hazard.to.itself()
except InvalidDefinition as e:
    print(e)
# Output: There are unreachable states. The statemachine graph should have a single component. Disconnected states: ['hazard']

By default, the declarative format uses a restricted evaluator for guards and expressions. To enable full Python capabilities—such as calling builtins (e.g., sum(), len()), using comprehensions, or using the script block to execute arbitrary Python statements that read/write model variables—you must pass trusted=True to the load function.

Warning: Only use trusted=True for documents you control, as it allows for arbitrary code execution.

>>> sc = load(
...     """
...     datamodel:
...       - {id: cart, expr: "[10, 25, 5]"}
...       - {id: total, expr: "0"}
...       - {id: tier, expr: "''"}
...     states:
...       pricing:
...         initial: true
...         enter:
...           - script: |
...               total = sum(cart)
...               tier = 'gold' if total >= 40 else 'silver'
...         transitions:
...           - {target: vip, cond: "tier == 'gold' and len(cart) >= 3"}
...           - {target: standard}
...       vip:
...         final: true
...       standard:
...         final: true
...     """,
...     format="yaml",
...     trusted=True,
...     validate=True,
... )
>>> sm = sc()
>>> sm.model.total, sm.model.tier
(40, 'gold')