Conservazione in Sidebar
This commit is contained in:
@@ -24,6 +24,10 @@ from fastapi import APIRouter, HTTPException, Query, status
|
||||
from app.config import get_settings as get_app_settings
|
||||
from app.dependencies import AdminUser, DB
|
||||
from app.schemas.tenant_settings import (
|
||||
ConservationStats,
|
||||
ConservationStatsResponse,
|
||||
ConservationTenantBreakdown,
|
||||
ConservationTriggerResult,
|
||||
ConservatoreTestResult,
|
||||
IndexingJobStatus,
|
||||
IndexingStats,
|
||||
@@ -418,6 +422,177 @@ async def get_rescan_status(
|
||||
return IndexingJobStatus(**status_data)
|
||||
|
||||
|
||||
# ─── Conservazione sostitutiva – statistiche e trigger ───────────────────────
|
||||
|
||||
def _build_conservation_stats(
|
||||
total: int,
|
||||
conserved: int,
|
||||
pending: int,
|
||||
last_conserved_at,
|
||||
) -> ConservationStats:
|
||||
not_queued = max(0, total - conserved - pending)
|
||||
coverage_pct = round((conserved / total * 100), 1) if total > 0 else 0.0
|
||||
last_dt = last_conserved_at.isoformat() if last_conserved_at else None
|
||||
return ConservationStats(
|
||||
total_messages=total,
|
||||
conserved=conserved,
|
||||
pending=pending,
|
||||
not_queued=not_queued,
|
||||
coverage_pct=coverage_pct,
|
||||
last_conserved_at=last_dt,
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/conservation/stats",
|
||||
response_model=ConservationStatsResponse,
|
||||
summary="Statistiche conservazione sostitutiva",
|
||||
description=(
|
||||
"Restituisce il numero di messaggi totali, conservati, in coda e non accodati. "
|
||||
"admin: statistiche del proprio tenant. "
|
||||
"super_admin con ?tenant_id=<uuid>: statistiche del tenant specificato. "
|
||||
"super_admin senza tenant_id: statistiche aggregate su tutti i tenant + breakdown per-tenant."
|
||||
),
|
||||
)
|
||||
async def get_conservation_stats(
|
||||
current_user: AdminUser,
|
||||
db: DB,
|
||||
tenant_id: Optional[uuid.UUID] = Query(
|
||||
default=None,
|
||||
description="(solo super_admin) UUID del tenant su cui operare; assente = tutti i tenant",
|
||||
),
|
||||
) -> ConservationStatsResponse:
|
||||
from sqlalchemy import func, select
|
||||
from app.models.message import Message
|
||||
from app.models.tenant import Tenant
|
||||
|
||||
# Se non e' super_admin, oppure e' super_admin con tenant_id specificato -> singolo tenant
|
||||
if current_user.role != "super_admin" or tenant_id is not None:
|
||||
target_tenant_id = _resolve_tenant_id(current_user, tenant_id)
|
||||
|
||||
row = await db.execute(
|
||||
select(
|
||||
func.count().label("total"),
|
||||
func.count().filter(Message.is_conserved == True).label("conserved"), # noqa: E712
|
||||
func.count().filter(
|
||||
Message.is_pending_conservation == True, # noqa: E712
|
||||
Message.is_conserved == False, # noqa: E712
|
||||
).label("pending"),
|
||||
func.max(Message.conserved_at).label("last_conserved_at"),
|
||||
).where(Message.tenant_id == target_tenant_id)
|
||||
)
|
||||
r = row.one()
|
||||
stats = _build_conservation_stats(
|
||||
total=r.total or 0,
|
||||
conserved=r.conserved or 0,
|
||||
pending=r.pending or 0,
|
||||
last_conserved_at=r.last_conserved_at,
|
||||
)
|
||||
return ConservationStatsResponse(stats=stats, per_tenant=None)
|
||||
|
||||
# super_admin senza tenant_id: aggregato + per-tenant
|
||||
rows = await db.execute(
|
||||
select(
|
||||
Message.tenant_id,
|
||||
func.count().label("total"),
|
||||
func.count().filter(Message.is_conserved == True).label("conserved"), # noqa: E712
|
||||
func.count().filter(
|
||||
Message.is_pending_conservation == True, # noqa: E712
|
||||
Message.is_conserved == False, # noqa: E712
|
||||
).label("pending"),
|
||||
func.max(Message.conserved_at).label("last_conserved_at"),
|
||||
).group_by(Message.tenant_id)
|
||||
)
|
||||
tenant_rows = rows.all()
|
||||
|
||||
# Carica nomi tenant
|
||||
all_tenant_ids = [r.tenant_id for r in tenant_rows]
|
||||
tenant_names: dict[uuid.UUID, tuple[str, str]] = {}
|
||||
if all_tenant_ids:
|
||||
t_rows = await db.execute(
|
||||
select(Tenant.id, Tenant.name, Tenant.slug).where(Tenant.id.in_(all_tenant_ids))
|
||||
)
|
||||
for t in t_rows.all():
|
||||
tenant_names[t.id] = (t.name, t.slug)
|
||||
|
||||
per_tenant: list[ConservationTenantBreakdown] = []
|
||||
agg_total = agg_conserved = agg_pending = 0
|
||||
agg_last: object = None
|
||||
|
||||
for r in tenant_rows:
|
||||
agg_total += r.total or 0
|
||||
agg_conserved += r.conserved or 0
|
||||
agg_pending += r.pending or 0
|
||||
if r.last_conserved_at and (agg_last is None or r.last_conserved_at > agg_last):
|
||||
agg_last = r.last_conserved_at
|
||||
|
||||
name, slug = tenant_names.get(r.tenant_id, ("(sconosciuto)", ""))
|
||||
per_tenant.append(ConservationTenantBreakdown(
|
||||
tenant_id=str(r.tenant_id),
|
||||
tenant_name=name,
|
||||
tenant_slug=slug,
|
||||
stats=_build_conservation_stats(
|
||||
total=r.total or 0,
|
||||
conserved=r.conserved or 0,
|
||||
pending=r.pending or 0,
|
||||
last_conserved_at=r.last_conserved_at,
|
||||
),
|
||||
))
|
||||
|
||||
aggregate = _build_conservation_stats(
|
||||
total=agg_total,
|
||||
conserved=agg_conserved,
|
||||
pending=agg_pending,
|
||||
last_conserved_at=agg_last,
|
||||
)
|
||||
return ConservationStatsResponse(stats=aggregate, per_tenant=per_tenant)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/conservation/trigger",
|
||||
response_model=ConservationTriggerResult,
|
||||
status_code=status.HTTP_202_ACCEPTED,
|
||||
summary="Avvia manualmente il job di conservazione sostitutiva",
|
||||
description=(
|
||||
"Accoda immediatamente il job run_conservation nel worker arq. "
|
||||
"Il job elabora tutti i tenant con messaggi pendenti di conservazione. "
|
||||
"Utile per eseguire il ciclo di conservazione senza attendere il cron giornaliero. "
|
||||
"Solo admin e super_admin possono avviarlo."
|
||||
),
|
||||
)
|
||||
async def trigger_conservation(
|
||||
current_user: AdminUser,
|
||||
db: DB,
|
||||
) -> ConservationTriggerResult:
|
||||
app_settings = get_app_settings()
|
||||
try:
|
||||
import urllib.parse
|
||||
from arq.connections import RedisSettings, create_pool
|
||||
|
||||
parsed = urllib.parse.urlparse(app_settings.redis_url)
|
||||
redis_settings = RedisSettings(
|
||||
host=parsed.hostname or "localhost",
|
||||
port=parsed.port or 6379,
|
||||
database=int(parsed.path.lstrip("/") or "0"),
|
||||
password=parsed.password or None,
|
||||
)
|
||||
redis = await create_pool(redis_settings)
|
||||
job = await redis.enqueue_job("run_conservation")
|
||||
await redis.aclose()
|
||||
|
||||
job_id = job.job_id if job else None
|
||||
return ConservationTriggerResult(
|
||||
queued=True,
|
||||
message="Job di conservazione accodato con successo. Verra' eseguito dal worker entro pochi secondi.",
|
||||
job_id=job_id,
|
||||
)
|
||||
except Exception as exc:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail=f"Impossibile accodare il job: {exc}",
|
||||
)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/indexing/rescan",
|
||||
response_model=IndexingJobStatus,
|
||||
|
||||
Reference in New Issue
Block a user