Spaces:
Running
Running
File size: 2,413 Bytes
c8365f5 | 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 70 71 72 73 | """DTO pubblico e sanificato dello stato del servizio.
Questa route non legge agent_tasks, sessioni operative o log. La tabella
public_dashboard_snapshot viene aggiornata dal backend con service_role e letta
qui tramite una whitelist di campi.
"""
from __future__ import annotations
import asyncio
import logging
from typing import Any
from fastapi import APIRouter
from .public_snapshot import get_snapshot_client
_logger = logging.getLogger("agente_ai.api.public_status")
router = APIRouter(prefix="/api/public", tags=["public"])
_PUBLIC_FIELDS = (
"singleton,service_status,active_sessions,queued_tasks,in_progress_tasks,"
"app_version,updated_at"
)
@router.get("/status")
async def public_status() -> dict[str, Any]:
"""Restituisce esclusivamente lo snapshot deliberatamente pubblico."""
client = get_snapshot_client()
if client is None:
return _degraded_snapshot("database_unavailable")
def operation():
return client.table("public_dashboard_snapshot").select(_PUBLIC_FIELDS).eq("singleton", True).limit(1).execute()
try:
result = await asyncio.to_thread(operation)
except Exception as exc:
_logger.warning("public status snapshot unavailable: %s", type(exc).__name__)
return _degraded_snapshot("snapshot_unavailable")
row = (result.data or [None])[0]
if not row:
return _degraded_snapshot("snapshot_not_initialized")
return {
"service_status": str(row.get("service_status") or "unknown"),
"active_sessions": int(row.get("active_sessions") or 0),
"queued_tasks": int(row.get("queued_tasks") or 0),
"in_progress_tasks": int(row.get("in_progress_tasks") or 0),
"app_version": row.get("app_version"),
"updated_at": row.get("updated_at"),
}
def _degraded_snapshot(reason: str) -> dict[str, Any]:
"""Safe public response while the operational snapshot is unavailable.
The public endpoint is used by lightweight status surfaces. Returning a
deliberate degraded state keeps those surfaces functional without
exposing database errors, internal topology, or operational records.
"""
return {
"service_status": "degraded",
"active_sessions": 0,
"queued_tasks": 0,
"in_progress_tasks": 0,
"app_version": None,
"updated_at": None,
"degraded": True,
"reason": reason,
}
|