Spaces:
Running
Running
File size: 2,399 Bytes
e0c9c7c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | """Protected diagnostics for the public dashboard snapshot.
The route is MACHINE-only and returns a sanitized Supabase error code so an
operator can distinguish missing schema, permissions, and connectivity without
exposing URLs, keys, SQL, or row contents.
"""
from __future__ import annotations
import re
from typing import Any, Optional
from fastapi import APIRouter, Depends
from .auth_guard import AuthRole, require_role
from .public_snapshot import get_snapshot_client
router = APIRouter(
prefix="/api/diagnostics/public-snapshot",
tags=["diagnostics"],
dependencies=[Depends(require_role(AuthRole.MACHINE))],
)
_PUBLIC_FIELDS = "singleton,service_status,active_sessions,queued_tasks,in_progress_tasks,app_version,updated_at"
def _safe_error(exc: Exception) -> dict[str, Optional[str]]:
code = getattr(exc, "code", None) or getattr(exc, "status_code", None)
raw = str(getattr(exc, "message", None) or exc)
# Keep useful PostgreSQL/Supabase codes, remove URLs, credentials and long SQL.
message = re.sub(r"https?://\S+", "[url]", raw)
message = re.sub(r"(?i)(authorization|apikey|api[_-]?key|token|password)=?\s*\S+", r"\1=[redacted]", message)
message = re.sub(r"\s+", " ", message).strip()[:240]
return {
"code": str(code) if code is not None else None,
"type": type(exc).__name__,
"message": message,
}
@router.get("")
async def public_snapshot_diagnostics() -> dict[str, Any]:
"""Return snapshot presence and a sanitized exact Supabase error, if any."""
client = get_snapshot_client()
if client is None:
return {
"ok": False,
"client_available": False,
"row_present": False,
"error": {"code": None, "type": "SupabaseClientUnavailable", "message": "client unavailable"},
}
try:
result = client.table("public_dashboard_snapshot").select(_PUBLIC_FIELDS).eq("singleton", True).limit(1).execute()
rows = result.data or []
return {
"ok": True,
"client_available": True,
"row_present": bool(rows),
"snapshot": rows[0] if rows else None,
"error": None,
}
except Exception as exc:
return {
"ok": False,
"client_available": True,
"row_present": False,
"error": _safe_error(exc),
}
|