e_lib.config

  1# -*- coding: utf-8 -*-
  2"""e_lib.config
  3
  4A small utility class for reading and writing configuration files in YAML or
  5JSON format.  The class supports optional validation using a Pydantic model.
  6
  7The implementation intentionally avoids external dependencies such as
  8``watchdog`` – the ``set_auto_reload`` method is kept for API compatibility but
  9simply logs that auto‑reload is not available.
 10"""
 11
 12from __future__ import annotations
 13
 14import json
 15import asyncio
 16from pathlib import Path
 17from typing import Any, Dict, Optional, Type, Union
 18
 19import yaml
 20
 21# from loguru import logger
 22from pydantic import BaseModel, ValidationError
 23
 24# ---------------------------------------------------------------------------
 25# Logging configuration – plain‑text to stderr with context information.
 26# ---------------------------------------------------------------------------
 27import logging
 28
 29logger = logging.getLogger(__name__)
 30
 31PathLike = Union[str, Path]
 32
 33
 34class Config:
 35    """Load and validate configuration files.
 36
 37    Parameters
 38    ----------
 39    path:
 40        Path to the configuration file.
 41    fmt:
 42        ``"yaml"`` or ``"json"``.  Defaults to ``"yaml"``.
 43    model:
 44        Optional :class:`pydantic.BaseModel` subclass used for validation.
 45    """
 46
 47    def __init__(
 48        self,
 49        path: PathLike,
 50        fmt: str = "yaml",
 51        model: Optional[Type[BaseModel]] = None,
 52    ) -> None:
 53        self.path = Path(path).expanduser().resolve()
 54        if not self.path.is_file():
 55            raise FileNotFoundError(f"Configuration file {self.path} does not exist")
 56
 57        fmt = fmt.lower()
 58        if fmt not in ("yaml", "json"):
 59            raise ValueError("fmt must be either 'yaml' or 'json'")
 60        self.fmt = fmt
 61        self._model = model
 62        self._data: Dict[str, Any] = {}
 63        self.load()
 64
 65    # ------------------------------------------------------------------
 66    # Internal helpers
 67    # ------------------------------------------------------------------
 68    def _load_file(self) -> Dict[str, Any]:
 69        """Read the file and return a dict.
 70
 71        Supports an ``include`` key that lists additional files to merge.  Paths are
 72        resolved relative to the current file.  Includes are processed recursively.
 73        Circular includes are detected and raise ValueError.
 74        """
 75
 76        def _load(path: Path, visited: set[Path]) -> Dict[str, Any]:
 77            resolved = path.resolve()
 78            if resolved in visited:
 79                raise ValueError(f"Circular include detected: {resolved}")
 80            visited = visited | {resolved}
 81            with open(path, "r", encoding="utf-8") as f:
 82                raw = (
 83                    yaml.safe_load(f)
 84                    if path.suffix.lower() in (".yaml", ".yml")
 85                    else json.load(f)
 86                )
 87            if not isinstance(raw, dict):
 88                raise ValueError(
 89                    "Configuration file must contain a mapping at top level"
 90                )
 91            includes = raw.pop("include", [])
 92            if isinstance(includes, str):
 93                includes = [includes]
 94            for inc in includes:
 95                inc_path = (path.parent / inc).resolve()
 96                if not inc_path.is_file():
 97                    raise FileNotFoundError(
 98                        f"Included config file {inc_path} does not exist"
 99                    )
100                raw.update(_load(inc_path, visited))
101            return raw
102
103        return _load(self.path, set())
104
105    def _validate(self, data: Dict[str, Any]) -> bool:
106        if not self._model:
107            return True
108        try:
109            self._model(**data)
110            return True
111        except ValidationError as exc:
112            logger.error("Config validation failed for %s: %s", self.path, exc)
113            raise ValueError(f"Validation failed for {self.path}: {exc}")
114
115    # ------------------------------------------------------------------
116    # Public API
117    # ------------------------------------------------------------------
118    def load(self) -> None:
119        new_data = self._load_file()
120        if self._validate(new_data):
121            self._data = new_data
122
123    async def async_load(self) -> None:
124        # Delegate to the synchronous loader in a thread
125        new_data = await asyncio.to_thread(self._load_file)
126        if self._validate(new_data):
127            self._data = new_data
128
129    def save(
130        self,
131        data: Optional[Dict[str, Any]] = None,
132        output_path: Optional[PathLike] = None,
133    ) -> None:
134        target = data if data is not None else self._data
135        if not isinstance(target, dict):
136            raise TypeError("save() expects a dict or None")
137        content = (
138            yaml.safe_dump(target)
139            if self.fmt == "yaml"
140            else json.dumps(target, indent=2)
141        )
142        dest = Path(output_path) if output_path is not None else self.path
143        tmp_path = dest.parent / f"{dest.name}.tmp"
144        try:
145            tmp_path.write_text(content, encoding="utf-8")
146            tmp_path.replace(dest)
147        except Exception:
148            if tmp_path.exists():
149                tmp_path.unlink(missing_ok=True)
150            raise
151
152    async def async_save(
153        self,
154        data: Optional[Dict[str, Any]] = None,
155        output_path: Optional[PathLike] = None,
156    ) -> None:
157        target = data if data is not None else self._data
158        content = (
159            yaml.safe_dump(target)
160            if self.fmt == "yaml"
161            else json.dumps(target, indent=2)
162        )
163        dest = Path(output_path) if output_path is not None else self.path
164        tmp_path = dest.parent / f"{dest.name}.tmp"
165        try:
166            tmp_path.write_text(content, encoding="utf-8")
167            await asyncio.to_thread(tmp_path.replace, dest)
168        except Exception:
169            if tmp_path.exists():
170                tmp_path.unlink(missing_ok=True)
171            raise
172
173    # ------------------------------------------------------------------
174    # Auto‑reload placeholder – no watchdog integration.
175    # ------------------------------------------------------------------
176    def set_auto_reload(self, enable: bool) -> None:
177        if enable:
178            logger.warning("Auto‑reload is not implemented in this simplified version")
179
180    # ------------------------------------------------------------------
181    # Convenience accessors
182    # ------------------------------------------------------------------
183    @property
184    def data(self) -> Dict[str, Any]:
185        return self._data.copy()
186
187    def get(self, key: str, default: Any = None) -> Any:
188        return self._data.get(key, default)
189
190    def set(self, key: str, value: Any) -> None:
191        self._data[key] = value
192
193    # ------------------------------------------------------------------
194    # Context‑manager support
195    # ------------------------------------------------------------------
196    def __enter__(self) -> "Config":
197        return self
198
199    def __exit__(
200        self,
201        exc_type: Optional[type],
202        exc_val: Optional[Exception],
203        exc_tb: Optional[Any],
204    ) -> None:
205        pass  # nothing to clean up in this simplified version
206
207
208# End of config.py