State

provide.foundation.state

TODO: Add module docstring.

Classes

ConfigManager

Thread-safe manager for versioned configurations.

Provides atomic updates and change tracking for configurations.

Functions

add_change_listener

add_change_listener(
    name: str,
    listener: Callable[
        [ImmutableState, ImmutableState], None
    ],
) -> None

Add a change listener for a configuration.

Parameters:

Name	Type	Description	Default
name	str	Configuration name	required
listener	Callable[[ImmutableState, ImmutableState], None]	Function called when configuration changes	required

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

def add_change_listener(
    self, name: str, listener: Callable[[ImmutableState, ImmutableState], None]
) -> None:
    """Add a change listener for a configuration.

    Args:
        name: Configuration name
        listener: Function called when configuration changes
    """
    with self._lock:
        if name not in self._change_listeners:
            self._change_listeners[name] = []
        self._change_listeners[name].append(listener)

clear_all

clear_all() -> None

Clear all configurations.

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

def clear_all(self) -> None:
    """Clear all configurations."""
    with self._lock:
        self._configs.clear()
        self._change_listeners.clear()

get_config

get_config(name: str) -> VersionedConfig | None

Get a configuration by name.

Parameters:

Name	Type	Description	Default
name	str	Configuration name	required

Returns:

Type	Description
VersionedConfig | None	Configuration instance or None if not found

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

def get_config(self, name: str) -> VersionedConfig | None:
    """Get a configuration by name.

    Args:
        name: Configuration name

    Returns:
        Configuration instance or None if not found
    """
    with self._lock:
        manager = self._configs.get(name)
        if manager:
            state = manager.current_state
            return state if isinstance(state, VersionedConfig) else None
        return None

get_config_generation

get_config_generation(name: str) -> int | None

Get the current generation of a configuration.

Parameters:

Name	Type	Description	Default
name	str	Configuration name	required

Returns:

Type	Description
int | None	Configuration generation or None if not found

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

def get_config_generation(self, name: str) -> int | None:
    """Get the current generation of a configuration.

    Args:
        name: Configuration name

    Returns:
        Configuration generation or None if not found
    """
    config = self.get_config(name)
    return config.generation if config else None

get_config_value

get_config_value(
    name: str, key: str, default: Any = None
) -> Any

Get a single configuration value.

Parameters:

Name	Type	Description	Default
name	str	Configuration name	required
key	str	Configuration key	required
default	Any	Default value if not found	None

Returns:

Type	Description
Any	Configuration value or default

Raises:

Type	Description
KeyError	If configuration not found

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

def get_config_value(self, name: str, key: str, default: Any = None) -> Any:
    """Get a single configuration value.

    Args:
        name: Configuration name
        key: Configuration key
        default: Default value if not found

    Returns:
        Configuration value or default

    Raises:
        KeyError: If configuration not found
    """
    config = self.get_config(name)
    if not config:
        raise KeyError(f"Configuration '{name}' not found")
    return config.get(key, default)

list_configs

list_configs() -> list[str]

List all registered configuration names.

Returns:

Type	Description
list[str]	List of configuration names

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

def list_configs(self) -> list[str]:
    """List all registered configuration names.

    Returns:
        List of configuration names
    """
    with self._lock:
        return list(self._configs.keys())

register_config

register_config(config: VersionedConfig) -> None

Register a new configuration.

Parameters:

Name	Type	Description	Default
config	VersionedConfig	Configuration to register	required

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

def register_config(self, config: VersionedConfig) -> None:
    """Register a new configuration.

    Args:
        config: Configuration to register
    """
    with self._lock:
        if config.config_name in self._configs:
            raise ValueError(f"Configuration '{config.config_name}' already registered")

        manager = StateManager(state=config)
        self._configs[config.config_name] = manager
        self._change_listeners[config.config_name] = []

        # Add observer for change notifications
        manager.add_observer(self._notify_listeners)

remove_change_listener

remove_change_listener(
    name: str,
    listener: Callable[
        [ImmutableState, ImmutableState], None
    ],
) -> None

Remove a change listener for a configuration.

Parameters:

Name	Type	Description	Default
name	str	Configuration name	required
listener	Callable[[ImmutableState, ImmutableState], None]	Listener function to remove	required

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

def remove_change_listener(
    self, name: str, listener: Callable[[ImmutableState, ImmutableState], None]
) -> None:
    """Remove a change listener for a configuration.

    Args:
        name: Configuration name
        listener: Listener function to remove
    """
    with self._lock:
        listeners = self._change_listeners.get(name, [])
        with contextlib.suppress(ValueError):
            listeners.remove(listener)

reset_config

reset_config(name: str) -> None

Reset a configuration to its initial state.

Parameters:

Name	Type	Description	Default
name	str	Configuration name	required

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

def reset_config(self, name: str) -> None:
    """Reset a configuration to its initial state.

    Args:
        name: Configuration name
    """
    with self._lock:
        manager = self._configs.get(name)
        if manager:
            # Find the root configuration (generation 0)
            current = manager.current_state
            if hasattr(current, "config_name"):
                initial_config = VersionedConfig(
                    config_name=current.config_name,
                    data={},
                    generation=0,
                    created_at=time.time(),
                )
                manager.replace_state(initial_config)

set_config_value

set_config_value(
    name: str, key: str, value: Any
) -> VersionedConfig

Set a single configuration value.

Parameters:

Name	Type	Description	Default
name	str	Configuration name	required
key	str	Configuration key	required
value	Any	Configuration value	required

Returns:

Type	Description
VersionedConfig	Updated configuration instance

Raises:

Type	Description
KeyError	If configuration not found

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

def set_config_value(self, name: str, key: str, value: Any) -> VersionedConfig:
    """Set a single configuration value.

    Args:
        name: Configuration name
        key: Configuration key
        value: Configuration value

    Returns:
        Updated configuration instance

    Raises:
        KeyError: If configuration not found
    """
    with self._lock:
        manager = self._configs.get(name)
        if not manager:
            raise KeyError(f"Configuration '{name}' not found")

        current_config = manager.current_state
        if not isinstance(current_config, VersionedConfig):
            raise TypeError(f"Expected VersionedConfig, got {type(current_config)}")
        new_config = current_config.set(key, value)
        manager.replace_state(new_config)
        return new_config

update_config

update_config(name: str, **updates: Any) -> VersionedConfig

Update a configuration with new values.

Parameters:

Name	Type	Description	Default
name	str	Configuration name	required
**updates	Any	Key-value pairs to update	{}

Returns:

Type	Description
VersionedConfig	Updated configuration instance

Raises:

Type	Description
KeyError	If configuration not found

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

def update_config(self, name: str, **updates: Any) -> VersionedConfig:
    """Update a configuration with new values.

    Args:
        name: Configuration name
        **updates: Key-value pairs to update

    Returns:
        Updated configuration instance

    Raises:
        KeyError: If configuration not found
    """
    with self._lock:
        manager = self._configs.get(name)
        if not manager:
            raise KeyError(f"Configuration '{name}' not found")

        current_config = manager.current_state
        if not isinstance(current_config, VersionedConfig):
            raise TypeError(f"Expected VersionedConfig, got {type(current_config)}")
        new_config = current_config.update(updates)
        manager.replace_state(new_config)
        return new_config

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

VersionedConfig

Bases: ImmutableState

Immutable configuration with generation tracking.

All Foundation configurations should inherit from this to ensure immutability and proper change tracking.

Functions

get

get(key: str, default: Any = None) -> Any

Get a configuration value.

Parameters:

Name	Type	Description	Default
key	str	Configuration key	required
default	Any	Default value if key not found	None

Returns:

Type	Description
Any	Configuration value or default

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

def get(self, key: str, default: Any = None) -> Any:
    """Get a configuration value.

    Args:
        key: Configuration key
        default: Default value if key not found

    Returns:
        Configuration value or default
    """
    return self.data.get(key, default)

merge

merge(other: VersionedConfig) -> VersionedConfig

Merge with another configuration.

Parameters:

Name	Type	Description	Default
other	VersionedConfig	Configuration to merge with	required

Returns:

Type	Description
VersionedConfig	New configuration instance with merged data

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

def merge(self, other: VersionedConfig) -> VersionedConfig:
    """Merge with another configuration.

    Args:
        other: Configuration to merge with

    Returns:
        New configuration instance with merged data
    """
    new_data = {**self.data, **other.data}
    return self.with_changes(
        data=new_data,
        parent_generation=max(self.generation, other.generation),
    )

remove

remove(key: str) -> VersionedConfig

Create a new config with the specified key removed.

Parameters:

Name	Type	Description	Default
key	str	Configuration key to remove	required

Returns:

Type	Description
VersionedConfig	New configuration instance

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

def remove(self, key: str) -> VersionedConfig:
    """Create a new config with the specified key removed.

    Args:
        key: Configuration key to remove

    Returns:
        New configuration instance
    """
    new_data = {k: v for k, v in self.data.items() if k != key}
    return self.with_changes(
        data=new_data,
        parent_generation=self.generation,
    )

set

set(key: str, value: Any) -> VersionedConfig

Create a new config with the specified key-value pair.

Parameters:

Name	Type	Description	Default
key	str	Configuration key	required
value	Any	Configuration value	required

Returns:

Type	Description
VersionedConfig	New configuration instance

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

def set(self, key: str, value: Any) -> VersionedConfig:
    """Create a new config with the specified key-value pair.

    Args:
        key: Configuration key
        value: Configuration value

    Returns:
        New configuration instance
    """
    new_data = {**self.data, key: value}
    return self.with_changes(
        data=new_data,
        parent_generation=self.generation,
    )

update

update(updates: dict[str, Any]) -> VersionedConfig

Create a new config with multiple updates.

Parameters:

Name	Type	Description	Default
updates	dict[str, Any]	Dictionary of key-value pairs to update	required

Returns:

Type	Description
VersionedConfig	New configuration instance

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

def update(self, updates: dict[str, Any]) -> VersionedConfig:
    """Create a new config with multiple updates.

    Args:
        updates: Dictionary of key-value pairs to update

    Returns:
        New configuration instance
    """
    new_data = {**self.data, **updates}
    return self.with_changes(
        data=new_data,
        parent_generation=self.generation,
    )

with_changes

with_changes(**changes: Any) -> VersionedConfig

Create a new state instance with the specified changes.

Parameters:

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

Returns:

Type	Description
VersionedConfig	New state instance with updated generation

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

def with_changes(self, **changes: Any) -> VersionedConfig:
    """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)