Config API Reference¶

class ConfigError(Exception):
    """Base exception for all config operations."""

class UndefinedValueError(ConfigError):
    """Raised when a required configuration value is missing."""

class Csv:
    """Parse comma-separated values with optional per-item casting.

    Example::

        Csv()("a, b, c")           # ["a", "b", "c"]
        Csv(cast=int)("1,2,3")     # [1, 2, 3]
        Csv(delimiter=";")("a;b")  # ["a", "b"]

    Args:
        cast: Callable applied to each item after splitting.
        delimiter: String to split on.
        strip: Characters to strip from each item. Use ``" %s"`` to strip
            whitespace around the format-string placeholder (default).
        post_process: Callable applied to the final list (e.g. ``tuple``).
    """

    def __init__(
        self,
        cast: Callable[[str], Any] = str,
        delimiter: str = ",",
        strip: str = " %s",
        post_process: Callable[[list[Any]], Any] = list,
    ) -> None:
        self.cast = cast
        self.delimiter = delimiter
        self.strip = strip
        self.post_process = post_process

    def __call__(self, value: str) -> Any:
        if isinstance(value, (list, tuple)):
            return self.post_process([self.cast(item) for item in value])
        parts = str(value).split(self.delimiter)
        result = []
        for part in parts:
            item = part.strip(self.strip.replace("%s", "")) if self.strip else part
            if item or not self.strip:
                result.append(self.cast(item))
        return self.post_process(result)

class Choices:
    """Validate that a value belongs to a fixed set.

    Example::

        Choices(["dev", "staging", "prod"])("dev")  # "dev"
        Choices([1, 2, 3], cast=int)("2")           # 2

    Args:
        choices: Allowed values (after casting).
        cast: Callable applied before validation.
    """

    def __init__(
        self,
        choices: Sequence[Any],
        cast: Callable[[str], Any] = str,
    ) -> None:
        self.choices = choices
        self.cast = cast

    def __call__(self, value: str) -> Any:
        casted = self.cast(value)
        if casted not in self.choices:
            raise ValueError(
                f"{casted!r} is not a valid choice. Allowed: {list(self.choices)}"
            )
        return casted

class Config:
    """Unified configuration loader with multi-source support.

    Sources are checked in priority order (highest first):

    1. Environment variables (``os.environ``), optionally prefixed
    2. ``.env`` file values (via sibling dotenv module)
    3. Config file values (JSON, JSONC, YAML, TOML, INI)
    4. Default value passed to ``get()``/``__call__()``

    Args:
        dotenv_path: Path to ``.env`` file. Use ``None`` to disable,
            or omit (default ``_AUTO``) to auto-discover by searching
            upward from the current directory.
        config_path: Path to a config file. Format is detected from the
            file extension.
        prefix: Prefix for environment variable lookups. For example,
            ``prefix="MYAPP_"`` means looking up ``"PORT"`` checks
            ``os.environ["MYAPP_PORT"]``.
        separator: Separator for nested key access. Default ``"__"``
            means ``"DATABASE__HOST"`` resolves to ``data["DATABASE"]["HOST"]``
            or ``data["database"]["host"]`` in config files.
        loaders: Override the file-format loader registry. Defaults to
            ``_UNSET`` (use built-in loaders with sibling auto-discovery
            for yaml/jsonc). Pass a ``dict`` mapping extensions (e.g.
            ``".yaml"``) to loader callables to replace the defaults.
            Stdlib loaders (json, toml, ini) are always available unless
            explicitly overridden.
        dotenv_loader: Override the dotenv loading mechanism. Defaults to
            ``_UNSET`` (auto-discover sibling ``dotenv`` module). Pass
            ``None`` to disable .env loading entirely, or a callable that
            returns ``(dotenv_values_fn, find_dotenv_fn)`` to inject a
            custom implementation.

    Example::

        cfg = Config(config_path="settings.yaml", prefix="MYAPP_")
        debug = cfg("DEBUG", default=False, cast=bool)
        db_host = cfg("DATABASE__HOST", default="localhost")
    """

    def __init__(
        self,
        *,
        dotenv_path: str | os.PathLike[str] | None | _Auto = _AUTO,
        config_path: str | os.PathLike[str] | None = None,
        prefix: str = "",
        separator: str = "__",
        loaders: dict[str, Callable[..., dict[str, Any]]] | None | _Unset = _UNSET,
        dotenv_loader: Callable[[], tuple[Callable[..., Any], Callable[..., Any]]]
        | None
        | _Unset = _UNSET,
    ) -> None:
        self._prefix = prefix
        self._separator = separator
        self._dotenv_data: dict[str, str | None] = {}
        self._config_data: dict[str, Any] = {}
        self._loaders: dict[str, Callable[..., dict[str, Any]]]

        # Resolve loader registry
        if isinstance(loaders, _Unset):
            self._loaders = _LOADERS
        elif loaders is None:
            self._loaders = {}
        else:
            self._loaders = loaders

        # Load .env file
        if dotenv_loader is None:
            pass  # .env explicitly disabled
        elif isinstance(dotenv_path, _Auto):
            try:
                if isinstance(dotenv_loader, _Unset):
                    dotenv_values, find_dotenv = _load_dotenv_helpers()
                else:
                    dotenv_values, find_dotenv = dotenv_loader()
            except ImportError:
                pass
            else:
                found = find_dotenv(usecwd=True)
                if found:
                    self._dotenv_data = dotenv_values(found)
        elif dotenv_path is not None:
            if isinstance(dotenv_loader, _Unset):
                dotenv_values, _find_dotenv = _load_dotenv_helpers()
            else:
                dotenv_values, _find_dotenv = dotenv_loader()
            self._dotenv_data = dotenv_values(str(dotenv_path))

        # Load config file
        if config_path is not None:
            self._load_config_file(config_path)

    def _load_config_file(self, path: str | os.PathLike[str]) -> None:
        """Load a config file based on its extension."""
        p = Path(path)
        ext = p.suffix.lower()
        loader = self._loaders.get(ext)
        if loader is None:
            raise ValueError(
                f"Unsupported config file format: {ext!r}. "
                f"Supported: {', '.join(sorted(self._loaders))}"
            )
        if loader in (_load_ini_file,):
            self._config_data = loader(p, separator=self._separator)
        else:
            self._config_data = loader(p)

    def _lookup(self, key: str) -> Any:
        """Look up a key across all sources in priority order.

        Returns the value if found, or *MISSING* if not found anywhere.
        """
        # 1. Environment variables (with prefix)
        env_key = f"{self._prefix}{key}" if self._prefix else key
        env_val = os.environ.get(env_key)
        if env_val is not None:
            return env_val

        # 2. .env file values (with prefix)
        dotenv_val = self._dotenv_data.get(env_key)
        if dotenv_val is not None:
            return dotenv_val

        # 3. Config file values (exact key first, then nested lookup)
        if self._config_data:
            # Exact key match (e.g. INI-flattened keys)
            if key in self._config_data:
                return self._config_data[key]
            # Nested lookup via separator splitting
            if self._separator in key:
                parts = key.split(self._separator)
                result = _deep_get(self._config_data, parts)
                if result is not MISSING:
                    return result

        return MISSING

    def get(
        self,
        key: str,
        *,
        default: Any = MISSING,
        cast: type | Callable[..., Any] | None = None,
    ) -> Any:
        """Retrieve a configuration value.

        Args:
            key: Configuration key to look up.
            default: Default value if key is not found. If not provided and
                the key is missing, ``UndefinedValueError`` is raised.
            cast: Type or callable to apply to the value. Built-in support
                for ``bool``, ``int``, ``float``, ``list``, ``tuple``.
                Also accepts ``Csv(...)`` and ``Choices(...)`` instances.

        Returns:
            The resolved and optionally cast configuration value.

        Raises:
            UndefinedValueError: If key is missing and no default is given.
        """
        value = self._lookup(key)

        if value is MISSING:
            if default is MISSING:
                raise UndefinedValueError(
                    f"{key!r} is not set and has no default value."
                )
            value = default

        return _apply_cast(value, cast)

    def __call__(
        self,
        key: str,
        *,
        default: Any = MISSING,
        cast: type | Callable[..., Any] | None = None,
    ) -> Any:
        """Shorthand for ``get()``. See :meth:`get` for details."""
        return self.get(key, default=default, cast=cast)

    def has(self, key: str) -> bool:
        """Check whether a key exists in any source."""
        return self._lookup(key) is not MISSING

    def as_dict(self) -> dict[str, Any]:
        """Return a merged view of all config sources (flattened).

        Lower-priority sources are merged first, so higher-priority
        sources overwrite them.
        """
        merged: dict[str, Any] = {}
        # 3. Config file (lowest priority)
        if self._config_data:
            merged.update(_flatten_dict(self._config_data, self._separator))
        # 2. .env file
        for k, v in self._dotenv_data.items():
            if v is not None:
                merged[k] = v
        # 1. Environment variables (only those matching prefix)
        if self._prefix:
            plen = len(self._prefix)
            for k, v in os.environ.items():
                if k.startswith(self._prefix):
                    merged[k[plen:]] = v
        else:
            merged.update(os.environ)
        return merged