Base

provide.foundation.state.base

TODO: Add module docstring.

Classes

ImmutableState

Base class for immutable state objects.

All state in Foundation should inherit from this to ensure immutability and provide consistent state management.

Functions

with_changes

with_changes(**changes: Any) -> ImmutableState

Create a new state instance with the specified changes.

Parameters:

Name	Type	Description	Default
**changes	Any	Field updates to apply	{}

Returns:

Type	Description
ImmutableState	New state instance with updated generation

Source code in provide/foundation/state/base.py

def with_changes(self, **changes: Any) -> ImmutableState:
    """Create a new state instance with the specified changes.

    Args:
        **changes: Field updates to apply

    Returns:
        New state instance with updated generation
    """
    # Increment generation for change tracking
    if "generation" not in changes:
        changes["generation"] = self.generation + 1

    # For attrs classes with slots, use attrs.evolve instead of __dict__
    import attrs

    return attrs.evolve(self, **changes)

StateMachine

StateMachine(initial_state: StateT)

Bases: Generic[StateT, EventT], ABC

Abstract base class for state machines.

Provides thread-safe state transitions with guards and actions.

Source code in provide/foundation/state/base.py

def __init__(self, initial_state: StateT) -> None:
    self._current_state = initial_state
    self._lock = threading.RLock()
    self._transitions: dict[tuple[StateT, EventT], StateTransition[StateT, EventT]] = {}
    self._state_history: list[tuple[float, StateT, EventT | None]] = []

    # Record initial state
    self._state_history.append((time.time(), initial_state, None))

Attributes

current_state property

current_state: StateT

Get the current state (thread-safe).

state_history property

state_history: list[tuple[float, StateT, EventT | None]]

Get the state transition history.

Functions

add_transition

add_transition(
    transition: StateTransition[StateT, EventT],
) -> None

Add a state transition to the machine.

Source code in provide/foundation/state/base.py

def add_transition(self, transition: StateTransition[StateT, EventT]) -> None:
    """Add a state transition to the machine."""
    key = (transition.from_state, transition.event)
    self._transitions[key] = transition

reset abstractmethod

reset() -> None

Reset the state machine to its initial state.

Source code in provide/foundation/state/base.py

@abstractmethod
def reset(self) -> None:
    """Reset the state machine to its initial state."""
    pass

transition

transition(event: EventT) -> bool

Attempt to transition to a new state based on the event.

Parameters:

Name	Type	Description	Default
event	EventT	Event that triggers the transition	required

Returns:

Type	Description
bool	True if transition was successful, False otherwise

Source code in provide/foundation/state/base.py

def transition(self, event: EventT) -> bool:
    """Attempt to transition to a new state based on the event.

    Args:
        event: Event that triggers the transition

    Returns:
        True if transition was successful, False otherwise
    """
    with self._lock:
        key = (self._current_state, event)
        transition = self._transitions.get(key)

        if not transition:
            return False

        if not transition.can_transition():
            return False

        # Execute transition
        old_state = self._current_state
        self._current_state = transition.to_state

        # Record transition
        self._state_history.append((time.time(), self._current_state, event))

        # Execute action (outside lock to avoid deadlocks)
        try:
            transition.execute_action()
        except Exception:
            # If action fails, revert state
            self._current_state = old_state
            self._state_history.pop()  # Remove failed transition
            return False

        return True

StateManager

Thread-safe manager for immutable state objects.

Provides atomic updates and version tracking for state objects.

Attributes

current_state property

current_state: ImmutableState

Get the current state (thread-safe).

generation property

generation: int

Get the current state generation.

Functions

add_observer

add_observer(
    observer: Callable[
        [ImmutableState, ImmutableState], None
    ],
) -> None

Add a state change observer.

Parameters:

Name	Type	Description	Default
observer	Callable[[ImmutableState, ImmutableState], None]	Function called with (old_state, new_state) on changes	required

Source code in provide/foundation/state/base.py

def add_observer(self, observer: Callable[[ImmutableState, ImmutableState], None]) -> None:
    """Add a state change observer.

    Args:
        observer: Function called with (old_state, new_state) on changes
    """
    with self._lock:
        self._observers.append(observer)

remove_observer

remove_observer(
    observer: Callable[
        [ImmutableState, ImmutableState], None
    ],
) -> None

Remove a state change observer.

Parameters:

Name	Type	Description	Default
observer	Callable[[ImmutableState, ImmutableState], None]	Observer function to remove	required

Source code in provide/foundation/state/base.py

def remove_observer(self, observer: Callable[[ImmutableState, ImmutableState], None]) -> None:
    """Remove a state change observer.

    Args:
        observer: Observer function to remove
    """
    with self._lock, contextlib.suppress(ValueError):
        self._observers.remove(observer)

replace_state

replace_state(new_state: ImmutableState) -> None

Replace the entire state object.

Parameters:

Name	Type	Description	Default
new_state	ImmutableState	New state to set	required

Source code in provide/foundation/state/base.py

def replace_state(self, new_state: ImmutableState) -> None:
    """Replace the entire state object.

    Args:
        new_state: New state to set
    """
    with self._lock:
        old_state = self._state
        self._state = new_state

        # Notify observers
        for observer in self._observers:
            with contextlib.suppress(Exception):
                observer(old_state, new_state)

update_state

update_state(**changes: Any) -> ImmutableState

Atomically update the state with the given changes.

Parameters:

Name	Type	Description	Default
**changes	Any	Field updates to apply	{}

Returns:

Type	Description
ImmutableState	New state instance

Source code in provide/foundation/state/base.py

def update_state(self, **changes: Any) -> ImmutableState:
    """Atomically update the state with the given changes.

    Args:
        **changes: Field updates to apply

    Returns:
        New state instance
    """
    with self._lock:
        old_state = self._state
        new_state = self._state.with_changes(**changes)
        self._state = new_state

        # Notify observers
        for observer in self._observers:
            with contextlib.suppress(Exception):
                observer(old_state, new_state)

        return new_state

StateTransition

Bases: Generic[StateT, EventT]

Represents a state transition in a state machine.

Functions

can_transition

can_transition() -> bool

Check if transition is allowed based on guard condition.

Source code in provide/foundation/state/base.py

def can_transition(self) -> bool:
    """Check if transition is allowed based on guard condition."""
    return self.guard() if self.guard else True

execute_action

execute_action() -> Any

Execute the transition action if present.

Source code in provide/foundation/state/base.py

def execute_action(self) -> Any:
    """Execute the transition action if present."""
    return self.action() if self.action else None