infrasynth-backend-kit/infrasynth/configs/registry.py

"""Typed configuration definitions and coercion for ``infrasynth.configs``.

A :class:`ConfigDefinition` is the schema for a single configuration key (its
type, default, grouping, and whether it is a secret). Definitions are registered
by the kit and by consuming apps — either imperatively via
:meth:`ConfigRegistry.register` in ``apps.py:ready()`` or declaratively through
``INFRASYNTH_CONFIGS["DEFINITIONS"]``.

All coercion failures raise :class:`~infrasynth.shared.exceptions.ValidationAppError`
with code ``VALIDATION_CONFIG_INVALID`` so the API envelope stays consistent and
callers never have to catch a bare ``ValueError``.
"""

from __future__ import annotations

import decimal
import json
import re
from dataclasses import dataclass
from enum import StrEnum
from typing import Any

from django.utils.module_loading import import_string

from infrasynth.shared.exceptions import ValidationAppError

__all__ = ["ConfigType", "ConfigDefinition", "ConfigRegistry"]

INVALID_CODE = "VALIDATION_CONFIG_INVALID"

_DURATION_RE = re.compile(r"^\s*(?P<amount>\d+(?:\.\d+)?)\s*(?P<unit>[a-zA-Z]*)\s*$")
_DURATION_UNITS = {
    "": 1,
    "s": 1,
    "sec": 1,
    "secs": 1,
    "m": 60,
    "min": 60,
    "mins": 60,
    "h": 3600,
    "hr": 3600,
    "hrs": 3600,
    "d": 86400,
    "day": 86400,
    "days": 86400,
}


def _invalid(key: str, issue: str) -> ValidationAppError:
    return ValidationAppError(
        f"Invalid value for configuration key '{key}'.",
        code=INVALID_CODE,
        details=[{"field": key, "issue": issue}],
    )


class ConfigType(StrEnum):
    """The storage/validation type of a configuration value."""

    STRING = "string"
    INT = "int"
    FLOAT = "float"
    BOOL = "bool"
    DECIMAL = "decimal"
    JSON = "json"
    CHOICE = "choice"
    DURATION = "duration"  # stored as int seconds


@dataclass(frozen=True)
class ConfigDefinition:
    """Immutable schema for one configuration key."""

    key: str
    type: ConfigType = ConfigType.JSON
    default: Any = None
    choices: tuple[Any, ...] = ()
    is_secret: bool = False
    label: str = ""
    group: str = ""
    description: str = ""
    validator: str | None = None  # dotted path; callable(value) -> None | raises


class ConfigRegistry:
    """Registry of typed configuration definitions.

    The kit's built-ins live in settings; consuming apps register their own in
    ``apps.py:ready()`` (mirroring ``FeatureRegistry``). Mutation is global by
    design — there is one process-wide schema per deployment.
    """

    _definitions: dict[str, ConfigDefinition] = {}

    @classmethod
    def register(
        cls,
        key: str,
        *,
        type: ConfigType | str = ConfigType.JSON,
        default: Any = None,
        choices: tuple[Any, ...] | list[Any] = (),
        is_secret: bool = False,
        label: str = "",
        group: str = "",
        description: str = "",
        validator: str | None = None,
    ) -> ConfigDefinition:
        definition = ConfigDefinition(
            key=key,
            type=ConfigType(type),
            default=default,
            choices=tuple(choices),
            is_secret=is_secret,
            label=label or key,
            group=group,
            description=description,
            validator=validator,
        )
        cls._definitions[key] = definition
        return definition

    @classmethod
    def get(cls, key: str) -> ConfigDefinition | None:
        return cls._definitions.get(key)

    @classmethod
    def all(cls) -> dict[str, ConfigDefinition]:
        return dict(cls._definitions)

    @classmethod
    def clear(cls) -> None:
        """Clears the registry. Tests only; never call from application code."""
        cls._definitions.clear()

    # --- coercion -----------------------------------------------------------

    @classmethod
    def coerce(cls, definition: ConfigDefinition, value: Any) -> Any:
        """Coerces/validates ``value`` against ``definition``.

        Raises :class:`ValidationAppError` (``VALIDATION_CONFIG_INVALID``) on any
        failure, including a failing custom ``validator``.
        """
        try:
            coerced = cls._coerce_value(definition, value)
            if definition.validator:
                import_string(definition.validator)(coerced)
            return coerced
        except ValidationAppError:
            raise
        except Exception as exc:  # noqa: BLE001 - normalised to a typed error
            raise _invalid(definition.key, str(exc)) from exc

    @classmethod
    def _coerce_value(cls, definition: ConfigDefinition, value: Any) -> Any:
        config_type = definition.type

        if config_type == ConfigType.STRING:
            if not isinstance(value, str):
                raise ValueError("expected a string")
            return value

        if config_type == ConfigType.INT:
            if isinstance(value, bool) or not isinstance(value, (int, float, str)):
                raise ValueError("expected an integer")
            if isinstance(value, float) and not value.is_integer():
                raise ValueError("expected a whole number")
            return int(value)

        if config_type == ConfigType.FLOAT:
            if isinstance(value, bool) or not isinstance(value, (int, float, str)):
                raise ValueError("expected a number")
            return float(value)

        if config_type == ConfigType.DECIMAL:
            if isinstance(value, bool):
                raise ValueError("expected a decimal number")
            if isinstance(value, float):
                value = repr(value)
            return decimal.Decimal(str(value))

        if config_type == ConfigType.BOOL:
            if not isinstance(value, bool):
                raise ValueError("expected a boolean")
            return value

        if config_type == ConfigType.CHOICE:
            if value not in definition.choices:
                raise ValueError(f"expected one of {list(definition.choices)!r}")
            return value

        if config_type == ConfigType.DURATION:
            return cls._coerce_duration(value)

        # ConfigType.JSON (and any future pass-through type).
        try:
            json.dumps(value)
        except (TypeError, ValueError) as exc:
            raise ValueError("expected a JSON-serializable value") from exc
        return value

    @staticmethod
    def _coerce_duration(value: Any) -> int:
        if isinstance(value, bool):
            raise ValueError("expected a duration in seconds or a string like '30s'")
        if isinstance(value, int):
            return value
        if isinstance(value, float):
            if not value.is_integer():
                raise ValueError("duration seconds must be a whole number")
            return int(value)
        if not isinstance(value, str):
            raise ValueError("expected an integer number of seconds or a string like '30s'")
        match = _DURATION_RE.match(value)
        if match is None:
            raise ValueError("expected an integer number of seconds or a string like '30s'")
        unit = match.group("unit").lower()
        if unit not in _DURATION_UNITS:
            raise ValueError(f"unknown duration unit '{unit}'")
        return int(float(match.group("amount")) * _DURATION_UNITS[unit])