from __future__ import annotations
from pathlib import Path
from typing import Any, ClassVar, Optional, Type
import yaml
from pydantic.fields import FieldInfo
from pydantic_settings import (
BaseSettings,
PydanticBaseSettingsSource,
SettingsConfigDict,
)
from kelvin.logs import logger
# ---------------------------------------------------------------------
# Defaults
# ---------------------------------------------------------------------
DEFAULT_KELVIN_PATHS: tuple[Path, ...] = (
Path("/opt/kelvin/share/config.yaml"),
Path("/opt/kelvin/share/app.yaml"),
Path("config.yaml"),
Path("app.yaml"),
)
# ---------------------------------------------------------------------
# External helper
# ---------------------------------------------------------------------
[docs]
def load_kelvin_yaml_config(paths: Optional[tuple[Path, ...]] = None) -> dict[str, Any]:
"""
Load Kelvin YAML configuration from the first existing file in `paths`.
Default search order when `paths` is None:
- /opt/kelvin/share/config.yaml
- /opt/kelvin/share/app.yaml
- config.yaml
- app.yaml
Extraction rules:
- If filename is 'config.yaml' (local or under /opt/kelvin/share), use the YAML root mapping.
- Otherwise, try nested:
app.docker.configuration
then fall back to:
defaults.configuration
Returns:
dict[str, Any] with configuration values. Empty dict if nothing found or usable.
"""
effective_paths = paths or DEFAULT_KELVIN_PATHS
for p in effective_paths:
try:
if not p.exists():
continue
with p.open("r", encoding="utf-8") as fh:
raw: Any = yaml.safe_load(fh) or {}
if not raw or not isinstance(raw, dict):
logger.warning("YAML root is not a mapping. Skipping file.", path=str(p))
continue
if p.name == "config.yaml":
cfg: Any = raw
else:
cfg = raw.get("app", {}).get("docker", {}).get("configuration")
if cfg is None:
cfg = raw.get("defaults", {}).get("configuration")
if isinstance(cfg, dict):
logger.info("Loaded configuration from YAML.", path=str(p))
return cfg
except Exception as exc:
logger.warning("Failed reading YAML config.", path=str(p), error=str(exc))
continue
logger.info("No YAML config found in provided or default paths.")
return {}
# ---------------------------------------------------------------------
# Pydantic Source
# ---------------------------------------------------------------------
[docs]
class KelvinYamlConfigSettingsSource(PydanticBaseSettingsSource):
"""
A Pydantic Settings Source that reads Kelvin-style YAML configuration files.
"""
def __init__(self, settings_cls: Type[BaseSettings]) -> None:
super().__init__(settings_cls)
# Uses the helper with default paths if settings_cls has none
paths = getattr(settings_cls, "YAML_PATHS", DEFAULT_KELVIN_PATHS)
self._data: dict[str, Any] = load_kelvin_yaml_config(paths)
def __call__(self) -> dict[str, Any]:
return self._data
[docs]
def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
if field.alias and field.alias in self._data:
return self._data[field.alias], "yaml", True
if field_name in self._data:
return self._data[field_name], "yaml", True
return None, "yaml", False
# ---------------------------------------------------------------------
# Base App Config
# ---------------------------------------------------------------------
[docs]
class KelvinAppConfig(BaseSettings):
"""
Kelvin application base configuration.
Load order and override precedence:
1. init kwargs
2. environment variables
3. .env file
4. YAML configuration (first match in YAML_PATHS or defaults)
5. file secrets
This means values from environment variables or `.env` override values from YAML.
YAML discovery:
- Reads from the first existing file among:
* /opt/kelvin/share/config.yaml
* /opt/kelvin/share/app.yaml
* config.yaml
* app.yaml
- If the file name is `config.yaml` (local or under /opt/kelvin/share),
the entire YAML root is used.
- Otherwise, the loader extracts:
app.docker.configuration
or falls back to:
defaults.configuration
Nested environment variables:
- Set `env_nested_delimiter="__"` so nested fields can be overridden.
Example: `MQTT__PASSWORD=secret` overrides `mqtt.password`.
Customization:
- Override YAML_PATHS on your subclass to change search locations.
- Adjust model_config.env_prefix to fit your application.
"""
model_config = SettingsConfigDict(
env_prefix="",
env_nested_delimiter="__",
env_file=".env",
env_file_encoding="utf-8",
extra="ignore",
)
YAML_PATHS: ClassVar[tuple[Path, ...]] = DEFAULT_KELVIN_PATHS
[docs]
@classmethod
def settings_customise_sources(
cls,
settings_cls: Type[BaseSettings],
init_settings: PydanticBaseSettingsSource,
env_settings: PydanticBaseSettingsSource,
dotenv_settings: PydanticBaseSettingsSource,
file_secret_settings: PydanticBaseSettingsSource,
) -> tuple[PydanticBaseSettingsSource, ...]:
# Order: init -> env -> dotenv -> YAML -> file secrets
return (
init_settings,
env_settings,
dotenv_settings,
KelvinYamlConfigSettingsSource(settings_cls),
file_secret_settings,
)