|
|
|
|
|
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) |
|
|
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() |
|
|
|
