| |
| import json |
| import os |
| from functools import lru_cache |
| from urllib.parse import quote |
|
|
| |
| 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) |
|
|
| |
| 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") |
| cfg["RABBIT_VHOST"] = local.get("RabbitVHost") |
| cfg["RABBIT_USE_TLS"] = local.get("UseTls") |
|
|
| |
| 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) |
| |
|
|
| |
| 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() |
|
|
