perf: eliminate for_instance() cache — store InstanceState in sm.__dict__

Remove the State descriptor protocol (__get__/__set__) and for_instance()
cache. InstanceState objects are now created eagerly in StateChart.__init__
and stored directly in sm.__dict__, making sm.<state_id> a plain dict
lookup instead of a descriptor call + cache lookup.

Configuration no longer holds a weakref to the machine; it receives a
dedicated instance_states dict and resolves active states via direct
dict lookup.

The __setattr__ guard on StateChart preserves the existing protection
against accidental state overriding. States whose id collides with an
event name are kept out of __dict__ to preserve Event descriptor
priority.

Fix type mismatches in engines/base.py and event_data.py that were
previously masked by the State.__set__ descriptor.

Author: fgmacedo

statemachine/configuration.py

@@ -1,8 +1,8 @@
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Dict
+from typing import Mapping
 from typing import MutableSet
-from weakref import ref
 
 from .exceptions import InvalidStateValue
 from .i18n import _
@@ -12,7 +12,6 @@
 
 if TYPE_CHECKING:
     from .state import State
-    from .statemachine import StateChart
 
 
 class Configuration:
@@ -25,28 +24,25 @@ class Configuration:
     """
 
     __slots__ = (
-        "_machine_ref",
+        "_instance_states",
         "_model",
         "_state_field",
         "_states_map",
-        "_for_instance",
         "_cached",
         "_cached_value",
     )
 
     def __init__(
         self,
-        machine: "StateChart",
+        instance_states: "Mapping[str, State]",
         model: Any,
         state_field: str,
         states_map: "Dict[Any, State]",
-        for_instance_cache: "Dict[State, State]",
     ):
-        self._machine_ref: "ref[StateChart]" = ref(machine)
+        self._instance_states = instance_states
         self._model = model
         self._state_field = state_field
         self._states_map = states_map
-        self._for_instance = for_instance_cache
         self._cached: "OrderedSet[State] | None" = None
         self._cached_value: Any = _SENTINEL
 
@@ -83,20 +79,11 @@ def states(self) -> "OrderedSet[State]":
         if csv is None:
             return OrderedSet()
 
-        machine = self._machine_ref()
-        assert machine is not None
-
+        instance_states = self._instance_states
         if not isinstance(csv, MutableSet):
-            result = OrderedSet(
-                [self._states_map[csv].for_instance(machine=machine, cache=self._for_instance)]
-            )
+            result = OrderedSet([instance_states[self._states_map[csv].id]])
         else:
-            result = OrderedSet(
-                [
-                    self._states_map[v].for_instance(machine=machine, cache=self._for_instance)
-                    for v in csv
-                ]
-            )
+            result = OrderedSet([instance_states[self._states_map[v].id] for v in csv])
 
         self._cached = result
         self._cached_value = csv

statemachine/engines/base.py

@@ -804,7 +804,8 @@ def add_descendant_states_to_enter(  # noqa: C901
                 )
 
                 for transition in state.transitions:
-                    info_history = StateTransition(transition=transition, state=transition.target)
+                    target = cast(State, transition.target)
+                    info_history = StateTransition(transition=transition, state=target)
                     default_history_content[parent_id].append(info_history)
                     self.add_descendant_states_to_enter(
                         info_history,
@@ -813,7 +814,8 @@ def add_descendant_states_to_enter(  # noqa: C901
                         default_history_content,
                     )  # noqa: E501
                 for transition in state.transitions:
-                    info_history = StateTransition(transition=transition, state=transition.target)
+                    target = cast(State, transition.target)
+                    info_history = StateTransition(transition=transition, state=target)
 
                     self.add_ancestor_states_to_enter(
                         info_history,

statemachine/event_data.py

@@ -63,8 +63,8 @@ class EventData:
     source: "State" = field(init=False)
     """The :ref:`State` which :ref:`statemachine` was in when the Event started."""
 
-    target: "State" = field(init=False)
-    """The destination :ref:`State` of the :ref:`transition`."""
+    target: "State | None" = field(init=False)
+    """The destination :ref:`State` of the :ref:`transition`, or ``None`` for targetless."""
 
     def __post_init__(self):
         self.state = self.transition.source

statemachine/state.py

@@ -1,7 +1,6 @@
 from enum import Enum
 from typing import TYPE_CHECKING
 from typing import Any
-from typing import Dict
 from typing import Generator
 from typing import List
 from typing import cast
@@ -12,7 +11,6 @@
 from .callbacks import CallbackSpecList
 from .event import _expand_event_id
 from .exceptions import InvalidDefinition
-from .exceptions import StateMachineError
 from .i18n import _
 from .invoke import normalize_invoke_callbacks
 from .transition import Transition
@@ -295,22 +293,6 @@ def __repr__(self):
     def __str__(self):
         return self.name
 
-    def __get__(self, machine, owner):
-        if machine is None:
-            return self
-        return self.for_instance(machine=machine, cache=machine._config._for_instance)
-
-    def __set__(self, instance, value):
-        raise StateMachineError(
-            _("State overriding is not allowed. Trying to add '{}' to {}").format(value, self.id)
-        )
-
-    def for_instance(self, machine: "StateChart", cache: Dict["State", "State"]) -> "State":
-        if self not in cache:
-            cache[self] = InstanceState(self, machine)
-
-        return cache[self]
-
     @property
     def id(self) -> str:
         return self._id

statemachine/statemachine.py

@@ -25,12 +25,14 @@
 from .event_data import TriggerData
 from .exceptions import InvalidDefinition
 from .exceptions import InvalidStateValue
+from .exceptions import StateMachineError
 from .exceptions import TransitionNotAllowed
 from .factory import StateMachineMetaclass
 from .graph import iterate_states_and_transitions
 from .i18n import _
 from .model import Model
 from .signature import SignatureAdapter
+from .state import InstanceState
 from .utils import run_async_from_sync
 
 if TYPE_CHECKING:
@@ -151,12 +153,18 @@ def __init__(
             [start_value] if start_value is not None else list(self.start_configuration_values)
         )
         self._callbacks = CallbacksRegistry()
+        instance_states: Dict[str, Any] = {}
+        events = self.__class__._events
+        for state in self.states_map.values():
+            ist = InstanceState(state, self)
+            instance_states[state.id] = ist
+            if state.id not in events:
+                vars(self)[state.id] = ist
         self._config = Configuration(
-            machine=self,
+            instance_states=instance_states,
             model=self.model,
             state_field=self.state_field,
             states_map=self.states_map,
-            for_instance_cache={},
         )
         self._listeners: Dict[int, Any] = {}
         """Listeners that provides attributes to be used as callbacks."""
@@ -212,6 +220,17 @@ def _processing_loop(self, caller_future: "Any | None" = None) -> Any:
             return result
         return run_async_from_sync(result)
 
+    def __setattr__(self, name, value):
+        # Fast path: internal/private attributes are never state IDs.
+        if not name.startswith("_") and name in self.__class__.states_map:
+            if not isinstance(value, InstanceState):
+                raise StateMachineError(
+                    _("State overriding is not allowed. Trying to add '{}' to {}").format(
+                        value, name
+                    )
+                )
+        super().__setattr__(name, value)
+
     def __repr__(self):
         configuration_ids = [s.id for s in self.configuration]
         return (
@@ -220,7 +239,7 @@ def __repr__(self):
         )
 
     def __getstate__(self):
-        state = self.__dict__.copy()
+        state = {k: v for k, v in self.__dict__.items() if not isinstance(v, InstanceState)}
         del state["_callbacks"]
         del state["_config"]
         del state["_engine"]
@@ -230,12 +249,18 @@ def __setstate__(self, state: Dict[str, Any]) -> None:
         listeners = state.pop("_listeners")
         self.__dict__.update(state)  # type: ignore[attr-defined]
         self._callbacks = CallbacksRegistry()
+        instance_states: Dict[str, Any] = {}
+        events = self.__class__._events
+        for sm_state in self.states_map.values():
+            ist = InstanceState(sm_state, self)
+            instance_states[sm_state.id] = ist
+            if sm_state.id not in events:
+                vars(self)[sm_state.id] = ist
         self._config = Configuration(
-            machine=self,
+            instance_states=instance_states,
             model=self.model,
             state_field=self.state_field,
             states_map=self.states_map,
-            for_instance_cache={},
         )
         self._listeners = {}
 

tests/test_configuration.py

@@ -33,8 +33,8 @@ def test_set_empty_configuration(self):
 
     def test_set_multi_element_configuration(self):
         sm = ParallelSM()
-        s1_inst = ParallelSM.s1.for_instance(machine=sm, cache=sm._config._for_instance)
-        s2_inst = ParallelSM.s2.for_instance(machine=sm, cache=sm._config._for_instance)
+        s1_inst = sm.s1
+        s2_inst = sm.s2
 
         sm.configuration = OrderedSet([s1_inst, s2_inst])
         assert isinstance(sm.current_state_value, OrderedSet)
@@ -55,8 +55,8 @@ def test_discard_nonmatching_scalar(self):
 class TestConfigurationCurrentState:
     def test_current_state_with_multiple_active_states(self):
         sm = ParallelSM()
-        s1_inst = ParallelSM.s1.for_instance(machine=sm, cache=sm._config._for_instance)
-        s2_inst = ParallelSM.s2.for_instance(machine=sm, cache=sm._config._for_instance)
+        s1_inst = sm.s1
+        s2_inst = sm.s2
         sm.configuration = OrderedSet([s1_inst, s2_inst])
 
         with warnings.catch_warnings():