# app/config.py import json import os import re from functools import lru_cache from urllib.parse import quote APPSETTINGS_PATH = os.environ.get("APPSETTINGS_JSON", "appsettings.json") class Settings: """Simple settings object loaded from appsettings.json + env vars.""" def __init__(self, data: dict): for k, v in data.items(): setattr(self, k, v) def _to_upper_snake(name: str) -> str: """Convert CamelCase or mixedCase to UPPER_SNAKE_CASE.""" s1 = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", name) s2 = re.sub("([a-z0-9])([A-Z])", r"\1_\2", s1) return s2.upper() def _load_json(path): if not os.path.exists(path): return {} with open(path, "r", encoding="utf-8") as f: return json.load(f) def _replace_env_vars(d, parent_key=None): if isinstance(d, dict): return {k: _replace_env_vars(v, k) for k, v in d.items()} elif isinstance(d, str) and d.strip() == ".env": env_key = _to_upper_snake(parent_key or "") return os.environ.get(env_key) else: return d def _normalize_keys(d): """Recursively normalize dict keys to UPPER_SNAKE_CASE.""" if isinstance(d, dict): return {_to_upper_snake(k): _normalize_keys(v) for k, v in d.items()} elif isinstance(d, list): return [_normalize_keys(i) for i in d] else: return d def build_amqp_url(cfg: dict): local = cfg.get("LOCAL_SYSTEM_URL") if not local: return None scheme = "amqps" if local.get("USE_TLS", True) else "amqp" host = local.get("RABBIT_HOST_NAME") port = local.get("RABBIT_PORT") or (5671 if scheme == "amqps" else 5672) user = local.get("RABBIT_USER_NAME") pwd = local.get("RABBIT_PASSWORD") or os.environ.get("RABBIT_PASSWORD") vhost = local.get("RABBIT_V_HOST") or "/" vhost_enc = quote(vhost, safe="") if not (host and user and pwd): return None return f"{scheme}://{user}:{pwd}@{host}:{port}/{vhost_enc}?heartbeat=30" @lru_cache def get_settings() -> Settings: cfg = _load_json(APPSETTINGS_PATH) cfg = _replace_env_vars(cfg) cfg = _normalize_keys(cfg) # normalize AFTER env replacement if not cfg.get("AMQP_URL"): amqp_url = build_amqp_url(cfg) if amqp_url: cfg["AMQP_URL"] = amqp_url return Settings(cfg) settings = get_settings()