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