File size: 2,834 Bytes
b2c2f23 db550a4 b2c2f23 db550a4 b2c2f23 0fe181b db550a4 9bce4c7 0fe181b 9bce4c7 d6828d2 db550a4 d6828d2 db550a4 d6828d2 0fe181b d6828d2 9bce4c7 d6828d2 0fe181b d6828d2 db550a4 9bce4c7 0fe181b d6828d2 0fe181b c5e43db d6828d2 db550a4 9bce4c7 b2c2f23 9bce4c7 d6828d2 0fe181b d6828d2 0fe181b d6828d2 0fe181b 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 82 83 84 85 86 87 88 |
# 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:
"""Settings object with attribute access."""
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 ".env" values with env var lookups using uppercase key 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(local: dict):
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)
# Flatten relevant LocalSystemUrl fields
local = cfg.get("LocalSystemUrl", {})
if local:
cfg["RABBIT_INSTANCE_NAME"] = local.get("RabbitInstanceName")
cfg["RABBIT_HOST_NAME"] = local.get("RabbitHostName")
cfg["RABBIT_PORT"] = local.get("RabbitPort")
cfg["RABBIT_USER_NAME"] = local.get("RabbitUserName")
cfg["RABBIT_PASSWORD"] = local.get("RabbitPassword") or os.environ.get("RABBIT_PASSWORD")
cfg["RABBIT_VHOST"] = local.get("RabbitVHost")
cfg["RABBIT_USE_TLS"] = local.get("UseTls")
# Map JSON keys to what Python code expects
cfg["SERVICE_ID"] = cfg.get("ServiceID")
cfg["RABBIT_ROUTING_KEY"] = cfg.get("RabbitRoutingKey")
cfg["RABBIT_EXCHANGE_TYPE"] = cfg.get("RabbitExhangeType") or "topic"
cfg["REDIS_URL"] = cfg.get("RedisUrl")
cfg["REDIS_SECRET"] = cfg.get("RedisSecret")
cfg["EXCHANGE_TYPES"] = cfg.get("EXCHANGE_TYPES", {}) # default empty dict
cfg["RABBIT_PREFETCH"] = cfg.get("RABBIT_PREFETCH", 1)
# Build AMQP_URL if not already provided
if not cfg.get("AMQP_URL"):
amqp_url = _build_amqp_url(local)
if amqp_url:
cfg["AMQP_URL"] = amqp_url
return Settings(cfg)
settings = get_settings()
|