Base

🤖 AI-Generated Content

This documentation was generated with AI assistance and is still being audited. Some, or potentially a lot, of this information may be inaccurate. Learn more.

provide.foundation.state.base

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."""

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