crypto_okx/lib/instance/runtime_config_lib.py

"""env 运行时覆盖:热生效项优先读 SQLite,再回退 os.environ."""
from __future__ import annotations

import os
from typing import Callable, Optional

from lib.env.env_file_lib import load_env_file_into_environ
from lib.instance.runtime_settings_lib import runtime_get, with_db

ENV_OVERRIDE_PREFIX = "env."


def runtime_env_key(name: str) -> str:
    return ENV_OVERRIDE_PREFIX + name


def get_config(key: str, get_db: Callable, default: Optional[str] = None) -> Optional[str]:
    def _read(conn):
        v = runtime_get(conn, runtime_env_key(key))
        return v

    try:
        v = with_db(get_db, _read)
        if v is not None:
            return v
    except Exception:
        pass
    raw = os.getenv(key)
    if raw is None or raw == "":
        return default
    return raw


def set_config_overrides(get_db: Callable, mapping: dict[str, str]) -> None:
    from lib.instance.runtime_settings_lib import runtime_set_many

    def _write(conn):
        payload = {runtime_env_key(k): str(v) for k, v in mapping.items()}
        runtime_set_many(conn, payload)

    with_db(get_db, _write)


def apply_env_reload(env_path: str, get_db: Callable, changed_keys: list[str], groups: list[dict]) -> dict[str, bool]:
    """写盘后同步 os.environ,并将可热生效项写入 runtime 覆盖."""
    load_env_file_into_environ(env_path)
    hot: dict[str, str] = {}
    field_map = {}
    for group in groups:
        for field in group.get("fields") or []:
            field_map[field["key"]] = field
    for key in changed_keys:
        meta = field_map.get(key) or {}
        if meta.get("hot_reload") and not meta.get("restart_required"):
            val = os.getenv(key)
            if val is not None:
                hot[key] = val
    if hot:
        set_config_overrides(get_db, hot)
    from lib.env.env_schema import updates_need_restart

    return {"restart_required": updates_need_restart(groups, changed_keys)}