54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
from typing import Optional
|
|
from sqlalchemy.orm import Session
|
|
from models import Setting
|
|
from config import get_settings
|
|
|
|
_settings_cache = {}
|
|
|
|
|
|
def get_setting(db: Session, key: str) -> Optional[str]:
|
|
"""Get setting from DB, fallback to .env config."""
|
|
s = db.query(Setting).filter(Setting.key == key).first()
|
|
if s:
|
|
return s.value
|
|
|
|
config = get_settings()
|
|
return getattr(config, key, None)
|
|
|
|
|
|
def set_setting(db: Session, key: str, value: Optional[str]) -> Setting:
|
|
s = db.query(Setting).filter(Setting.key == key).first()
|
|
if s:
|
|
s.value = value
|
|
else:
|
|
s = Setting(key=key, value=value)
|
|
db.add(s)
|
|
db.commit()
|
|
db.refresh(s)
|
|
_settings_cache[key] = value
|
|
return s
|
|
|
|
|
|
def get_all_settings(db: Session) -> dict:
|
|
"""Return merged DB settings and .env defaults."""
|
|
config = get_settings()
|
|
result = {}
|
|
keys = [
|
|
"AGENT_TOKEN",
|
|
"OFFLINE_THRESHOLD_MINUTES",
|
|
"NOTIFICATION_WEBHOOK_URL",
|
|
"NOTIFICATION_TELEGRAM_BOT_TOKEN",
|
|
"NOTIFICATION_TELEGRAM_CHAT_ID",
|
|
"TIMEZONE",
|
|
]
|
|
for key in keys:
|
|
db_val = db.query(Setting).filter(Setting.key == key).first()
|
|
value = db_val.value if db_val else getattr(config, key, None)
|
|
if key == "OFFLINE_THRESHOLD_MINUTES" and value is not None:
|
|
try:
|
|
value = int(value)
|
|
except (TypeError, ValueError):
|
|
value = getattr(config, key, None)
|
|
result[key] = value
|
|
return result
|