kenaito-conf/python-client/kenaito_conf/client.py

import logging
import threading
import time

import requests

from kenaito_conf.cache import Cache
from kenaito_conf.config import Config
from kenaito_conf.crypto import decrypt_config
from kenaito_conf.listener import Listener

logger = logging.getLogger("kenaito_conf")


class ConfigClient:
    def __init__(self, config: Config):
        self._config = config
        self._cache = Cache(config.cache_dir)
        self._listener = Listener()
        self._configs: dict[str, str] = {}
        self._secret: str = ""
        self._global_secret: str = ""
        self._last_time: int = 0
        self._lock = threading.RLock()
        self._stop_event = threading.Event()
        self._started = False

    def on_change(self, handler) -> None:
        self._listener.on_change(handler)

    def get(self, key: str) -> str | None:
        with self._lock:
            return self._configs.get(key)

    def get_all(self) -> dict[str, str]:
        with self._lock:
            return dict(self._configs)

    def start(self) -> None:
        with self._lock:
            if self._started:
                return
            self._started = True

        try:
            self._fetch_all()
        except Exception as e:
            logger.warning("[ConfigClient] fetch all configs error: %s, trying cache", e)
            cached = self._cache.read()
            if cached is not None:
                self._apply_configs(cached)
                logger.info("[ConfigClient] loaded %d configs from cache", len(self._configs))
            else:
                raise RuntimeError(f"fetch configs failed and no cache: {e}") from e

        t = threading.Thread(target=self._sync_loop, daemon=True)
        t.start()

    def stop(self) -> None:
        self._stop_event.set()

    def _fetch_all(self) -> None:
        url = (
            f"{self._config.base_url}/api/fetchAllConfigs"
            f"?appName={self._config.app_name}"
            f"&env={self._config.env}"
            f"&version={self._config.version}"
            f"&ignoreGlobal={str(self._config.ignore_global).lower()}"
        )
        resp = requests.get(url, timeout=10)
        resp.raise_for_status()
        data = resp.json()

        if data.get("code", 0) != 0:
            raise RuntimeError(f"server error: {data.get('msg', 'unknown')}")

        secret = data.get("kenaito.config.encrypt-secret")
        if secret:
            self._secret = secret

        global_secret = data.get("kenaito.config.global-encrypt-secret")
        if global_secret:
            self._global_secret = global_secret

        ts = data.get("latest.fetch.timestamp")
        if ts:
            self._last_time = int(ts)

        self._cache.write(data)
        self._apply_configs(data)
        logger.info("[ConfigClient] fetched %d configs", len(self._configs))

    def _fetch_changed(self) -> None:
        url = f"{self._config.base_url}/api/fetchChangedConfigs"
        payload = {
            "appName": self._config.app_name,
            "env": self._config.env,
            "version": self._config.version,
            "lastTime": self._last_time,
            "ignoreGlobal": self._config.ignore_global,
        }
        resp = requests.post(url, json=payload, timeout=10)
        resp.raise_for_status()
        result = resp.json()

        if result.get("code", -1) != 0:
            raise RuntimeError(f"server error code: {result.get('code')}")

        changed_data = result.get("data", {})

        ts = changed_data.pop("latest.fetch.timestamp", None)
        if ts:
            self._last_time = int(ts)

        if changed_data:
            changed = {}
            for k, v in changed_data.items():
                decrypted = self._try_decrypt(v)
                changed[k] = decrypted

            with self._lock:
                self._configs.update(changed)

            self._listener.notify(changed)
            logger.info("[ConfigClient] %d configs changed", len(changed))

    def _apply_configs(self, data: dict) -> None:
        with self._lock:
            for k, v in data.items():
                if k.startswith("kenaito.") or k == "latest.fetch.timestamp":
                    continue
                if not isinstance(v, str):
                    continue
                self._configs[k] = self._try_decrypt(v)

    def _try_decrypt(self, value: str) -> str:
        decrypted = decrypt_config(self._secret, value)
        if decrypted == value and self._global_secret:
            decrypted = decrypt_config(self._global_secret, value)
        return decrypted

    def _sync_loop(self) -> None:
        while not self._stop_event.is_set():
            self._stop_event.wait(self._config.sync_interval)
            if self._stop_event.is_set():
                break
            try:
                self._fetch_changed()
            except Exception:
                logger.exception("[ConfigClient] sync error")