File size: 1,879 Bytes
b2c2f23 db550a4 b2c2f23 db550a4 b2c2f23 db550a4 9bce4c7 d6828d2 db550a4 d6828d2 db550a4 d6828d2 db550a4 9bce4c7 db550a4 d6828d2 9bce4c7 d6828d2 9bce4c7 d6828d2 db550a4 9bce4c7 d6828d2 c5e43db d6828d2 db550a4 9bce4c7 b2c2f23 9bce4c7 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 |
# 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")
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 _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):
"""
Replace any ".env" string with the corresponding environment variable.
Uses UPPERCASE parent key as env var name.
"""
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 = (parent_key or "").upper()
return os.environ.get(env_key)
else:
return d
def build_amqp_url(cfg: dict):
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() -> Settings:
cfg = _load_json(APPSETTINGS_PATH)
cfg = _replace_env_vars(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()
|