"""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\d+(?:\.\d+)?)\s*(?P[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])