3.0.0.md

StateMachine 3.0.0

Not released yet

What's new in 3.0.0

Statecharts are there! 🎉

Statecharts are a powerful extension to state machines, in a way to organize complex reactive systems as a hierarchical state machine. They extend the concept of state machines by adding two new kinds of states: parallel states and compound states.

Parallel states are states that can be active at the same time. They are useful for separating the state machine in multiple orthogonal state machines that can be active at the same time.

Compound states are states that have inner states. They are useful for breaking down complex state machines into multiple simpler ones.

The support for statecharts in this release follows the SCXML specification*, which is a W3C standard for statecharts notation. Adhering as much as possible to this specification ensures compatibility with other tools and platforms that also implement SCXML, but more important, sets a standard on the expected behaviour that the library should assume on various edge cases, enabling easier integration and interoperability in complex systems.

To verify the standard adoption, now the automated tests suite includes several .scxml testcases provided by the W3C group. Many thanks for this amazing work! Some of the tests are still failing, in such cases, we've added an xfail mark by including a test<number>.scxml.md markdown file with details of the execution output.

While these are exiting news for the library and our community, it also introduces several backwards incompatible changes. Due to the major version release, the new behaviour is assumed by default, but we put a lot of effort to minimize the changes needed in your codebase, and also introduced a few configuration options that you can enable to restore the old behaviour when possible. The following sections navigate to the new features and includes a migration guide.

Invoke

States can now spawn external work when entered and cancel it when exited, following the SCXML <invoke> semantics (similar to UML's do/ activity). Handlers run in a daemon thread (sync engine) or a thread executor wrapped in an asyncio Task (async engine). Invoke is a first-class callback group — convention naming (on_invoke_<state>), decorators (@state.invoke), inline callables, and the full SignatureAdapter dependency injection all work out of the box.

>>> from statemachine import State, StateChart

>>> class FetchMachine(StateChart):
...     loading = State(initial=True, invoke=lambda: {"status": "ok"})
...     ready = State(final=True)
...     done_invoke_loading = loading.to(ready)

>>> sm = FetchMachine()
>>> import time; time.sleep(0.1)  # wait for background invoke to complete
>>> "ready" in sm.configuration_values
True

Use {func}~statemachine.invoke.invoke_group to run multiple callables concurrently and wait for all results:

>>> from statemachine.invoke import invoke_group

>>> class BatchFetch(StateChart):
...     loading = State(initial=True, invoke=invoke_group(lambda: "a", lambda: "b"))
...     ready = State(final=True)
...     done_invoke_loading = loading.to(ready)
...
...     def on_enter_ready(self, data=None, **kwargs):
...         self.results = data

>>> sm = BatchFetch()
>>> import time; time.sleep(0.2)
>>> sm.results
['a', 'b']

Constructor keyword arguments are forwarded to initial state callbacks, so self-contained machines can receive context at creation time:

>>> class Greeter(StateChart):
...     idle = State(initial=True)
...     done = State(final=True)
...     idle.to(done)
...
...     def on_enter_idle(self, name=None, **kwargs):
...         self.greeting = f"Hello, {name}!"

>>> sm = Greeter(name="Alice")
>>> sm.greeting
'Hello, Alice!'

See {ref}invoke for full documentation.

Compound states

Compound states have inner child states. Use State.Compound to define them with Python class syntax — the class body becomes the state's children:

>>> from statemachine import State, StateChart

>>> class ShireToRoad(StateChart):
...     class shire(State.Compound):
...         bag_end = State(initial=True)
...         green_dragon = State()
...         visit_pub = bag_end.to(green_dragon)
...
...     road = State(final=True)
...     depart = shire.to(road)

>>> sm = ShireToRoad()
>>> set(sm.configuration_values) == {"shire", "bag_end"}
True

>>> sm.send("visit_pub")
>>> "green_dragon" in sm.configuration_values
True

>>> sm.send("depart")
>>> set(sm.configuration_values) == {"road"}
True

Entering a compound activates both the parent and its initial child. Exiting removes the parent and all descendants. See {ref}statecharts for full details.

Parallel states

Parallel states activate all child regions simultaneously. Use State.Parallel:

>>> from statemachine import State, StateChart

>>> class WarOfTheRing(StateChart):
...     validate_disconnected_states = False
...     class war(State.Parallel):
...         class frodos_quest(State.Compound):
...             shire = State(initial=True)
...             mordor = State(final=True)
...             journey = shire.to(mordor)
...         class aragorns_path(State.Compound):
...             ranger = State(initial=True)
...             king = State(final=True)
...             coronation = ranger.to(king)

>>> sm = WarOfTheRing()
>>> "shire" in sm.configuration_values and "ranger" in sm.configuration_values
True

>>> sm.send("journey")
>>> "mordor" in sm.configuration_values and "ranger" in sm.configuration_values
True

Events in one region don't affect others. See {ref}statecharts for full details.

History pseudo-states

The History pseudo-state records the configuration of a compound state when it is exited. Re-entering via the history state restores the previously active child. Supports both shallow (HistoryState()) and deep (HistoryState(deep=True)) history:

>>> from statemachine import HistoryState, State, StateChart

>>> class GollumPersonality(StateChart):
...     validate_disconnected_states = False
...     class personality(State.Compound):
...         smeagol = State(initial=True)
...         gollum = State()
...         h = HistoryState()
...         dark_side = smeagol.to(gollum)
...         light_side = gollum.to(smeagol)
...     outside = State()
...     leave = personality.to(outside)
...     return_via_history = outside.to(personality.h)

>>> sm = GollumPersonality()
>>> sm.send("dark_side")
>>> "gollum" in sm.configuration_values
True

>>> sm.send("leave")
>>> sm.send("return_via_history")
>>> "gollum" in sm.configuration_values
True

See {ref}statecharts for full details on shallow vs deep history.

Eventless (automatic) transitions

Transitions without an event trigger fire automatically when their guard condition is met:

>>> from statemachine import State, StateChart

>>> class BeaconChain(StateChart):
...     class beacons(State.Compound):
...         first = State(initial=True)
...         second = State()
...         last = State(final=True)
...         first.to(second)
...         second.to(last)
...     signal_received = State(final=True)
...     done_state_beacons = beacons.to(signal_received)

>>> sm = BeaconChain()
>>> set(sm.configuration_values) == {"signal_received"}
True

The entire eventless chain cascades in a single macrostep. See {ref}statecharts.

DoneData on final states

Final states can provide data to done.state handlers via the donedata parameter:

>>> from statemachine import Event, State, StateChart

>>> class QuestCompletion(StateChart):
...     class quest(State.Compound):
...         traveling = State(initial=True)
...         completed = State(final=True, donedata="get_result")
...         finish = traveling.to(completed)
...         def get_result(self):
...             return {"hero": "frodo", "outcome": "victory"}
...     epilogue = State(final=True)
...     done_state_quest = Event(quest.to(epilogue, on="capture_result"))
...     def capture_result(self, hero=None, outcome=None, **kwargs):
...         self.result = f"{hero}: {outcome}"

>>> sm = QuestCompletion()
>>> sm.send("finish")
>>> sm.result
'frodo: victory'

The done_state_ naming convention automatically registers the done.state.{suffix} form — no explicit id= needed. See {ref}done-state-convention for details.

Create state machine class from a dict definition

Dinamically create state machine classes by using create_machine_class_from_definition.

>>> from statemachine.io import create_machine_class_from_definition

>>> machine = create_machine_class_from_definition(
...     "TrafficLightMachine",
...     **{
...         "states": {
...             "green": {"initial": True, "on": {"change": [{"target": "yellow"}]}},
...             "yellow": {"on": {"change": [{"target": "red"}]}},
...             "red": {"on": {"change": [{"target": "green"}]}},
...         },
...     }
... )

>>> sm = machine()
>>> sm.green.is_active
True
>>> sm.send("change")
>>> sm.yellow.is_active
True

In(state) checks in condition expressions

Now a condition can check if the state machine current set of active states (a.k.a configuration) contains a state using the syntax cond="In('<state-id>')".

Preparing events

You can use the prepare_event method to add custom information that will be included in **kwargs to all other callbacks.

A not so usefull example:

>>> class ExampleStateMachine(StateMachine):
...     initial = State(initial=True)
...
...     loop = initial.to.itself()
...
...     def prepare_event(self):
...         return {"foo": "bar"}
...
...     def on_loop(self, foo):
...         return f"On loop: {foo}"
...

>>> sm = ExampleStateMachine()

>>> sm.loop()
'On loop: bar'

Event matching following SCXML spec

Now events matching follows the SCXML spec:

For example, a transition with an event attribute of "error foo" will match event names error, error.send, error.send.failed, etc. (or foo, foo.bar etc.) but would not match events named errors.my.custom, errorhandler.mistake, error.send or foobar.

An event designator consisting solely of * can be used as a wildcard matching any sequence of tokens, and thus any event.

Error handling with error.execution

When error_on_execution is enabled (default in StateChart), runtime exceptions during transitions are caught and result in an internal error.execution event. This follows the SCXML error handling specification.

A naming convention makes this easy to use: any event attribute starting with error_ automatically matches both the underscore and dot-notation forms:

>>> from statemachine import State, StateChart

>>> class MyChart(StateChart):
...     s1 = State("s1", initial=True)
...     error_state = State("error_state", final=True)
...
...     go = s1.to(s1, on="bad_action")
...     error_execution = s1.to(error_state)  # matches "error.execution" automatically
...
...     def bad_action(self):
...         raise RuntimeError("something went wrong")

>>> sm = MyChart()
>>> sm.send("go")
>>> sm.configuration == {sm.error_state}
True

The error object is available as error in handler kwargs. See {ref}error-execution for full details.

Delayed events

Specify an event to run in the near future using delay (in milliseconds). The engine will keep track of the execution time and only process the event when now > execution_time.

# Send with delay
sm.send("light_beacons", delay=500)  # fires after 500ms

# Define delay on the Event itself
light = Event(dark.to(lit), delay=100)

# Cancel a delayed event before it fires
sm.send("light_beacons", delay=5000, send_id="beacon_signal")
sm.cancel_event("beacon_signal")  # event is removed from the queue

New send() parameters

The send() method now accepts additional optional parameters:

• delay (float): Time in milliseconds before the event is processed.
• send_id (str): Identifier for the event, useful for cancelling delayed events.
• internal (bool): If True, the event is placed in the internal queue and processed in the current macrostep.

Existing calls to send() are fully backward compatible.

raise_() method

A new raise_() method sends events to the internal queue, equivalent to send(..., internal=True). Internal events are processed immediately within the current macrostep, before any external events.

sm.raise_("error_event")  # processed in the current macrostep

cancel_event() method

Cancel delayed events by their send_id:

sm.send("timeout", delay=5000, send_id="my_timer")
sm.cancel_event("my_timer")  # event is removed from the queue

is_terminated property

A new read-only property that returns True when the state machine has reached a final state and the engine is no longer running:

if sm.is_terminated:
    print("State machine has finished.")

Disable single graph component validation

Since SCXML don't require that all states should be reachable by transitions, we added a class-level flag validate_disconnected_states: bool = True that can be used to disable this validation.

It's already disabled when parsing SCXML files.

Typed models with Generic[TModel]

StateChart now supports a generic type parameter for the model, enabling full type inference and IDE autocompletion on sm.model:

>>> from statemachine import State, StateChart

>>> class MyModel:
...     name: str = ""
...     value: int = 0

>>> class MySM(StateChart["MyModel"]):
...     idle = State(initial=True)
...     active = State(final=True)
...     go = idle.to(active)

>>> sm = MySM(model=MyModel())
>>> sm.model.name
''

With this declaration, type checkers infer sm.model as MyModel (not Any), so accessing sm.model.name or sm.model.value gets full autocompletion and type safety. When no type parameter is given, StateChart defaults to StateChart[Any] for backward compatibility. See {ref}domain models for details.

Improved type checking with pyright

The library now supports pyright in addition to mypy. Type annotations have been improved throughout the codebase, and a catch-all __getattr__ that previously returned Any has been removed — type checkers can now detect misspelled attribute names and unresolved references on StateChart subclasses.

Weighted (probabilistic) transitions

A new contrib module statemachine.contrib.weighted provides weighted_transitions(), enabling probabilistic transition selection based on relative weights. This works entirely through the existing condition system — no engine changes required:

from statemachine.contrib.weighted import weighted_transitions

class GameCharacter(StateChart):
    standing = State(initial=True)
    shift_weight = State()
    adjust_hair = State()
    bang_shield = State()

    idle = weighted_transitions(
        standing,
        (shift_weight, 70),
        (adjust_hair, 20),
        (bang_shield, 10),
        seed=42,
    )

    finish = shift_weight.to(standing) | adjust_hair.to(standing) | bang_shield.to(standing)

See {ref}weighted-transitions for full documentation.

Class-level listener declarations

Listeners can now be declared at the class level using the listeners attribute, so they are automatically attached to every instance. The list accepts callables (classes, partial, lambdas) as factories that create a fresh listener per instance, or pre-built instances that are shared.

A setup() protocol allows factory-created listeners to receive runtime dependencies (DB sessions, Redis clients, etc.) via **kwargs forwarded from the SM constructor.

Inheritance is supported: child listeners are appended after parent listeners, unless listeners_inherit = False is set to replace them entirely.

See {ref}observers for full documentation.

Async concurrent event result routing

When multiple coroutines send events concurrently via asyncio.gather, each caller now receives its own event's result (or exception). Previously, only the first caller to acquire the processing lock would get a result — subsequent callers received None and exceptions could leak to the wrong caller.

This is implemented by attaching an asyncio.Future to each externally enqueued event in the async engine. See {ref}async for details.

Fixes #509.

Bugfixes in 3.0.0

• Fixes #XXX.

Misc in 3.0.0

TODO.

Known limitations

The following SCXML features are not yet implemented and are deferred to a future release:

• <invoke> — invoking external services or sub-machines from within a state
• HTTP and other external communication targets
• <finalize> — processing data returned from invoked services

These features are tracked for v3.1+.

For a step-by-step migration guide with before/after examples, see
{ref}`Upgrading from 2.x to 3.0 <Upgrading from 2.x to 3.0>`.

Backward incompatible changes in 3.0

Python compatibility in 3.0.0

We've dropped support for Python 3.7 and 3.8. If you need support for these versios use the 2.* series.

StateMachine 3.0.0 supports Python 3.9, 3.10, 3.11, 3.12, 3.13, and 3.14.

Non-RTC model removed

This option was deprecated on version 2.3.2. Now all new events are put on a queue before being processed.

Multiple current states

Due to the support of compound and parallel states, it's now possible to have multiple active states at the same time.

This introduces an impedance mismatch into the old public API, specifically, sm.current_state is deprecated and sm.current_state_value can returns a flat value if no compound state or a set instead.

To allow a smooth migration, these properties still work as before if there's no compound/parallel states in the state machine definition.

Old

    def current_state(self) -> "State":

New

    def current_state(self) -> "State | MutableSet[State]":

We strongly recomend using the new sm.configuration that has a stable API returning an OrderedSet on all cases:

    @property
    def configuration(self) -> OrderedSet["State"]:

Entering and exiting states

Previous versions performed an atomic update of the active state just after the execution of the transition on actions.

Now, we follow the SCXML spec:

To execute a microstep, the SCXML Processor MUST execute the transitions in the corresponding optimal enabled transition set. To execute a set of transitions, the SCXML Processor MUST first exit all the states in the transitions' exit set in exit order. It MUST then execute the executable content contained in the transitions in document order. It MUST then enter the states in the transitions' entry set in entry order.

This introduces backward-incompatible changes, as previously, the current_state was never empty, allowing queries on sm.current_state or sm.<any_state>.is_active even while executing an on transition action.

Now, by default, during a transition, all states in the exit set are exited first, performing the before and exit callbacks. The on callbacks are then executed in an intermediate state that contains only the states that will not be exited, which can be an empty set. Following this, the states in the enter set are entered, with enter callbacks executed for each state in document order, and finally, the after callbacks are executed with the state machine in the final new configuration.

We have added two new keyword arguments available only in the on callbacks to assist with queries that were performed against sm.current_state or active states using <state>.is_active:

• previous_configuration: OrderedSet[State]: Contains the set of states that were active before the microstep was taken.
• new_configuration: OrderedSet[State]: Contains the set of states that will be active after the microstep finishes.

Additionally, you can create a state machine instance by passing atomic_configuration_update=True (default False) to restore the old behavior. When set to False, the sm.configuration will be updated only once per microstep, just after the on callbacks with the new_configuration, the set of states that should be active after the microstep.

Consider this example that needs to be upgraded:

class ApprovalMachine(StateMachine):
    "A workflow"

    requested = State(initial=True)
    accepted = State()
    rejected = State()
    completed = State(final=True)

    validate = (
        requested.to(accepted, cond="is_ok") | requested.to(rejected) | accepted.to(completed)
    )
    retry = rejected.to(requested)

    def on_validate(self):
        if self.accepted.is_active and self.model.is_ok():
            return "congrats!"

The validate event is bound to several transitions, and the on_validate is expected to return congrats only when the state machine was with the accepted state active before the event occurs. In the old behavior, checking for accepted.is_active evaluates to True because the state were not exited before the on callback.

Due to the new behaviour, at the time of the on_validate call, the state machine configuration (a.k.a the current set of active states) is empty. So at this point in time accepted.is_active evaluates to False. To mitigate this case, now you can request one of the two new keyword arguments: previous_configuration and new_configration in on callbacks.

New way using previous_configuration:

def on_validate(self, previous_configuration):
    if self.accepted in previous_configuration and self.model.is_ok():
        return "congrats!"

add_observer() removed

The method add_observer, deprecated since v2.3.2, has been removed. Use add_listener instead.

TransitionNotAllowed exception changes

The TransitionNotAllowed exception now stores a configuration attribute (a MutableSet[State]) instead of a single state attribute, reflecting support for multiple active states. The event attribute can also be None.

Configuring the event without transition behaviour

The allow_event_without_transition was previously configured as an init parameter, now it's a class-level attribute.

Defaults to False in StateMachine class to preserve maximum backwards compatibility.

States.from_enum default use_enum_instance=True

The use_enum_instance parameter of States.from_enum now defaults to True (was False in 2.x). This means state values are the enum instances themselves, not their raw values.

If your code relies on raw enum values (e.g., integers), pass use_enum_instance=False explicitly.

Short registry names removed

State machine classes are now only registered by their fully-qualified name (qualname). The short-name lookup (by cls.__name__) that was deprecated since v0.8 has been removed.

If you use get_machine_cls() (e.g., via MachineMixin), make sure you pass the fully-qualified dotted path.

strict_states parameter removed

The strict_states class parameter (introduced in v2.2.0) has been removed and replaced by two independent class-level attributes that default to True:

• validate_trap_states: non-final states must have at least one outgoing transition.
• validate_final_reachability: when final states exist, all non-final states must have a path to at least one final state.

Migration:

• Remove strict_states=True — this is now the default behavior.
• Recommended: fix your state machine definition so that terminal states are marked final=True:

>>> from statemachine import State, StateChart

>>> class MySM(StateChart):
...     s1 = State(initial=True)
...     s2 = State(final=True)
...     go = s1.to(s2)

• If you intentionally have non-final trap states, replace strict_states=False with validate_trap_states = False and/or validate_final_reachability = False:

>>> class MySM(StateChart):
...     validate_trap_states = False
...     s1 = State(initial=True)
...     s2 = State()
...     go = s1.to(s2)