Spaces:
Running
Running
File size: 2,425 Bytes
cef045d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | """Config loader with hot-reload support."""
import os
try:
import tomllib
except ModuleNotFoundError:
import tomli as tomllib
import tomli_w
_CONFIG_PATH = os.path.abspath(
os.path.join(os.path.dirname(__file__), "..", "..", "config", "setting.toml")
)
def _load():
with open(_CONFIG_PATH, "rb") as f:
return tomllib.load(f)
def _save(cfg: dict):
with open(_CONFIG_PATH, "wb") as f:
tomli_w.dump(cfg, f)
_cfg = _load()
API_KEY = _cfg["global"]["api_key"]
HOST = _cfg["server"]["host"]
PORT = _cfg["server"]["port"]
# Admin credentials
ADMIN_USERNAME = _cfg.get("admin", {}).get("username", "admin")
ADMIN_PASSWORD = _cfg.get("admin", {}).get("password", "admin")
# Proxy
PROXY_URL = _cfg.get("proxy", {}).get("url", "")
# Retry
MAX_RETRIES = _cfg.get("retry", {}).get("max_retries", 3)
RETRY_DELAY = _cfg.get("retry", {}).get("retry_delay", 1)
# OB-1 config
OB1_CREDENTIALS_PATH = _cfg["ob1"].get("credentials_path", "")
OB1_WORKOS_AUTH_URL = _cfg["ob1"]["workos_auth_url"]
OB1_WORKOS_CLIENT_ID = _cfg["ob1"]["workos_client_id"]
OB1_API_BASE = _cfg["ob1"]["api_base"]
OB1_REFRESH_BUFFER = _cfg["ob1"].get("refresh_buffer_seconds", 600)
OB1_ROTATION_MODE = _cfg["ob1"].get("rotation_mode", "cache-first")
OB1_REFRESH_INTERVAL = _cfg["ob1"].get("refresh_interval", 0)
# Logging
LOG_LEVEL = _cfg.get("logging", {}).get("level", "INFO")
def reload():
"""Reload config from disk into module-level variables."""
global _cfg, API_KEY, ADMIN_USERNAME, ADMIN_PASSWORD, PROXY_URL, MAX_RETRIES, RETRY_DELAY, OB1_ROTATION_MODE, OB1_REFRESH_INTERVAL, LOG_LEVEL
_cfg = _load()
API_KEY = _cfg["global"]["api_key"]
ADMIN_USERNAME = _cfg.get("admin", {}).get("username", "admin")
ADMIN_PASSWORD = _cfg.get("admin", {}).get("password", "admin")
PROXY_URL = _cfg.get("proxy", {}).get("url", "")
MAX_RETRIES = _cfg.get("retry", {}).get("max_retries", 3)
RETRY_DELAY = _cfg.get("retry", {}).get("retry_delay", 1)
OB1_ROTATION_MODE = _cfg["ob1"].get("rotation_mode", "cache-first")
OB1_REFRESH_INTERVAL = _cfg["ob1"].get("refresh_interval", 0)
LOG_LEVEL = _cfg.get("logging", {}).get("level", "INFO")
def update_setting(section: str, key: str, value):
"""Update a single setting, persist to disk, and reload."""
cfg = _load()
if section not in cfg:
cfg[section] = {}
cfg[section][key] = value
_save(cfg)
reload()
|