105 lines
3.9 KiB
Python
105 lines
3.9 KiB
Python
"""Cross-service reads for the dashboard — RQ Score (Momentum/QX) + user-service (Day streak).
|
|
|
|
Direct REST, Clerk-mesh-authed. Both degrade to None on any failure (timeout / 404 / service down) —
|
|
a missing KPI is omitted by the UI, never faked, and never crashes the dashboard.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import math
|
|
from collections.abc import Mapping
|
|
|
|
import httpx
|
|
|
|
from app.config import get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _score_value(value: object) -> float | None:
|
|
"""Parse one canonical RQ Score value; malformed values are absent."""
|
|
if isinstance(value, bool) or value is None:
|
|
return None
|
|
try:
|
|
score = float(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
return score if math.isfinite(score) else None
|
|
|
|
|
|
async def fetch_qx(user_uuid: str | None) -> dict | None:
|
|
"""RQ Score for the user → {qx, quotients}. GET /v1/qscore/{uuid}?org_id (qscore-service)."""
|
|
if not user_uuid:
|
|
return None
|
|
s = get_settings()
|
|
try:
|
|
async with httpx.AsyncClient(timeout=4.0) as c:
|
|
r = await c.get(f"{s.QSCORE_BASE_URL}/v1/qscore/{user_uuid}", params={"org_id": s.QSCORE_ORG_ID})
|
|
if r.status_code == 404: # no score computed yet — honest absence
|
|
return None
|
|
r.raise_for_status()
|
|
payload = r.json()
|
|
if not isinstance(payload, Mapping):
|
|
return None
|
|
qx = _score_value(payload.get("rq_score"))
|
|
if qx is None:
|
|
return None
|
|
quotients = payload.get("quotients")
|
|
return {"qx": round(qx, 1), "quotients": quotients if isinstance(quotients, Mapping) else {}}
|
|
except (httpx.HTTPError, ValueError, TypeError) as e:
|
|
logger.warning("fetch_qx failed (non-fatal): %s", e)
|
|
return None
|
|
|
|
|
|
async def fetch_qx_trend(user_uuid: str | None) -> list[float] | None:
|
|
"""RQ Score trend (oldest→newest) ← qscore-service GET /v1/qscore/{uuid}/history. Same source as
|
|
the live QX, so the trend's latest point matches the Momentum value (consistent, not stale).
|
|
None on failure → the UI keeps the Readiness card locked."""
|
|
if not user_uuid:
|
|
return None
|
|
s = get_settings()
|
|
try:
|
|
async with httpx.AsyncClient(timeout=4.0) as c:
|
|
r = await c.get(
|
|
f"{s.QSCORE_BASE_URL}/v1/qscore/{user_uuid}/history",
|
|
params={"org_id": s.QSCORE_ORG_ID, "points": 10},
|
|
)
|
|
if r.status_code == 404:
|
|
return None
|
|
payload = r.json()
|
|
if not isinstance(payload, Mapping):
|
|
return None
|
|
points = payload.get("points")
|
|
if not isinstance(points, list):
|
|
return None
|
|
pts = []
|
|
for point in points:
|
|
if not isinstance(point, Mapping):
|
|
continue
|
|
score = _score_value(point.get("rq_score"))
|
|
if score is not None:
|
|
pts.append(round(score, 1))
|
|
return pts if len(pts) >= 2 else None # need ≥2 points for a trend
|
|
except (httpx.HTTPError, ValueError, TypeError) as e:
|
|
logger.warning("fetch_qx_trend failed (non-fatal): %s", e)
|
|
return None
|
|
|
|
|
|
async def fetch_streak(clerk_id: str | None) -> int | None:
|
|
"""Day streak ← user-service metadata.current_streak. GET /api/state/{clerk_id} (Bearer A2A)."""
|
|
if not clerk_id:
|
|
return None
|
|
s = get_settings()
|
|
try:
|
|
async with httpx.AsyncClient(timeout=4.0) as c:
|
|
r = await c.get(
|
|
f"{s.USER_SERVICE_BASE_URL}/api/state/{clerk_id}",
|
|
headers={"Authorization": f"Bearer {s.A2A_OUTBOUND_KEY}"},
|
|
)
|
|
r.raise_for_status()
|
|
streak = (r.json().get("metadata") or {}).get("current_streak")
|
|
return int(streak) if streak is not None else None
|
|
except (httpx.HTTPError, ValueError, TypeError) as e:
|
|
logger.warning("fetch_streak failed (non-fatal): %s", e)
|
|
return None
|