from __future__ import annotations
from pathlib import Path
from typing import Any, ClassVar, Optional, cast
import yaml
from pydantic.fields import FieldInfo
from pydantic_settings import (
BaseSettings,
PydanticBaseSettingsSource,
SettingsConfigDict,
)
from typing_extensions import override
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 {} # pyright: ignore[reportUnknownMemberType]
if not raw or not isinstance(raw, dict):
logger.warning("YAML root is not a mapping. Skipping file.", path=str(p))
continue
# Narrow type after isinstance check
raw_dict = cast(dict[str, Any], raw)
cfg: Optional[dict[str, Any]] = None
if p.name == "config.yaml":
cfg = raw_dict
else:
app_section: Any = raw_dict.get("app")
if isinstance(app_section, dict):
app_dict = cast(dict[str, Any], app_section)
docker_section: Any = app_dict.get("docker")
if isinstance(docker_section, dict):
docker_dict = cast(dict[str, Any], docker_section)
docker_cfg: Any = docker_dict.get("configuration")
if isinstance(docker_cfg, dict):
cfg = cast(dict[str, Any], docker_cfg)
if cfg is None:
defaults_section: Any = raw_dict.get("defaults")
if isinstance(defaults_section, dict):
defaults_dict = cast(dict[str, Any], defaults_section)
defaults_cfg: Any = defaults_dict.get("configuration")
if isinstance(defaults_cfg, dict):
cfg = cast(dict[str, Any], defaults_cfg)
if cfg is not None:
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)
@override
def __call__(self) -> dict[str, Any]:
return self._data
[docs]
@override
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: ClassVar[SettingsConfigDict] = 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
@override
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,
)