"""Logic for the config handler
Thread-safe singleton class that loads the TOML config file once and makes it accessible
from anywhere in the project.
Usage:
# At the very start of your entry point (optional) (NOTE: Only called once)
Config.set_path("config/prod.toml")
# Anywhere else in the codebase
host = Config().get("db", "host", default="localhost")
port = Config().get("db", "port", default=5432, cast=True) # "5432" -> 5432
fresh = Config().get("db", "host", refresh=True) # re-read from disk first
Warning: The first `Config()` call **must** be called **after** `Config.set_path(...)`.
"""
import tomllib
from pathlib import Path
from threading import Lock
from typing import Any, ClassVar, Dict, Tuple
from pi_weather.constants import DEFAULT_CONFIG_FILE
from pi_weather.exceptions import (
ConfigPathAlreadySetException,
SettingsNotImportedException,
)
from pi_weather.utils.logs import get_logger
logger = get_logger(__name__)
[docs]
class Config:
"""Singleton config handler backed by a TOML file."""
# Config instance
_instance: ClassVar["Config | None"] = None
# Dict to store imported settings
_settings: ClassVar[Dict[str, Any] | None] = None
# Config file path
_config_path: ClassVar[Path] = Path(DEFAULT_CONFIG_FILE)
# Setting indicating whether the config file path has already been set
_path_locked: ClassVar[bool] = False
# threading.Lock to guarantee thread-safety
_lock: ClassVar[Lock] = Lock()
[docs]
def __new__(cls, *args, **kwargs) -> "Config":
"""
(Real c-tor) Instantiate Config, importing settings on cold start.
:param args: Positional arguments
:param kwargs: Keyword arguments
:raises SettingsNotImportedException: if the TOML file can't be read/parsed
:return: (Config)
"""
if cls._instance is None:
with cls._lock:
if cls._instance is None: # double-checked locking
logger.info(
f"Instantiating config (cold start) from {cls._config_path}"
)
status, settings = cls._import_settings()
if not status:
logger.error("Settings not imported yet.")
raise SettingsNotImportedException(
f"Could not import settings from {cls._config_path}"
)
instance = super().__new__(cls)
cls._settings = settings
cls._path_locked = True
cls._instance = instance # only set once settings actually loaded
return cls._instance
# ------------------------------------------------------------------
# Path configuration
# ------------------------------------------------------------------
[docs]
@classmethod
def set_path(cls, filename: str | Path) -> None:
"""
Set the config file path. Must be called before the first
instantiation/access - the path is locked in after that (either
because the singleton was created, or because set_path() already
ran once successfully).
:param filename: Path to the config.toml file
:type filename: str | Path
:raises ConfigPathAlreadySetException: if the path is already locked in
"""
if cls._path_locked or cls._instance is not None:
raise ConfigPathAlreadySetException(
"Config path is already locked in; call set_path() before "
"the first Config() access."
)
cls._config_path = Path(filename)
cls._path_locked = True
logger.info(f"Config path set to {cls._config_path}")
[docs]
@classmethod
def _import_settings(cls) -> Tuple[bool, Dict[str, Any]]:
"""
Import settings from cls._config_path.
Keys are lowercased recursively (values are left untouched)
so lookups are case-insensitive regardless of how the TOML file is written.
:return: (success, settings dict)
:rtype: Tuple[bool, Dict[str, Any]]
"""
try:
with open(cls._config_path, "rb") as f:
logger.info(f"Importing settings from {cls._config_path}")
settings = tomllib.load(f)
settings = cls._lowercase_keys(settings)
logger.info("Successfully imported settings")
return True, settings
except Exception as exc_:
logger.error(f"An error occurred while importing settings: {exc_}")
return False, {}
[docs]
@classmethod
def _lowercase_keys(cls, obj: Any) -> Any:
"""
Recursively lowercase all dict keys (values untouched), including
dicts nested inside lists (e.g. TOML arrays of tables).
:param obj: Object to normalize
:type obj: Any
:return: Object with all nested dict keys lowercased
:rtype: Any
"""
if isinstance(obj, dict):
return {
(k.lower() if isinstance(k, str) else k): cls._lowercase_keys(v)
for k, v in obj.items()
}
if isinstance(obj, list):
return [cls._lowercase_keys(item) for item in obj]
return obj
[docs]
@classmethod
def reload(cls) -> bool:
"""
Force a reload of settings from disk (e.g. you edited the file
on disk and don't want to restart the process).
:return: True if reload succeeded (settings replaced), False otherwise
:rtype: bool
"""
with cls._lock:
status, settings = cls._import_settings()
if not status:
logger.error("Reload failed; keeping previous settings in memory")
return False
cls._settings = settings
logger.info(f"Settings reloaded from {cls._config_path}")
return True
[docs]
@classmethod
def force_reload(cls) -> bool:
"""
Explicit alias for reload() - re-reads the config file from disk
and updates _settings in place, independent of any get() call.
Call this whenever you know the on-disk file changed, e.g.:
Config.force_reload()
:return: True if reload succeeded, False otherwise
:rtype: bool
"""
return cls.reload()
# ------------------------------------------------------------------
# Accessors
# ------------------------------------------------------------------
[docs]
def get_settings(self, refresh: bool = False) -> Dict[str, Any]:
"""
Return the full settings dictionary.
:param refresh: Reload from disk before returning
:type refresh: bool
:return: Settings dictionary, empty dict if none loaded
:rtype: Dict[str, Any]
"""
if refresh:
self.reload()
if self._settings is None:
logger.warning("Settings not imported yet.")
return {}
return self._settings
[docs]
def get(
self,
*args: str,
default: Any | None = None,
refresh: bool = False,
cast: bool = False,
) -> Any:
"""
Get a (possibly nested) key from settings.
:param args: Sequence of nested keys, e.g. get("db", "host")
:type args: str
:param default: Value returned if the key path doesn't exist
:type default: Any | None
:param refresh: Reload settings from disk before reading
:type refresh: bool
:param cast: Attempt to type-cast string values to bool/int/float/None
before returning. Only affects strings - values TOML already
parsed as numbers/bools are untouched.
:type cast: bool
:return: Retrieved value (or default)
:rtype: Any
"""
if refresh:
self.reload()
if self._settings is None:
logger.warning("Settings not imported yet.")
return default
if len(args) == 0:
return self._settings
keys = tuple(a.lower() if isinstance(a, str) else a for a in args)
v: Any = self._settings
for key in keys[:-1]:
if isinstance(v, dict):
v = v.get(key, {})
else:
logger.warning(f"Key {key} does not exist. Skipping remaining keys.")
return default
if isinstance(v, dict):
v = v.get(keys[-1], default)
else:
logger.warning(f"Key {keys[-1]} does not exist")
return default
return self._maybe_cast(v) if cast else v
[docs]
@staticmethod
def _maybe_cast(value: Any) -> Any:
"""
Best-effort cast of a string value to bool/int/float/None.
Non-string values are returned unchanged.
:param value: Value to cast
:type value: Any
:return: Casted value, or original value if no cast applies
:rtype: Any
"""
if not isinstance(value, str):
return value
stripped = value.strip()
lowered = stripped.lower()
if lowered in ("true", "false"):
return lowered == "true"
if lowered in ("none", "null", "~"):
return None
try:
return int(stripped)
except ValueError:
pass
try:
return float(stripped)
except ValueError:
pass
return value
[docs]
def __getattr__(self, key: str) -> Any | None:
"""
Dict-style attribute access, e.g. Config().db instead of .get("db").
Note: this only fires when normal attribute lookup fails, so a TOML
key named e.g. "get" or "reload" would be shadowed by the real
method - use get() explicitly for keys that might collide.
:param key: Key name
:type key: str
:return: Settings value matching key, or None if missing
:rtype: Any | None
"""
if self._settings is None:
logger.error("Settings not imported yet.")
return None
key = key.lower()
if key not in self._settings:
logger.error(
f"Key {key} not found in settings",
)
return None
return self._settings.get(key)