tracker/backup_api.py

"""Authenticated download of a consistent SQLite snapshot of tracker.db (HF backup before redeploy)."""

from __future__ import annotations

import sqlite3
import tempfile
from pathlib import Path

from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import FileResponse
from starlette.background import BackgroundTask

router = APIRouter(prefix="/api/admin", tags=["admin"])


def snapshot_tracker_db_to_temp_file(source: Path) -> Path:
    """Copy the open database to a new file using SQLite backup API (WAL-safe snapshot)."""
    tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".db")
    tmp_path = Path(tmp.name)
    tmp.close()
    dest_conn = sqlite3.connect(str(tmp_path))
    src_conn = sqlite3.connect(str(source))
    src_conn.backup(dest_conn)
    dest_conn.close()
    src_conn.close()
    return tmp_path


@router.get("/tracker-db")
def download_tracker_db(request: Request) -> FileResponse:
    cfg = request.app.state.tracker_config
    source = cfg.tracker_db_full_path
    if not source.is_file():
        raise HTTPException(
            status_code=503,
            detail=f"Tracker database not found at {source}",
        )
    tmp_path = snapshot_tracker_db_to_temp_file(source)
    return FileResponse(
        path=str(tmp_path),
        filename="tracker.db",
        media_type="application/x-sqlite3",
        background=BackgroundTask(tmp_path.unlink),
    )