Praxis CI
a7f7c4e1cc
TASK-08-01: server/operator/cohort.py — GET /api/operator/cohort practice
volume view (sessions_count, active_learners_count), k-anon display.
TASK-08-02: server/operator/mastery.py — GET /api/operator/mastery
(gate_open_rate, median_mastery_score, rubric_criterion_means).
TASK-08-03: server/operator/failure_patterns.py — GET /api/operator/failure-patterns
(failure_mode:*, branch:* metrics).
TASK-08-04: server/operator/credentials.py — GET /api/operator/credentials
+ POST /api/operator/credentials/{id}/revoke (VC management, D-057).
TASK-08-05: tests/test_operator_endpoints.py — 401/200/503, suppressed
cells value=null, last_updated=max, revoke, no per-learner data (R-DASH-02).
server/operator/_common.py — shared Cell/PathView/ViewResponse models +
all_recent_aggregates query + group_by_path helper.
All endpoints auth-gated via Depends(current_operator) (D-057). No PII
beyond aggregates (D-031, R-DASH-02).
---ci---
project: praxis
phase: 2
milestone: v0.4
status: execute
persona: backend-engineer
task: 08-01..08-05
requirements:
covered: [REQ-DASH-01, REQ-NFR-DASH-01]
---/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"] |