Commit
·
9bce4c7
1
Parent(s):
d6828d2
config.py
CHANGED
|
@@ -6,42 +6,36 @@ from urllib.parse import quote
|
|
| 6 |
|
| 7 |
APPSETTINGS_PATH = os.environ.get("APPSETTINGS_JSON", "appsettings.json")
|
| 8 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 9 |
def _load_json(path):
|
| 10 |
if not os.path.exists(path):
|
| 11 |
return {}
|
| 12 |
with open(path, "r", encoding="utf-8") as f:
|
| 13 |
return json.load(f)
|
| 14 |
|
| 15 |
-
def _resolve_env_placeholders(data):
|
| 16 |
-
"""
|
| 17 |
-
Recursively replace values that are ".env" with the matching environment variable.
|
| 18 |
-
"""
|
| 19 |
-
if isinstance(data, dict):
|
| 20 |
-
return {k: _resolve_env_placeholders(v) for k, v in data.items()}
|
| 21 |
-
elif isinstance(data, list):
|
| 22 |
-
return [_resolve_env_placeholders(v) for v in data]
|
| 23 |
-
elif isinstance(data, str) and data.strip() == ".env":
|
| 24 |
-
# Use the key name uppercased from parent call if available
|
| 25 |
-
# This requires that parent dict was processed with context
|
| 26 |
-
return None # will be handled by the parent context
|
| 27 |
-
return data
|
| 28 |
|
| 29 |
def _replace_env_vars(d, parent_key=None):
|
| 30 |
"""
|
| 31 |
-
|
|
|
|
| 32 |
"""
|
| 33 |
if isinstance(d, dict):
|
| 34 |
-
|
| 35 |
-
for k, v in d.items():
|
| 36 |
-
result[k] = _replace_env_vars(v, k)
|
| 37 |
-
return result
|
| 38 |
elif isinstance(d, str) and d.strip() == ".env":
|
| 39 |
-
env_key = parent_key.upper()
|
| 40 |
return os.environ.get(env_key)
|
| 41 |
else:
|
| 42 |
return d
|
| 43 |
|
| 44 |
-
|
|
|
|
| 45 |
local = cfg.get("LocalSystemUrl")
|
| 46 |
if not local:
|
| 47 |
return None
|
|
@@ -56,15 +50,16 @@ def build_amqp_url(cfg):
|
|
| 56 |
return None
|
| 57 |
return f"{scheme}://{user}:{pwd}@{host}:{port}/{vhost_enc}?heartbeat=30"
|
| 58 |
|
|
|
|
| 59 |
@lru_cache
|
| 60 |
-
def get_settings():
|
| 61 |
cfg = _load_json(APPSETTINGS_PATH)
|
| 62 |
cfg = _replace_env_vars(cfg)
|
| 63 |
-
# If AMQP_URL is not directly set, build it
|
| 64 |
if not cfg.get("AMQP_URL"):
|
| 65 |
amqp_url = build_amqp_url(cfg)
|
| 66 |
if amqp_url:
|
| 67 |
cfg["AMQP_URL"] = amqp_url
|
| 68 |
-
return cfg
|
|
|
|
| 69 |
|
| 70 |
settings = get_settings()
|
|
|
|
| 6 |
|
| 7 |
APPSETTINGS_PATH = os.environ.get("APPSETTINGS_JSON", "appsettings.json")
|
| 8 |
|
| 9 |
+
|
| 10 |
+
class Settings:
|
| 11 |
+
"""Simple settings object loaded from appsettings.json + env vars."""
|
| 12 |
+
def __init__(self, data: dict):
|
| 13 |
+
for k, v in data.items():
|
| 14 |
+
setattr(self, k, v)
|
| 15 |
+
|
| 16 |
+
|
| 17 |
def _load_json(path):
|
| 18 |
if not os.path.exists(path):
|
| 19 |
return {}
|
| 20 |
with open(path, "r", encoding="utf-8") as f:
|
| 21 |
return json.load(f)
|
| 22 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 23 |
|
| 24 |
def _replace_env_vars(d, parent_key=None):
|
| 25 |
"""
|
| 26 |
+
Replace any ".env" string with the corresponding environment variable.
|
| 27 |
+
Uses UPPERCASE parent key as env var name.
|
| 28 |
"""
|
| 29 |
if isinstance(d, dict):
|
| 30 |
+
return {k: _replace_env_vars(v, k) for k, v in d.items()}
|
|
|
|
|
|
|
|
|
|
| 31 |
elif isinstance(d, str) and d.strip() == ".env":
|
| 32 |
+
env_key = (parent_key or "").upper()
|
| 33 |
return os.environ.get(env_key)
|
| 34 |
else:
|
| 35 |
return d
|
| 36 |
|
| 37 |
+
|
| 38 |
+
def build_amqp_url(cfg: dict):
|
| 39 |
local = cfg.get("LocalSystemUrl")
|
| 40 |
if not local:
|
| 41 |
return None
|
|
|
|
| 50 |
return None
|
| 51 |
return f"{scheme}://{user}:{pwd}@{host}:{port}/{vhost_enc}?heartbeat=30"
|
| 52 |
|
| 53 |
+
|
| 54 |
@lru_cache
|
| 55 |
+
def get_settings() -> Settings:
|
| 56 |
cfg = _load_json(APPSETTINGS_PATH)
|
| 57 |
cfg = _replace_env_vars(cfg)
|
|
|
|
| 58 |
if not cfg.get("AMQP_URL"):
|
| 59 |
amqp_url = build_amqp_url(cfg)
|
| 60 |
if amqp_url:
|
| 61 |
cfg["AMQP_URL"] = amqp_url
|
| 62 |
+
return Settings(cfg)
|
| 63 |
+
|
| 64 |
|
| 65 |
settings = get_settings()
|