File indexing completed on 2026-06-26 08:40:21
0001 """Engine config - engine-level + DB + email.
0002
0003 Alarm configurations live in the DB as kind='alarm' entries, not in TOML.
0004 Operators edit them via the /swf-monitor/alarms/<name>/edit/ UI.
0005 """
0006 from __future__ import annotations
0007
0008 import os
0009 import tomllib
0010 from dataclasses import dataclass
0011 from pathlib import Path
0012 from urllib.parse import quote_plus
0013
0014
0015 @dataclass
0016 class EngineConfig:
0017 service_base_url: str
0018 request_timeout: int = 20
0019 log_path: str | None = None
0020
0021
0022 @dataclass
0023 class EmailConfig:
0024 provider: str
0025 region: str
0026 from_addr: str
0027
0028
0029 @dataclass
0030 class Config:
0031 engine: EngineConfig
0032 email: EmailConfig
0033 db_dsn: str
0034 raw: dict
0035
0036
0037 def _parse_dotenv(path: str) -> dict:
0038 out: dict[str, str] = {}
0039 if not os.path.exists(path):
0040 return out
0041 with open(path) as f:
0042 for line in f:
0043 line = line.strip()
0044 if not line or line.startswith('#') or '=' not in line:
0045 continue
0046 k, v = line.split('=', 1)
0047 v = v.strip()
0048 if len(v) >= 2 and v[0] == v[-1] and v[0] in ('"', "'"):
0049 v = v[1:-1]
0050 out[k.strip()] = v
0051 return out
0052
0053
0054 def _compose_dsn(db_section: dict) -> str:
0055 if db_section.get('dsn'):
0056 return str(db_section['dsn'])
0057
0058 env_path = os.path.expanduser(
0059 db_section.get('env_path', '/opt/swf-monitor/config/env/production.env'))
0060 env = _parse_dotenv(env_path)
0061
0062 def pick(k, ek, default=None):
0063 if db_section.get(k) is not None:
0064 return db_section[k]
0065 if env.get(ek) is not None:
0066 return env[ek]
0067 if k in os.environ:
0068 return os.environ[k]
0069 return default
0070
0071 host = pick('host', 'DB_HOST', 'localhost')
0072 port = pick('port', 'DB_PORT', '5432')
0073 name = pick('name', 'DB_NAME', 'swfdb')
0074 user = pick('user', 'DB_USER', 'admin')
0075 password = pick('password', 'DB_PASSWORD', '')
0076
0077 userinfo = quote_plus(str(user))
0078 if password:
0079 userinfo += ':' + quote_plus(str(password))
0080 return f"postgresql://{userinfo}@{host}:{port}/{name}"
0081
0082
0083 def load(path: str | Path) -> Config:
0084 with open(path, 'rb') as f:
0085 raw = tomllib.load(f)
0086 eng = raw['engine']
0087 service_base_url = (
0088 eng.get('service_base_url') or eng.get('swf_remote_base_url') or ''
0089 ).rstrip('/')
0090 if not service_base_url:
0091 raise KeyError('engine.service_base_url')
0092 engine = EngineConfig(
0093 service_base_url=service_base_url,
0094 request_timeout=int(eng.get('request_timeout', 20)),
0095 log_path=os.path.expanduser(eng['log_path']) if eng.get('log_path') else None,
0096 )
0097 e = raw['email']
0098 email = EmailConfig(provider=e['provider'], region=e['region'],
0099 from_addr=e['from'])
0100 db_dsn = _compose_dsn(raw.get('db', {}))
0101 return Config(engine=engine, email=email, db_dsn=db_dsn, raw=raw)
