66 lines
2.4 KiB
Python
66 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
from app.models import AppConfig
|
|
|
|
_ENV_PATTERN = re.compile(r"\$\{([A-Z0-9_]+)\}")
|
|
|
|
|
|
def _expand_env(value):
|
|
if isinstance(value, str):
|
|
return _ENV_PATTERN.sub(lambda m: os.getenv(m.group(1), ""), value)
|
|
if isinstance(value, list):
|
|
return [_expand_env(v) for v in value]
|
|
if isinstance(value, dict):
|
|
return {k: _expand_env(v) for k, v in value.items()}
|
|
return value
|
|
|
|
|
|
def _load_yaml(path: Path) -> dict[str, Any]:
|
|
if not path.is_file():
|
|
return {}
|
|
raw = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
|
|
if not isinstance(raw, dict):
|
|
raise ValueError(f"Config file must contain a top-level mapping: {path}")
|
|
return raw
|
|
|
|
|
|
def _section_value(raw: dict[str, Any], section: str) -> Any:
|
|
if section in raw:
|
|
return raw[section]
|
|
return raw
|
|
|
|
|
|
def _load_split_config(path: Path) -> dict[str, Any]:
|
|
config_dir = path if path.is_dir() else path.parent
|
|
legacy_raw = _load_yaml(path) if path.is_file() else {}
|
|
|
|
app_raw = _load_yaml(config_dir / "app.yaml")
|
|
sources_raw = _load_yaml(config_dir / "sources.yaml")
|
|
regions_raw = _load_yaml(config_dir / "regions.yaml")
|
|
groups_raw = _load_yaml(config_dir / "policy-groups.yaml")
|
|
rules_raw = _load_yaml(config_dir / "rules.yaml")
|
|
clients_raw = _load_yaml(config_dir / "clients.yaml")
|
|
|
|
merged: dict[str, Any] = {}
|
|
merged.update({key: value for key, value in legacy_raw.items() if key not in AppConfig.model_fields})
|
|
merged["public_path"] = app_raw.get("public_path", legacy_raw.get("public_path"))
|
|
merged["sources"] = _section_value(sources_raw, "sources") if sources_raw else legacy_raw.get("sources", {})
|
|
merged["regions"] = _section_value(regions_raw, "regions") if regions_raw else legacy_raw.get("regions", {})
|
|
merged["selector_groups"] = groups_raw.get("selector_groups", legacy_raw.get("selector_groups", []))
|
|
merged["policy_groups"] = groups_raw.get("policy_groups", legacy_raw.get("policy_groups", []))
|
|
merged["rules"] = _section_value(rules_raw, "rules") if rules_raw else legacy_raw.get("rules", {})
|
|
merged["clients"] = _section_value(clients_raw, "clients") if clients_raw else legacy_raw.get("clients", {})
|
|
return merged
|
|
|
|
|
|
def load_app_config(path: Path) -> AppConfig:
|
|
expanded = _expand_env(_load_split_config(path))
|
|
return AppConfig.model_validate(expanded)
|