menzeljandClaude Opus 4.8
8d19b09abd
- Backup/restore: per-stack tar.gz incl. named-volume snapshots (helper container), upload restore with rename/overwrite/conflict detection. - Notifications: ntfy/Discord/Slack/Gotify/generic webhooks, per-event subscriptions; wired into the update checker and stack lifecycle. - Settings page: update-check interval, webhook CRUD + test, user management (with last-admin safeguards). - Audit log page (searchable, paginated). - Mobile-responsive sidebar/layout. Multi-host agents and remote backup destinations (SFTP/S3) deferred. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
46 lines
1.3 KiB
Python
46 lines
1.3 KiB
Python
"""Runtime settings stored in the DB (key/value), with env fallbacks."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any, Optional
|
|
|
|
from sqlmodel import Session, select
|
|
|
|
from config import settings as env_settings
|
|
from database import engine
|
|
from models.setting import Setting
|
|
|
|
KEY_UPDATE_INTERVAL = "update_check_interval_minutes"
|
|
|
|
|
|
def get(session: Session, key: str, default: Any = None) -> Any:
|
|
row = session.get(Setting, key)
|
|
if row is None:
|
|
return default
|
|
try:
|
|
return json.loads(row.value)
|
|
except json.JSONDecodeError:
|
|
return default
|
|
|
|
|
|
def set_value(session: Session, key: str, value: Any) -> None:
|
|
row = session.get(Setting, key)
|
|
encoded = json.dumps(value)
|
|
if row is None:
|
|
session.add(Setting(key=key, value=encoded))
|
|
else:
|
|
row.value = encoded
|
|
session.add(row)
|
|
session.commit()
|
|
|
|
|
|
def get_update_interval(session: Optional[Session] = None) -> int:
|
|
"""Effective update-check interval in minutes (DB override or env default)."""
|
|
if session is None:
|
|
with Session(engine) as own:
|
|
return get_update_interval(own)
|
|
val = get(session, KEY_UPDATE_INTERVAL)
|
|
if isinstance(val, int) and val > 0:
|
|
return val
|
|
return env_settings.UPDATE_CHECK_INTERVAL_MINUTES
|