File size: 2,362 Bytes
b2c2f23 db550a4 b2c2f23 c8d97e7 b2c2f23 db550a4 b2c2f23 db550a4 9bce4c7 c8d97e7 d6828d2 db550a4 d6828d2 db550a4 d6828d2 9bce4c7 d6828d2 c8d97e7 d6828d2 db550a4 9bce4c7 c8d97e7 9bce4c7 c8d97e7 d6828d2 c8d97e7 c5e43db d6828d2 db550a4 9bce4c7 b2c2f23 9bce4c7 d6828d2 c8d97e7 d6828d2 9bce4c7 bf292d9 b2c2f23 |
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 78 79 80 81 |
# 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()
|