Codex
64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
from .db import db, utc_now
|
|
from .migrations import DEFAULT_SETTINGS
|
|
|
|
|
|
INT_KEYS = {"default_retention_value", "max_concurrent_backups"}
|
|
BOOL_KEYS = {"setup_complete"}
|
|
|
|
|
|
def parse_value(key: str, value: str):
|
|
if key in INT_KEYS:
|
|
return int(value)
|
|
if key in BOOL_KEYS:
|
|
return value.lower() == "true"
|
|
if key == "allowed_cors_origins":
|
|
return [item.strip() for item in value.split(",") if item.strip()]
|
|
return value
|
|
|
|
|
|
def serialize_value(key: str, value) -> str:
|
|
if key in BOOL_KEYS:
|
|
return "true" if bool(value) else "false"
|
|
if key == "allowed_cors_origins":
|
|
if isinstance(value, list):
|
|
return ",".join(str(item).strip() for item in value if str(item).strip())
|
|
return str(value)
|
|
|
|
|
|
def get_settings() -> dict:
|
|
with db() as conn:
|
|
rows = conn.execute("SELECT key, value FROM settings").fetchall()
|
|
raw = {row["key"]: row["value"] for row in rows}
|
|
for key, value in DEFAULT_SETTINGS.items():
|
|
raw.setdefault(key, value)
|
|
return {key: parse_value(key, value) for key, value in raw.items()}
|
|
|
|
|
|
def set_settings(values: dict) -> dict:
|
|
now = utc_now()
|
|
valid_keys = set(DEFAULT_SETTINGS)
|
|
unknown = set(values) - valid_keys
|
|
if unknown:
|
|
raise ValueError(f"Unknown settings: {', '.join(sorted(unknown))}")
|
|
with db() as conn:
|
|
for key, value in values.items():
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO settings(key, value, updated_at)
|
|
VALUES (?, ?, ?)
|
|
ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at
|
|
""",
|
|
(key, serialize_value(key, value), now),
|
|
)
|
|
return get_settings()
|
|
|
|
|
|
def get_setting(key: str, default: str = "") -> str:
|
|
with db() as conn:
|
|
row = conn.execute("SELECT value FROM settings WHERE key = ?", (key,)).fetchone()
|
|
if row:
|
|
return row["value"]
|
|
return default
|