File size: 2,320 Bytes
b2c2f23 db550a4 b2c2f23 db550a4 b2c2f23 db550a4 d6828d2 db550a4 d6828d2 db550a4 d6828d2 db550a4 d6828d2 db550a4 d6828d2 db550a4 d6828d2 db550a4 d6828d2 db550a4 d6828d2 c5e43db d6828d2 db550a4 b2c2f23 d6828d2 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 |
# 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()
|