"""Admin-only backup: produce and download a restorable archive. The archive is built on demand rather than served from a directory, so it always reflects the database at the moment of the request. """ from __future__ import annotations import shutil import tempfile from pathlib import Path from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.responses import FileResponse, JSONResponse from starlette.background import BackgroundTask from ..auth import Ctx from ..backup import export_backup from ..db import record_audit from ..deps import get_ctx router = APIRouter() @router.get("/v1/backup") def download_backup(request: Request, ctx: Ctx = Depends(get_ctx)): """Admin: stream a full backup archive (database rows + plugin artifacts).""" admin = ctx.require_admin(request) # Build in the system temp dir, not data/backups: this one is a response # body, and leaving a copy behind on every click would fill the disk. tmpdir = Path(tempfile.mkdtemp(prefix="dsh-backup-")) try: archive = export_backup(ctx.cfg, tmpdir / "dsh-backup.tar.gz") except Exception as e: # noqa: BLE001 - surface any failure as a 500 with context raise HTTPException(500, f"backup failed: {e}") record_audit(ctx.conn, admin["id"], "backup.download", {"size_bytes": archive.stat().st_size}) return FileResponse( archive, media_type="application/gzip", filename="dsh-backup.tar.gz", background=BackgroundTask(shutil.rmtree, tmpdir, ignore_errors=True), ) @router.get("/v1/backup/info") def backup_info(request: Request, ctx: Ctx = Depends(get_ctx)): """Admin: what a backup would contain, without building one.""" ctx.require_admin(request) counts = {} for table in ("users", "tokens", "settings", "plugins", "audit_log"): counts[table] = ctx.conn.execute(f"SELECT COUNT(*) AS c FROM {table}").fetchone()["c"] artifacts = 0 for p in ctx.cfg.plugins_dir.rglob("*.tar.gz"): if p.is_file(): artifacts += 1 return JSONResponse({"row_counts": counts, "artifacts": artifacts})