File size: 3,033 Bytes
b2c2f23 db550a4 b2c2f23 db550a4 b2c2f23 7f43efb afb8095 0fe181b db550a4 9bce4c7 0fe181b 9bce4c7 5cd52c5 9bce4c7 d6828d2 db550a4 d6828d2 db550a4 d6828d2 0fe181b d0eeff5 0fe181b d6828d2 9bce4c7 d6828d2 d0eeff5 d6828d2 db550a4 9bce4c7 d0eeff5 0fe181b d6828d2 0fe181b c5e43db d6828d2 db550a4 9bce4c7 b2c2f23 9bce4c7 d6828d2 728cbca 0fe181b 728cbca 0fe181b 728cbca 0fe181b fa17022 0fe181b 17e2644 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 89 90 91 92 93 94 95 |
# app/config.py
import json
import os
from functools import lru_cache
from urllib.parse import quote
# Make sure ZeroGPU sees it
if "HF_TOKEN" in os.environ:
os.environ["HUGGINGFACE_HUB_TOKEN"] = os.environ["HF_TOKEN"]
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 the same key name
as in appsettings.json (case-sensitive).
"""
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 ""
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) # <- replaces .env everywhere first
# 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") # already resolved
cfg["RABBIT_VHOST"] = local.get("RabbitVHost")
cfg["RABBIT_USE_TLS"] = local.get("UseTls")
# Map JSON keys to Python attributes
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("RabbitMQ", {}).get("ExchangeTypes", cfg.get("EXCHANGE_TYPES", {}))
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()
|