Back to home page

EIC code displayed by LXR

 
 

    


File indexing completed on 2026-04-27 07:41:44

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