feat: add backup monitoring CRUD operations

2026-05-11 20:38:13 +02:00
parent bd9765d4d9
commit dd375a7e16
2 changed files with 321 additions and 1 deletions
@@ -9,13 +9,14 @@ from __future__ import annotations

 import json
 import sqlite3
-from io import StringIO
 import time
 import uuid
+from io import StringIO
 from pathlib import Path
 from typing import Any

 import paramiko
+
 from media_library_viewer_api.config import get_settings

 DEFAULT_SETTINGS_PATH = Path(".cache/media_library_viewer/settings.sqlite")
@@ -911,6 +912,299 @@ class SettingsStore:
        with self.connect() as conn:
            conn.execute("DELETE FROM dashboard_shortcuts WHERE id = ?", (shortcut_id,))

+    # ------------------------------------------------------------------
+    # Backup jobs
+    # ------------------------------------------------------------------
+
+    def _row_to_job(self, row: sqlite3.Row) -> dict[str, Any]:
+        return {
+            "id": row["id"],
+            "name": row["name"],
+            "source": row["source"],
+            "target": row["target"],
+            "schedule_interval_seconds": row["schedule_interval_seconds"],
+            "created_at": row["created_at"],
+        }
+
+    def _normalize_backup_job_payload(
+        self, payload: dict[str, Any], job_id: str | None = None
+    ) -> dict[str, Any]:
+        current = self.get_backup_job(job_id) if job_id else None
+        job_id = str(payload.get("id") or job_id or uuid.uuid4().hex[:12]).strip() or uuid.uuid4().hex[:12]
+        name = str(payload.get("name") or (current or {}).get("name") or job_id).strip() or job_id
+        source = str(payload.get("source") if payload.get("source") is not None else (current or {}).get("source", "") or "").strip()
+        target = str(payload.get("target") if payload.get("target") is not None else (current or {}).get("target", "") or "").strip()
+        schedule_interval_seconds = payload.get("schedule_interval_seconds")
+        if schedule_interval_seconds is None:
+            schedule_interval_seconds = (current or {}).get("schedule_interval_seconds")
+        if schedule_interval_seconds is not None:
+            schedule_interval_seconds = int(schedule_interval_seconds)
+        return {
+            "id": job_id,
+            "name": name,
+            "source": source,
+            "target": target,
+            "schedule_interval_seconds": schedule_interval_seconds,
+        }
+
+    def get_backup_job_by_name(self, name: str) -> dict[str, Any] | None:
+        self.init_schema()
+        with self.connect() as conn:
+            row = conn.execute("SELECT * FROM backup_jobs WHERE name = ?", (name,)).fetchone()
+        return self._row_to_job(row) if row else None
+
+    def upsert_backup_job(self, payload: dict[str, Any]) -> dict[str, Any]:
+        self.init_schema()
+        job = self._normalize_backup_job_payload(payload)
+        now = int(time.time())
+        with self.connect() as conn:
+            existing = conn.execute("SELECT created_at FROM backup_jobs WHERE id = ?", (job["id"],)).fetchone()
+            created_at = int(existing[0]) if existing else now
+            conn.execute(
+                """
+                INSERT INTO backup_jobs (id, name, source, target, schedule_interval_seconds, created_at)
+                VALUES (?, ?, ?, ?, ?, ?)
+                ON CONFLICT(id) DO UPDATE SET
+                    name = excluded.name,
+                    source = excluded.source,
+                    target = excluded.target,
+                    schedule_interval_seconds = excluded.schedule_interval_seconds
+                ON CONFLICT(name) DO UPDATE SET
+                    source = excluded.source,
+                    target = excluded.target,
+                    schedule_interval_seconds = excluded.schedule_interval_seconds
+                """,
+                (
+                    job["id"],
+                    job["name"],
+                    job["source"],
+                    job["target"],
+                    job["schedule_interval_seconds"],
+                    created_at,
+                ),
+            )
+        return self.get_backup_job(job["id"]) or job
+
+    def get_backup_job(self, job_id: str | None) -> dict[str, Any] | None:
+        if not job_id:
+            return None
+        self.init_schema()
+        with self.connect() as conn:
+            row = conn.execute("SELECT * FROM backup_jobs WHERE id = ?", (job_id,)).fetchone()
+        return self._row_to_job(row) if row else None
+
+    def list_backup_jobs(self) -> list[dict[str, Any]]:
+        self.init_schema()
+        with self.connect() as conn:
+            rows = conn.execute("SELECT * FROM backup_jobs ORDER BY created_at DESC").fetchall()
+        return [self._row_to_job(row) for row in rows]
+
+    # ------------------------------------------------------------------
+    # Backup runs
+    # ------------------------------------------------------------------
+
+    def _row_to_run(self, row: sqlite3.Row) -> dict[str, Any]:
+        details = json.loads(row["details_json"]) if row["details_json"] else {}
+        return {
+            "id": row["id"],
+            "job_id": row["job_id"],
+            "started_at": row["started_at"],
+            "ended_at": row["ended_at"],
+            "status": row["status"],
+            "bytes_transferred": row["bytes_transferred"],
+            "duration_ms": row["duration_ms"],
+            "error_message": row["error_message"],
+            "details": details,
+            "created_at": row["created_at"],
+        }
+
+    def create_backup_run(self, payload: dict[str, Any]) -> dict[str, Any]:
+        self.init_schema()
+        run_id = str(payload.get("id") or uuid.uuid4().hex[:12]).strip() or uuid.uuid4().hex[:12]
+        now = int(time.time())
+        job_id = str(payload["job_id"])
+        started_at = int(payload["started_at"])
+        ended_at = payload.get("ended_at")
+        if ended_at is not None:
+            ended_at = int(ended_at)
+        status = str(payload.get("status", "pending"))
+        bytes_transferred = payload.get("bytes_transferred")
+        if bytes_transferred is not None:
+            bytes_transferred = int(bytes_transferred)
+        duration_ms = payload.get("duration_ms")
+        if duration_ms is not None:
+            duration_ms = int(duration_ms)
+        error_message = str(payload.get("error_message") or "")
+        details = payload.get("details", {})
+        details_json = json.dumps(details) if details else None
+        with self.connect() as conn:
+            conn.execute(
+                """
+                INSERT INTO backup_runs (id, job_id, started_at, ended_at, status, bytes_transferred, duration_ms, error_message, details_json, created_at)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (run_id, job_id, started_at, ended_at, status, bytes_transferred, duration_ms, error_message, details_json, now),
+            )
+        return self.get_backup_run(run_id) or {
+            "id": run_id,
+            "job_id": job_id,
+            "started_at": started_at,
+            "ended_at": ended_at,
+            "status": status,
+            "bytes_transferred": bytes_transferred,
+            "duration_ms": duration_ms,
+            "error_message": error_message,
+            "details": details,
+            "created_at": now,
+        }
+
+    def get_backup_run(self, run_id: str | None) -> dict[str, Any] | None:
+        if not run_id:
+            return None
+        self.init_schema()
+        with self.connect() as conn:
+            row = conn.execute("SELECT * FROM backup_runs WHERE id = ?", (run_id,)).fetchone()
+        return self._row_to_run(row) if row else None
+
+    def list_backup_runs(
+        self,
+        job_id: str | None = None,
+        status: str | None = None,
+        limit: int = 50,
+    ) -> list[dict[str, Any]]:
+        self.init_schema()
+        clauses: list[str] = []
+        params: list[Any] = []
+        if job_id:
+            clauses.append("job_id = ?")
+            params.append(job_id)
+        if status:
+            clauses.append("status = ?")
+            params.append(status)
+        where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
+        sql = f"SELECT * FROM backup_runs {where} ORDER BY created_at DESC LIMIT ?"
+        params.append(max(1, min(int(limit), 200)))
+        with self.connect() as conn:
+            rows = conn.execute(sql, params).fetchall()
+        return [self._row_to_run(row) for row in rows]
+
+    def get_latest_backup_run(self, job_id: str) -> dict[str, Any] | None:
+        self.init_schema()
+        with self.connect() as conn:
+            row = conn.execute(
+                "SELECT * FROM backup_runs WHERE job_id = ? ORDER BY created_at DESC LIMIT 1",
+                (job_id,),
+            ).fetchone()
+        return self._row_to_run(row) if row else None
+
+    # ------------------------------------------------------------------
+    # Backup alerts
+    # ------------------------------------------------------------------
+
+    def _row_to_alert(self, row: sqlite3.Row) -> dict[str, Any]:
+        return {
+            "id": row["id"],
+            "job_id": row["job_id"],
+            "run_id": row["run_id"],
+            "alert_type": row["alert_type"],
+            "severity": row["severity"],
+            "message": row["message"],
+            "acknowledged": bool(row["acknowledged"]),
+            "resolved_at": row["resolved_at"],
+            "created_at": row["created_at"],
+        }
+
+    def create_backup_alert(self, payload: dict[str, Any]) -> dict[str, Any]:
+        self.init_schema()
+        alert_id = str(payload.get("id") or uuid.uuid4().hex[:12]).strip() or uuid.uuid4().hex[:12]
+        now = int(time.time())
+        job_id = str(payload["job_id"])
+        run_id = payload.get("run_id")
+        if run_id is not None:
+            run_id = str(run_id)
+        alert_type = str(payload.get("alert_type", "generic"))
+        severity = str(payload.get("severity", "warning"))
+        message = str(payload.get("message", ""))
+        with self.connect() as conn:
+            conn.execute(
+                """
+                INSERT INTO backup_alerts (id, job_id, run_id, alert_type, severity, message, acknowledged, resolved_at, created_at)
+                VALUES (?, ?, ?, ?, ?, ?, 0, NULL, ?)
+                """,
+                (alert_id, job_id, run_id, alert_type, severity, message, now),
+            )
+        return self.get_backup_alert(alert_id) or {
+            "id": alert_id,
+            "job_id": job_id,
+            "run_id": run_id,
+            "alert_type": alert_type,
+            "severity": severity,
+            "message": message,
+            "acknowledged": False,
+            "resolved_at": None,
+            "created_at": now,
+        }
+
+    def get_backup_alert(self, alert_id: str | None) -> dict[str, Any] | None:
+        if not alert_id:
+            return None
+        self.init_schema()
+        with self.connect() as conn:
+            row = conn.execute("SELECT * FROM backup_alerts WHERE id = ?", (alert_id,)).fetchone()
+        return self._row_to_alert(row) if row else None
+
+    def list_backup_alerts(
+        self,
+        job_id: str | None = None,
+        acknowledged: bool | None = None,
+        severity: str | None = None,
+    ) -> list[dict[str, Any]]:
+        self.init_schema()
+        clauses: list[str] = []
+        params: list[Any] = []
+        if job_id:
+            clauses.append("job_id = ?")
+            params.append(job_id)
+        if acknowledged is not None:
+            clauses.append("acknowledged = ?")
+            params.append(1 if acknowledged else 0)
+        if severity:
+            clauses.append("severity = ?")
+            params.append(severity)
+        where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
+        sql = f"SELECT * FROM backup_alerts {where} ORDER BY created_at DESC"
+        with self.connect() as conn:
+            rows = conn.execute(sql, params).fetchall()
+        return [self._row_to_alert(row) for row in rows]
+
+    def acknowledge_backup_alert(self, alert_id: str) -> dict[str, Any] | None:
+        self.init_schema()
+        with self.connect() as conn:
+            conn.execute(
+                "UPDATE backup_alerts SET acknowledged = 1 WHERE id = ?",
+                (alert_id,),
+            )
+        return self.get_backup_alert(alert_id)
+
+    def resolve_backup_alerts_for_job(self, job_id: str, alert_type: str) -> int:
+        self.init_schema()
+        now = int(time.time())
+        with self.connect() as conn:
+            cur = conn.execute(
+                "UPDATE backup_alerts SET resolved_at = ? WHERE job_id = ? AND alert_type = ? AND resolved_at IS NULL",
+                (now, job_id, alert_type),
+            )
+            return int(cur.rowcount or 0)
+
+    def prune_backup_alerts(self, cutoff_ts: int) -> int:
+        self.init_schema()
+        with self.connect() as conn:
+            cur = conn.execute(
+                "DELETE FROM backup_alerts WHERE created_at < ?",
+                (int(cutoff_ts),),
+            )
+            return int(cur.rowcount or 0)
+

 _store: SettingsStore | None = None

@@ -29,3 +29,29 @@ def test_backup_report_model():
    )
    assert report.name == "test-backup"
    assert report.status == "success"
+
+
+def test_create_job_and_run():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        db_path = Path(tmpdir) / "test_settings.sqlite"
+        store = SettingsStore(db_path)
+        store.init_schema()
+
+        job = store.upsert_backup_job({
+            "name": "test-job",
+            "source": "server-a:/data",
+            "target": "server-b:/backups",
+            "schedule_interval_seconds": 86400,
+        })
+        assert job["name"] == "test-job"
+
+        run = store.create_backup_run({
+            "job_id": job["id"],
+            "started_at": 1715392800,
+            "status": "success",
+            "ended_at": 1715393700,
+            "duration_ms": 900000,
+            "bytes_transferred": 1024,
+        })
+        assert run["job_id"] == job["id"]
+        assert run["status"] == "success"