wehub-resource-sync
128 lines
4.1 KiB
Python
128 lines
4.1 KiB
Python
"""Settings loader, override switch, and disk persistence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from pydantic import AliasChoices, BaseModel
|
|
|
|
from strix.config.settings import Settings
|
|
|
|
|
|
if TYPE_CHECKING:
|
|
from pydantic.fields import FieldInfo
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
_DEFAULT_PATH: Path = Path.home() / ".strix" / "cli-config.json"
|
|
_override: Path | None = None
|
|
_cached: Settings | None = None
|
|
|
|
|
|
def load_settings() -> Settings:
|
|
"""Resolve settings from env + JSON file + defaults. Memoized.
|
|
|
|
Precedence: env vars win, then the JSON file, then field defaults.
|
|
"""
|
|
global _cached # noqa: PLW0603
|
|
if _cached is None:
|
|
source_path = _override or _DEFAULT_PATH
|
|
init_kwargs: dict[str, Any] = _read_json_overrides(source_path)
|
|
_cached = Settings(**init_kwargs)
|
|
logger.debug(
|
|
"load_settings: resolved (override=%s, file_used=%s, json_keys=%d)",
|
|
_override is not None,
|
|
source_path.exists(),
|
|
sum(len(v) for v in init_kwargs.values()),
|
|
)
|
|
return _cached
|
|
|
|
|
|
def apply_config_override(path: Path) -> None:
|
|
"""Switch the JSON source to ``path`` and invalidate the cache."""
|
|
global _override, _cached # noqa: PLW0603
|
|
_override = path
|
|
_cached = None
|
|
logger.info("config override applied: %s", path)
|
|
|
|
|
|
def persist_current() -> None:
|
|
"""Write currently-set env vars to the active config file (0o600)."""
|
|
s = load_settings()
|
|
target = _override or _DEFAULT_PATH
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
env_block: dict[str, str] = {}
|
|
for sub_name in s.model_fields:
|
|
sub_model = getattr(s, sub_name)
|
|
if not isinstance(sub_model, BaseModel):
|
|
continue
|
|
for finfo in type(sub_model).model_fields.values():
|
|
for alias in _aliases_for(finfo):
|
|
value = os.environ.get(alias.upper())
|
|
if value:
|
|
env_block[alias.upper()] = value
|
|
break
|
|
|
|
target.write_text(json.dumps({"env": env_block}, indent=2), encoding="utf-8")
|
|
with contextlib.suppress(OSError):
|
|
target.chmod(0o600)
|
|
|
|
|
|
def _aliases_for(finfo: FieldInfo) -> list[str]:
|
|
"""Collect every env-var name that should populate ``finfo``."""
|
|
aliases: list[str] = []
|
|
if finfo.alias:
|
|
aliases.append(finfo.alias)
|
|
va = finfo.validation_alias
|
|
if isinstance(va, AliasChoices):
|
|
aliases.extend(c for c in va.choices if isinstance(c, str))
|
|
elif isinstance(va, str):
|
|
aliases.append(va)
|
|
return aliases
|
|
|
|
|
|
def _read_json_overrides(path: Path) -> dict[str, dict[str, Any]]:
|
|
"""Read ``{"env": {...}}`` from ``path`` and remap to nested kwargs.
|
|
|
|
Only includes keys whose env var is NOT already set, so env always
|
|
wins over the persisted file.
|
|
"""
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except (json.JSONDecodeError, OSError):
|
|
return {}
|
|
env_block = data.get("env", {}) if isinstance(data, dict) else {}
|
|
if not isinstance(env_block, dict):
|
|
return {}
|
|
|
|
env_block_upper = {str(k).upper(): v for k, v in env_block.items()}
|
|
env_present = {k.upper() for k in os.environ}
|
|
|
|
nested: dict[str, dict[str, Any]] = {}
|
|
for sub_name, sub_finfo in Settings.model_fields.items():
|
|
sub_cls = sub_finfo.annotation
|
|
if not (isinstance(sub_cls, type) and issubclass(sub_cls, BaseModel)):
|
|
continue
|
|
sub_data: dict[str, Any] = {}
|
|
for fname, finfo in sub_cls.model_fields.items():
|
|
aliases = [alias.upper() for alias in _aliases_for(finfo)]
|
|
if any(alias in env_present for alias in aliases):
|
|
continue # env wins under some alias; skip the JSON file for this field
|
|
for alias in aliases:
|
|
if alias in env_block_upper:
|
|
sub_data[fname] = env_block_upper[alias]
|
|
break
|
|
if sub_data:
|
|
nested[sub_name] = sub_data
|
|
return nested
|