Config

provide.foundation.state.config

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

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)