Praxis CI
ec6fcc64bc
Phase 2 complete — Cohort Dashboard + Aggregation: - Cohort aggregation pipeline (k-anon ≥10 write-time suppression, async hook, nightly 03:00 CT reconcile) - 4 auth-gated operator API endpoints (cohort, mastery, failure-patterns, credentials) - React cohort dashboard (BrowserRouter, login, 3 views, inline SVG sparklines, auth gate) - SPA fallback via SpaStaticFiles subclass (G-041 — NOT catch-all route) - G-038 differencing-attack test (unit + API e2e) - 317 pytest pass, 36 skip, 0 fail; 17/17 vitest pass; npm build + typecheck clean ---ci--- project: praxis phase: 2 milestone: v0.4 status: complete requirements: covered: [REQ-DASH-01, REQ-NFR-DASH-01, REQ-NFR-DASH-02, REQ-MT-02] partial: [] ---/ci---
44 lines
1.4 KiB
Python
44 lines
1.4 KiB
Python
"""GET /api/operator/failure-patterns — failure patterns view (TASK-08-03, D-053).
|
|
|
|
Auth-gated. Returns failure pattern metrics: failure_mode frequency (cells
|
|
with metric prefix `failure_mode:`) + branch outcome distribution (cells
|
|
with metric prefix `branch:`). Weak-spot rubric criteria (mean < 3.0) are
|
|
highlighted by the frontend. All k-anonymized.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as _dt
|
|
|
|
from fastapi import APIRouter, Depends, Request
|
|
|
|
from server.auth.dependencies import current_operator
|
|
from server.auth.models import Operator
|
|
from server.operator._common import (
|
|
ViewResponse,
|
|
all_recent_aggregates,
|
|
group_by_path,
|
|
require_pg_store,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/operator", tags=["operator-failure-patterns"])
|
|
|
|
|
|
def _is_failure_metric(metric: str) -> bool:
|
|
return metric.startswith("failure_mode:") or metric.startswith("branch:")
|
|
|
|
|
|
@router.get("/failure-patterns", response_model=ViewResponse)
|
|
async def failure_patterns_view(
|
|
request: Request,
|
|
op: Operator = Depends(current_operator),
|
|
) -> ViewResponse:
|
|
pg_store = await require_pg_store(request)
|
|
since = _dt.date.today() - _dt.timedelta(days=30)
|
|
rows = await all_recent_aggregates(pg_store, since)
|
|
failure_rows = [r for r in rows if _is_failure_metric(r.get("metric", ""))]
|
|
views, last_updated = group_by_path(failure_rows)
|
|
return ViewResponse(views=views, last_updated=last_updated)
|
|
|
|
|
|
__all__ = ["router"] |