57 lines
2.1 KiB
Python
57 lines
2.1 KiB
Python
"""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})
|