Add backend API, running status and 30-day run history

- FastAPI + SQLite backend: tenants, jobs and runs, with an idempotent
  /api/v1/ingest endpoint authenticated by per-collector API keys
- Runs store started_at and finished_at; "Last Run" is the most recent
  finish, and in-progress runs show as Running
- Job modal shows a 30-day daily strip, success rate, and a filterable
  run list with per-run messages
- nginx: HTTPS plus basic auth, ingest endpoint exempt; systemd unit
- manage.py for API keys, job cleanup and demo data; pytest suite

Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
2026-10-04 21:53:24 -06:00
co-authored by Claude Opus 5.5
parent 44e3065efa
commit bf752a904f
25 changed files with 2148 additions and 1 deletions
View File
+42
View File
@@ -0,0 +1,42 @@
"""Collector (machine) authentication.
Dashboard *users* are authenticated by nginx in front of the app (basic auth today,
oauth2-proxy + Microsoft Entra ID later), so nothing here deals with people.
"""
import hashlib
import secrets
import sqlite3
from fastapi import Depends, Header, HTTPException
from .db import get_db
from .timeutil import to_db, utcnow
KEY_PREFIX = "bdk_"
def generate_key() -> str:
return KEY_PREFIX + secrets.token_urlsafe(32)
def hash_key(key: str) -> str:
# Keys are 256-bit random, so a fast unsalted hash is fine (no dictionary to attack).
return hashlib.sha256(key.encode()).hexdigest()
def require_ingest_key(
authorization: str | None = Header(None),
conn: sqlite3.Connection = Depends(get_db),
) -> str:
scheme, _, token = (authorization or "").partition(" ")
token = token.strip()
if scheme.lower() != "bearer" or not token:
raise HTTPException(401, "Missing bearer token", headers={"WWW-Authenticate": "Bearer"})
row = conn.execute(
"SELECT id, name FROM api_keys WHERE key_hash = ? AND revoked_at IS NULL", (hash_key(token),)
).fetchone()
if row is None:
raise HTTPException(401, "Invalid or revoked API key", headers={"WWW-Authenticate": "Bearer"})
conn.execute("UPDATE api_keys SET last_used_at = ? WHERE id = ?", (to_db(utcnow()), row["id"]))
return row["name"]
+99
View File
@@ -0,0 +1,99 @@
import sqlite3
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path
from fastapi import Request
SCHEMA_VERSION = 1
SCHEMA = """
CREATE TABLE IF NOT EXISTS tenants (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL UNIQUE COLLATE NOCASE
);
-- A job is identified by (tenant, source system, the source's own job id), so two
-- products can't collide and renaming a job in Veeam doesn't create a duplicate.
CREATE TABLE IF NOT EXISTS jobs (
id INTEGER PRIMARY KEY,
tenant_id INTEGER NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
source_system TEXT NOT NULL,
external_id TEXT NOT NULL,
name TEXT NOT NULL,
next_run_at TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
UNIQUE (tenant_id, source_system, external_id)
);
CREATE TABLE IF NOT EXISTS runs (
id INTEGER PRIMARY KEY,
job_id INTEGER NOT NULL REFERENCES jobs(id) ON DELETE CASCADE,
external_id TEXT NOT NULL,
status TEXT NOT NULL CHECK (status IN ('running', 'success', 'warning', 'failed')),
started_at TEXT NOT NULL,
finished_at TEXT,
message TEXT NOT NULL DEFAULT '',
size_bytes INTEGER,
details TEXT NOT NULL DEFAULT '{}',
received_at TEXT NOT NULL,
CHECK ((status = 'running') = (finished_at IS NULL)),
UNIQUE (job_id, external_id)
);
CREATE INDEX IF NOT EXISTS runs_job_started ON runs (job_id, started_at);
CREATE INDEX IF NOT EXISTS runs_job_finished ON runs (job_id, finished_at);
-- Keys used by collectors to push results. Only the SHA-256 is stored.
CREATE TABLE IF NOT EXISTS api_keys (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL UNIQUE,
key_hash TEXT NOT NULL UNIQUE,
created_at TEXT NOT NULL,
last_used_at TEXT,
revoked_at TEXT
);
"""
def connect(path: Path | str) -> sqlite3.Connection:
# Autocommit mode; writes that must be atomic go through transaction().
# check_same_thread=False: FastAPI may run a dependency and its endpoint on different
# threadpool threads, but a connection is only ever used by one request at a time.
conn = sqlite3.connect(path, timeout=10, isolation_level=None, check_same_thread=False)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
conn.execute("PRAGMA busy_timeout = 10000")
return conn
def init_db(path: Path | str) -> None:
conn = connect(path)
try:
conn.execute("PRAGMA journal_mode = WAL")
version = conn.execute("PRAGMA user_version").fetchone()[0]
if version > SCHEMA_VERSION:
raise RuntimeError(f"Database schema v{version} is newer than this app (v{SCHEMA_VERSION})")
conn.executescript(SCHEMA)
conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
finally:
conn.close()
@contextmanager
def transaction(conn: sqlite3.Connection) -> Iterator[None]:
conn.execute("BEGIN IMMEDIATE")
try:
yield
except BaseException:
conn.execute("ROLLBACK")
raise
conn.execute("COMMIT")
def get_db(request: Request) -> Iterator[sqlite3.Connection]:
conn = connect(request.app.state.settings.db_path)
try:
yield conn
finally:
conn.close()
+118
View File
@@ -0,0 +1,118 @@
import asyncio
import logging
import sqlite3
from contextlib import asynccontextmanager
from datetime import timedelta
from pathlib import Path
from fastapi import APIRouter, Depends, FastAPI, HTTPException, Query, Request
from . import db, repo
from .auth import require_ingest_key
from .models import IngestIn
from .settings import HISTORY_DAYS, Settings, load_settings
from .timeutil import to_db, utcnow
log = logging.getLogger("backup_dashboard")
PURGE_INTERVAL_SECONDS = 6 * 3600
router = APIRouter(prefix="/api/v1")
@router.get("/health")
def health(conn: sqlite3.Connection = Depends(db.get_db)):
conn.execute("SELECT 1")
return {"status": "ok"}
@router.get("/jobs")
def list_jobs(conn: sqlite3.Connection = Depends(db.get_db)):
return {"generated_at": to_db(utcnow()), "jobs": repo.list_jobs(conn)}
@router.get("/jobs/{job_id}")
def get_job(job_id: int, conn: sqlite3.Connection = Depends(db.get_db)):
job = repo.get_job(conn, job_id)
if job is None:
raise HTTPException(404, "Job not found")
return job
@router.get("/jobs/{job_id}/runs")
def list_runs(
job_id: int,
days: int = Query(HISTORY_DAYS, ge=1, le=HISTORY_DAYS),
conn: sqlite3.Connection = Depends(db.get_db),
):
if conn.execute("SELECT 1 FROM jobs WHERE id = ?", (job_id,)).fetchone() is None:
raise HTTPException(404, "Job not found")
since = utcnow() - timedelta(days=days)
return {"job_id": job_id, "since": to_db(since), "runs": repo.list_runs(conn, job_id, since)}
@router.get("/runs/{run_id}")
def get_run(run_id: int, conn: sqlite3.Connection = Depends(db.get_db)):
run = repo.get_run(conn, run_id)
if run is None:
raise HTTPException(404, "Run not found")
return run
@router.post("/ingest")
def ingest(
item: IngestIn,
key_name: str = Depends(require_ingest_key),
conn: sqlite3.Connection = Depends(db.get_db),
):
job_id, run_id = repo.ingest(conn, item)
log.info("ingest key=%s tenant=%r job=%r run=%s", key_name, item.tenant, item.job.name, run_id)
return {"job_id": job_id, "run_id": run_id}
async def _purge_loop(settings: Settings) -> None:
def purge() -> int:
conn = db.connect(settings.db_path)
try:
return repo.purge_old_runs(conn, settings.retention_days)
finally:
conn.close()
while True:
try:
if deleted := await asyncio.to_thread(purge):
log.info("purged %d runs older than %d days", deleted, settings.retention_days)
except Exception:
log.exception("run purge failed")
await asyncio.sleep(PURGE_INTERVAL_SECONDS)
def create_app(settings: Settings | None = None) -> FastAPI:
settings = settings or load_settings()
@asynccontextmanager
async def lifespan(app: FastAPI):
db.init_db(settings.db_path)
task = asyncio.create_task(_purge_loop(settings))
try:
yield
finally:
task.cancel()
# Interactive docs are off: nginx's strict CSP would block their CDN assets anyway.
app = FastAPI(
title="Backup Dashboard API",
lifespan=lifespan,
docs_url=None,
redoc_url=None,
openapi_url="/api/v1/openapi.json",
)
app.state.settings = settings
app.include_router(router)
if settings.serve_static:
from fastapi.staticfiles import StaticFiles
public = Path(__file__).resolve().parents[2] / "public"
app.mount("/", StaticFiles(directory=public, html=True), name="static")
return app
+63
View File
@@ -0,0 +1,63 @@
import json
from typing import Annotated, Any, Literal
from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, StringConstraints, field_validator, model_validator
RunStatus = Literal["running", "success", "warning", "failed"]
Name = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=200)]
MAX_MESSAGE_CHARS = 32_000
MAX_DETAILS_BYTES = 16_000
class Strict(BaseModel):
# Reject unknown fields so a typo in a collector script fails loudly instead of being dropped.
model_config = ConfigDict(extra="forbid")
class JobIn(Strict):
external_id: Name = Field(description="The source system's own stable ID for the job")
name: Name
# Omit to leave the stored value untouched; send null to clear it.
next_run_at: AwareDatetime | None = None
class RunIn(Strict):
external_id: Name | None = Field(
None, description="The source's ID for this run. Defaults to started_at, so a 'running' "
"report and its later completion report merge into one run.")
status: RunStatus
started_at: AwareDatetime
# Omit for a finished run to use the time the report was received.
finished_at: AwareDatetime | None = None
message: str = ""
size_bytes: int | None = Field(None, ge=0)
details: dict[str, Any] = Field(default_factory=dict)
@field_validator("message")
@classmethod
def truncate_message(cls, v: str) -> str:
# Backup logs can be huge; keep the head, which is where the summary usually is.
return v if len(v) <= MAX_MESSAGE_CHARS else v[:MAX_MESSAGE_CHARS] + "\n… [truncated]"
@field_validator("details")
@classmethod
def limit_details(cls, v: dict[str, Any]) -> dict[str, Any]:
if len(json.dumps(v)) > MAX_DETAILS_BYTES:
raise ValueError(f"details must serialize to at most {MAX_DETAILS_BYTES} bytes")
return v
@model_validator(mode="after")
def check_times(self) -> "RunIn":
if self.status == "running" and self.finished_at is not None:
raise ValueError("a running run cannot have finished_at")
if self.finished_at is not None and self.finished_at < self.started_at:
raise ValueError("finished_at is before started_at")
return self
class IngestIn(Strict):
tenant: Name = Field(description="Company name as shown on the dashboard")
source_system: Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=100)]
job: JobIn
run: RunIn | None = Field(None, description="Omit to only register/update the job (e.g. next_run_at)")
+162
View File
@@ -0,0 +1,162 @@
import json
import sqlite3
from datetime import datetime, timedelta
from typing import Any
from .db import transaction
from .models import IngestIn
from .timeutil import to_db, utcnow
# One row per job: `cur` is the most recent run (may be in progress), `fin` the most recent
# finished one. "Last run" on the dashboard is fin.finished_at, so a job that's currently
# running still shows when its previous run completed.
_JOB_SQL = """
SELECT j.id, t.name AS company_name, j.name AS job_name, j.source_system, j.next_run_at,
cur.status,
CASE WHEN cur.status = 'running' THEN cur.started_at END AS running_since,
fin.status AS last_status,
fin.started_at AS last_started_at,
fin.finished_at AS last_run_at,
unixepoch(fin.finished_at) - unixepoch(fin.started_at) AS duration_seconds,
fin.size_bytes,
fin.details,
COALESCE(fin.message, cur.message, '') AS message
FROM jobs j
JOIN tenants t ON t.id = j.tenant_id
LEFT JOIN runs cur ON cur.id = (
SELECT r.id FROM runs r WHERE r.job_id = j.id
ORDER BY r.started_at DESC, r.id DESC LIMIT 1)
LEFT JOIN runs fin ON fin.id = (
SELECT r.id FROM runs r WHERE r.job_id = j.id AND r.finished_at IS NOT NULL
ORDER BY r.finished_at DESC, r.id DESC LIMIT 1)
"""
_SUMMARY_FIELDS = (
"id", "company_name", "job_name", "source_system", "status",
"running_since", "last_run_at", "next_run_at", "message",
)
def list_jobs(conn: sqlite3.Connection) -> list[dict[str, Any]]:
rows = conn.execute(_JOB_SQL + " ORDER BY t.name, j.name").fetchall()
return [{k: row[k] for k in _SUMMARY_FIELDS} for row in rows]
def get_job(conn: sqlite3.Connection, job_id: int) -> dict[str, Any] | None:
row = conn.execute(_JOB_SQL + " WHERE j.id = ?", (job_id,)).fetchone()
if row is None:
return None
job = dict(row)
job["details"] = json.loads(job["details"]) if job["details"] else {}
return job
def list_runs(conn: sqlite3.Connection, job_id: int, since: datetime) -> list[dict[str, Any]]:
# Messages/details are left out: an hourly job has ~720 runs a month and logs can be large.
# The UI fetches a single run's full record on demand.
rows = conn.execute(
"""
SELECT id, status, started_at, finished_at,
unixepoch(finished_at) - unixepoch(started_at) AS duration_seconds, size_bytes
FROM runs WHERE job_id = ? AND started_at >= ?
ORDER BY started_at DESC, id DESC
""",
(job_id, to_db(since)),
).fetchall()
return [dict(r) for r in rows]
def get_run(conn: sqlite3.Connection, run_id: int) -> dict[str, Any] | None:
row = conn.execute(
"""
SELECT id, job_id, status, started_at, finished_at,
unixepoch(finished_at) - unixepoch(started_at) AS duration_seconds,
size_bytes, message, details
FROM runs WHERE id = ?
""",
(run_id,),
).fetchone()
if row is None:
return None
run = dict(row)
run["details"] = json.loads(run["details"])
return run
def ingest(conn: sqlite3.Connection, item: IngestIn) -> tuple[int, int | None]:
"""Upsert tenant → job → run. Safe to call repeatedly with the same payload."""
now = utcnow()
now_s = to_db(now)
with transaction(conn):
conn.execute("INSERT INTO tenants (name) VALUES (?) ON CONFLICT (name) DO NOTHING", (item.tenant,))
tenant_id = conn.execute("SELECT id FROM tenants WHERE name = ?", (item.tenant,)).fetchone()[0]
job_id = conn.execute(
"""
INSERT INTO jobs (tenant_id, source_system, external_id, name, next_run_at, created_at, updated_at)
VALUES (:tenant_id, :source, :ext, :name, :next_run, :now, :now)
ON CONFLICT (tenant_id, source_system, external_id) DO UPDATE SET
name = excluded.name,
next_run_at = CASE WHEN :set_next_run THEN excluded.next_run_at ELSE jobs.next_run_at END,
updated_at = excluded.updated_at
RETURNING id
""",
{
"tenant_id": tenant_id,
"source": item.source_system,
"ext": item.job.external_id,
"name": item.job.name,
"next_run": to_db(item.job.next_run_at) if item.job.next_run_at else None,
"set_next_run": "next_run_at" in item.job.model_fields_set,
"now": now_s,
},
).fetchone()[0]
if item.run is None:
return job_id, None
run = item.run
started = to_db(run.started_at)
finished = None
if run.status != "running":
finished = to_db(run.finished_at or max(now, run.started_at))
params = {
"job_id": job_id,
"ext": run.external_id or started,
"status": run.status,
"started": started,
"finished": finished,
"message": run.message,
"size": run.size_bytes,
"details": json.dumps(run.details),
"now": now_s,
}
# The WHERE stops a late/retried "running" report from reopening a run that already finished.
row = conn.execute(
"""
INSERT INTO runs (job_id, external_id, status, started_at, finished_at, message,
size_bytes, details, received_at)
VALUES (:job_id, :ext, :status, :started, :finished, :message, :size, :details, :now)
ON CONFLICT (job_id, external_id) DO UPDATE SET
status = excluded.status,
started_at = excluded.started_at,
finished_at = excluded.finished_at,
message = excluded.message,
size_bytes = excluded.size_bytes,
details = excluded.details,
received_at = excluded.received_at
WHERE NOT (runs.status != 'running' AND excluded.status = 'running')
RETURNING id
""",
params,
).fetchone()
if row is None:
row = conn.execute(
"SELECT id FROM runs WHERE job_id = :job_id AND external_id = :ext", params
).fetchone()
return job_id, row[0]
def purge_old_runs(conn: sqlite3.Connection, retention_days: int) -> int:
cutoff = to_db(utcnow() - timedelta(days=retention_days))
return conn.execute("DELETE FROM runs WHERE started_at < ?", (cutoff,)).rowcount
+22
View File
@@ -0,0 +1,22 @@
import os
from dataclasses import dataclass
from pathlib import Path
# The modal shows this many days of run history; retention can never be shorter.
HISTORY_DAYS = 30
@dataclass(frozen=True)
class Settings:
db_path: Path
retention_days: int = 90
# Dev only: serve ../public from the API so you can run without nginx.
serve_static: bool = False
def load_settings() -> Settings:
return Settings(
db_path=Path(os.environ.get("BACKUP_DASH_DB", "backup-dashboard.db")),
retention_days=max(HISTORY_DAYS, int(os.environ.get("BACKUP_DASH_RETENTION_DAYS", "90"))),
serve_static=os.environ.get("BACKUP_DASH_SERVE_STATIC") == "1",
)
+12
View File
@@ -0,0 +1,12 @@
from datetime import UTC, datetime
# Every timestamp in the DB uses this exact format, so string comparison == chronological comparison.
DB_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
def utcnow() -> datetime:
return datetime.now(UTC).replace(microsecond=0)
def to_db(dt: datetime) -> str:
return dt.astimezone(UTC).strftime(DB_FORMAT)