Config API Reference

Multi-source configuration management.

Config Class

dspu.config.config.Config

Main configuration loader with Pydantic validation.

Loads configuration from multiple sources with priority ordering (later sources override earlier ones) and validates using Pydantic models.

Example

from pydantic import BaseModel from dspu.config import Config, ConfigSource

class AppConfig(BaseModel): ... debug: bool = False ... port: int = 8000 ... config = Config.load( ... AppConfig, ... sources=[ ... ConfigSource.file("config.yaml"), ... ConfigSource.env(prefix="APP_"), ... ] ... ) config.port 8000

WatchedConfig

dspu.config.watched.WatchedConfig

WatchedConfig(
    config_class: type[T],
    sources: Sequence[ConfigSource],
    reload_interval: float = 5.0,
)

Bases: Generic[T]

Configuration that watches for file changes and automatically reloads.

This class wraps a configuration and monitors the source files for changes. When a change is detected, the configuration is automatically reloaded and registered callbacks are invoked.

Example

watched = WatchedConfig.load( ... AppConfig, ... source=FileSource("config.yaml"), ... reload_interval=5.0, ... )

Use current config

app = Application(watched.current)

Register callback

watched.on_change(lambda old, new: app.reconfigure(new))

When config.yaml changes, app.reconfigure() is called

Stop watching when done

watched.stop()

Note

• Only works with FileSource (needs file to watch)
• Callbacks are called in a background thread
• Always call stop() when done or use as context manager

Initialize watched configuration.

Parameters:

Name	Type	Description	Default
config_class	type[T]	Pydantic model class for validation.	required
sources	Sequence[ConfigSource]	Configuration sources to watch.	required
reload_interval	float	How often to check for changes (seconds).	5.0

Raises:

Type	Description
ConfigurationError	If no file sources are provided.

Source code in src/dspu/config/watched.py

def __init__(
    self,
    config_class: type[T],
    sources: Sequence[ConfigSource],
    reload_interval: float = 5.0,
) -> None:
    """Initialize watched configuration.

    Args:
        config_class: Pydantic model class for validation.
        sources: Configuration sources to watch.
        reload_interval: How often to check for changes (seconds).

    Raises:
        ConfigurationError: If no file sources are provided.
    """
    self._config_class = config_class
    self._sources = list(sources)
    self._reload_interval = reload_interval

    # Extract file paths to watch
    self._watched_files: list[Path] = []
    for source in sources:
        if isinstance(source, FileSource) and source.path.exists():
            self._watched_files.append(source.path)

    if not self._watched_files:
        raise ConfigurationError(
            "WatchedConfig requires at least one FileSource",
            suggestion="Add a FileSource to your sources list",
        )

    # Load initial configuration
    self._current = Config.load(config_class, sources)
    self._lock = threading.RLock()
    self._callbacks: list[Callable[[T, T], None]] = []

    # File modification times
    self._mtimes: dict[Path, float] = {}
    for path in self._watched_files:
        self._mtimes[path] = path.stat().st_mtime

    # Background watcher thread
    self._stop_event = threading.Event()
    self._watcher_thread = threading.Thread(
        target=self._watch_loop,
        daemon=True,
        name="config-watcher",
    )
    self._watcher_thread.start()

Configuration Sources

dspu.config.sources.FileSource

FileSource(
    path: str | Path,
    *,
    format: str | None = None,
    required: bool = True,
)

Bases: ConfigSource

Configuration source from a file.

Supports multiple formats: - JSON (.json) - YAML (.yaml, .yml) - TOML (.toml) - HOCON (.conf, .hocon) - ENV (.env)

Format is auto-detected from file extension or can be specified explicitly.

Example

source = FileSource("config.yaml") config = source.load()

Explicit format

source = FileSource("settings.txt", format="json") config = source.load()

Initialize file source.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to configuration file.	required
format	str | None	Format override (json, yaml, toml, hocon, env). If None, auto-detect from extension.	None
required	bool	If True, raise error if file doesn't exist.	True

Source code in src/dspu/config/sources.py

def __init__(
    self,
    path: str | Path,
    *,
    format: str | None = None,  # noqa: A002
    required: bool = True,
) -> None:
    """Initialize file source.

    Args:
        path: Path to configuration file.
        format: Format override (json, yaml, toml, hocon, env).
               If None, auto-detect from extension.
        required: If True, raise error if file doesn't exist.
    """
    self.path = Path(path)
    self.format = format
    self.required = required

Functions

load

load() -> dict[str, Any]

Load configuration from file.

Returns:

Type	Description
dict[str, Any]	Configuration dictionary.

Raises:

Type	Description
ConfigurationError	If file cannot be read or parsed.

Source code in src/dspu/config/sources.py

def load(self) -> dict[str, Any]:
    """Load configuration from file.

    Returns:
        Configuration dictionary.

    Raises:
        ConfigurationError: If file cannot be read or parsed.
    """
    if not self.path.exists():
        if self.required:
            raise ConfigurationError(
                f"Configuration file not found: {self.path}",
                field="path",
                suggestion=f"Create the file at {self.path}",
            )
        return {}

    # Determine format
    fmt = self.format or self._detect_format()

    try:
        content = self.path.read_text()

        if fmt == "json":
            return cast(dict[str, Any], json.loads(content))
        if fmt in ("yaml", "yml"):
            return cast(dict[str, Any], yaml.safe_load(content) or {})
        if fmt == "toml":
            return tomllib.loads(content)
        if fmt in ("hocon", "conf"):
            return cast(dict[str, Any], ConfigFactory.parse_string(content))
        if fmt == "env":
            # dotenv returns a dict directly
            return dict(dotenv_values(self.path))
        raise ConfigurationError(
            f"Unsupported format: {fmt}",
            field="format",
            suggestion="Supported formats: json, yaml, toml, hocon, env",
        )

    except json.JSONDecodeError as e:
        raise ConfigurationError(
            f"Invalid JSON in {self.path}: {e}",
            field="format",
            suggestion="Check JSON syntax",
        ) from e
    except yaml.YAMLError as e:
        raise ConfigurationError(
            f"Invalid YAML in {self.path}: {e}",
            field="format",
            suggestion="Check YAML syntax",
        ) from e
    except tomllib.TOMLDecodeError as e:
        raise ConfigurationError(
            f"Invalid TOML in {self.path}: {e}",
            field="format",
            suggestion="Check TOML syntax",
        ) from e
    except Exception as e:
        raise ConfigurationError(
            f"Failed to load config from {self.path}: {e}",
            suggestion="Check file format and permissions",
        ) from e

dspu.config.sources.EnvSource

EnvSource(
    *,
    prefix: str = "",
    separator: str = "__",
    lowercase: bool = True,
)

Bases: ConfigSource

Configuration source from environment variables.

Supports: - Optional prefix filtering - Nested keys using underscores or double underscores - Type conversion (numbers, booleans)

Example

With prefix: APP_DEBUG=true, APP_PORT=8000

source = EnvSource(prefix="APP_") config = source.load() config

With nested keys: APP__DATABASE__HOST=localhost

source = EnvSource(prefix="APP__", separator="__") config = source.load() config {'database': {'host': 'localhost'}}

Initialize environment source.

Parameters:

Name	Type	Description	Default
prefix	str	Only load variables starting with this prefix.	''
separator	str	Separator for nested keys (default: double underscore).	'__'
lowercase	bool	Convert keys to lowercase.	True

Source code in src/dspu/config/sources.py

def __init__(
    self,
    *,
    prefix: str = "",
    separator: str = "__",
    lowercase: bool = True,
) -> None:
    """Initialize environment source.

    Args:
        prefix: Only load variables starting with this prefix.
        separator: Separator for nested keys (default: double underscore).
        lowercase: Convert keys to lowercase.
    """
    self.prefix = prefix
    self.separator = separator
    self.lowercase = lowercase

Functions

load

load() -> dict[str, Any]

Load configuration from environment variables.

Returns:

Type	Description
dict[str, Any]	Configuration dictionary with nested structure if separator is used.

Source code in src/dspu/config/sources.py

def load(self) -> dict[str, Any]:
    """Load configuration from environment variables.

    Returns:
        Configuration dictionary with nested structure if separator is used.
    """
    config: dict[str, Any] = {}

    for key, value in os.environ.items():
        if not key.startswith(self.prefix):
            continue

        # Remove prefix
        clean_key = key[len(self.prefix) :]

        # Convert to lowercase if requested
        if self.lowercase:
            clean_key = clean_key.lower()

        # Handle nested keys
        if self.separator in clean_key:
            parts = clean_key.split(self.separator)
            self._set_nested(config, parts, value)
        else:
            config[clean_key] = value

    return config

dspu.config.sources.DictSource

DictSource(data: dict[str, Any])

Bases: ConfigSource

Configuration source from a Python dictionary.

Useful for testing and programmatic configuration.

Example

source = DictSource({"debug": True, "port": 8000}) config = source.load() config["debug"] True

Initialize dictionary source.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Configuration dictionary.	required

Source code in src/dspu/config/sources.py

def __init__(self, data: dict[str, Any]) -> None:
    """Initialize dictionary source.

    Args:
        data: Configuration dictionary.
    """
    self.data = data

Functions

load

load() -> dict[str, Any]

Load configuration from dictionary.

Source code in src/dspu/config/sources.py

def load(self) -> dict[str, Any]:
    """Load configuration from dictionary."""
    return self.data.copy()

Usage

from dspu.config import Config, WatchedConfig

# Load from file
config = Config.from_file("config.yaml")
db_host = config.get("database.host")

# Merge from environment
config.merge_from_env(prefix="APP_")

# Watch for changes
watched = WatchedConfig.from_file("config.yaml")
watched.start_watching()

See Also

• Config Examples
• Config User Guide