# app/config.py import json import os from functools import lru_cache from urllib.parse import quote APPSETTINGS_PATH = os.environ.get("APPSETTINGS_JSON", "appsettings.json") 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 _resolve_env_placeholders(data): """ Recursively replace values that are ".env" with the matching environment variable. """ if isinstance(data, dict): return {k: _resolve_env_placeholders(v) for k, v in data.items()} elif isinstance(data, list): return [_resolve_env_placeholders(v) for v in data] elif isinstance(data, str) and data.strip() == ".env": # Use the key name uppercased from parent call if available # This requires that parent dict was processed with context return None # will be handled by the parent context return data def _replace_env_vars(d, parent_key=None): """ Same as _resolve_env_placeholders but knows the parent key for .env resolution. """ if isinstance(d, dict): result = {} for k, v in d.items(): result[k] = _replace_env_vars(v, k) return result elif isinstance(d, str) and d.strip() == ".env": env_key = parent_key.upper() return os.environ.get(env_key) else: return d def build_amqp_url(cfg): local = cfg.get("LocalSystemUrl") if not local: return None scheme = "amqps" if local.get("UseTls", True) else "amqp" host = local.get("RabbitHostName") port = local.get("RabbitPort") or (5671 if scheme == "amqps" else 5672) user = local.get("RabbitUserName") pwd = local.get("RabbitPassword") or os.environ.get("RABBIT_PASSWORD") vhost = local.get("RabbitVHost") 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(): cfg = _load_json(APPSETTINGS_PATH) cfg = _replace_env_vars(cfg) # If AMQP_URL is not directly set, build it if not cfg.get("AMQP_URL"): amqp_url = build_amqp_url(cfg) if amqp_url: cfg["AMQP_URL"] = amqp_url return cfg settings = get_settings()