257 lines
9.5 KiB
Python
257 lines
9.5 KiB
Python
"""Running class markbook / gradebook API (/api/markbook).
|
||
|
||
A markbook is a term-long teacher-editable ledger: roster rows × arbitrary assessment
|
||
columns. It deliberately does not depend on exam-marker batches. All user-facing reads and
|
||
writes use the as-user Supabase client so class/markbook RLS gates are enforced.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import csv
|
||
import io
|
||
import os
|
||
from datetime import date as Date
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException
|
||
from fastapi.responses import Response
|
||
from pydantic import BaseModel, Field
|
||
|
||
from modules.logger_tool import initialise_logger
|
||
from routers.exam.dependencies import ExamContext, get_exam_context, resolve_student_names
|
||
|
||
logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), "default", True)
|
||
router = APIRouter()
|
||
|
||
|
||
class CreateAssessmentRequest(BaseModel):
|
||
title: str = Field(..., min_length=1, max_length=160)
|
||
date: Optional[Date] = None
|
||
max_marks: float = Field(default=100, gt=0)
|
||
|
||
|
||
class MarkUpsertRequest(BaseModel):
|
||
mark: Optional[float] = Field(default=None, ge=0)
|
||
|
||
|
||
def _rows(result: Any) -> List[Dict[str, Any]]:
|
||
data = getattr(result, "data", None)
|
||
if not data:
|
||
return []
|
||
return data if isinstance(data, list) else [data]
|
||
|
||
|
||
def _first(result: Any) -> Optional[Dict[str, Any]]:
|
||
rows = _rows(result)
|
||
return rows[0] if rows else None
|
||
|
||
|
||
def _round(value: Optional[float]) -> Optional[float]:
|
||
return None if value is None else round(float(value), 1)
|
||
|
||
|
||
def _fetch_class_or_404(ctx: ExamContext, class_id: str) -> Dict[str, Any]:
|
||
row = _first(ctx.supabase.table("classes").select("id, name, institute_id").eq("id", class_id).limit(1).execute())
|
||
if not row:
|
||
raise HTTPException(status_code=404, detail="Class not found")
|
||
return row
|
||
|
||
|
||
def _active_roster(ctx: ExamContext, class_id: str) -> List[Dict[str, Any]]:
|
||
roster = _rows(
|
||
ctx.supabase.table("class_students")
|
||
.select("student_id, status, enrolled_at")
|
||
.eq("class_id", class_id)
|
||
.eq("status", "active")
|
||
.execute()
|
||
)
|
||
names = resolve_student_names([r["student_id"] for r in roster if r.get("student_id")])
|
||
return [
|
||
{
|
||
"student_id": r["student_id"],
|
||
"student_name": names.get(r["student_id"]) or r["student_id"],
|
||
"status": r.get("status"),
|
||
"enrolled_at": r.get("enrolled_at"),
|
||
}
|
||
for r in roster
|
||
if r.get("student_id")
|
||
]
|
||
|
||
|
||
def _assessments(ctx: ExamContext, class_id: str) -> List[Dict[str, Any]]:
|
||
return _rows(
|
||
ctx.supabase.table("class_assessments")
|
||
.select("id, class_id, tenant_id, title, date, max_marks, created_at, updated_at")
|
||
.eq("class_id", class_id)
|
||
.order("date")
|
||
.order("created_at")
|
||
.execute()
|
||
)
|
||
|
||
|
||
def _marks(ctx: ExamContext, assessment_ids: List[str]) -> List[Dict[str, Any]]:
|
||
if not assessment_ids:
|
||
return []
|
||
return _rows(
|
||
ctx.supabase.table("assessment_marks")
|
||
.select("assessment_id, student_id, mark, updated_at, updated_by")
|
||
.in_("assessment_id", assessment_ids)
|
||
.execute()
|
||
)
|
||
|
||
|
||
def _assemble_grid(ctx: ExamContext, class_id: str) -> Dict[str, Any]:
|
||
cls = _fetch_class_or_404(ctx, class_id)
|
||
roster = _active_roster(ctx, class_id)
|
||
assessments = _assessments(ctx, class_id)
|
||
assessment_ids = [a["id"] for a in assessments]
|
||
marks = _marks(ctx, assessment_ids)
|
||
|
||
marks_by_student: Dict[str, Dict[str, Optional[float]]] = {r["student_id"]: {} for r in roster}
|
||
for m in marks:
|
||
sid = m.get("student_id")
|
||
aid = m.get("assessment_id")
|
||
if isinstance(sid, str) and isinstance(aid, str) and sid in marks_by_student and aid in assessment_ids:
|
||
marks_by_student[sid][aid] = m.get("mark")
|
||
|
||
max_total = sum(float(a.get("max_marks") or 0) for a in assessments)
|
||
students = []
|
||
all_entered: List[float] = []
|
||
for idx, student in enumerate(roster, start=1):
|
||
entered = [
|
||
float(v)
|
||
for v in (marks_by_student.get(student["student_id"], {}).get(aid) for aid in assessment_ids)
|
||
if v is not None
|
||
]
|
||
total = sum(entered) if entered else None
|
||
students.append(
|
||
{
|
||
**student,
|
||
"row_number": idx,
|
||
"marks": {aid: marks_by_student.get(student["student_id"], {}).get(aid) for aid in assessment_ids},
|
||
"total": total,
|
||
"percentage": _round((total / max_total) * 100) if total is not None and max_total > 0 else None,
|
||
}
|
||
)
|
||
all_entered.extend(entered)
|
||
|
||
assessment_summaries = []
|
||
for a in assessments:
|
||
vals = []
|
||
for s in roster:
|
||
maybe_mark = marks_by_student.get(s["student_id"], {}).get(a["id"])
|
||
if maybe_mark is not None:
|
||
vals.append(float(maybe_mark))
|
||
max_marks = float(a.get("max_marks") or 0)
|
||
assessment_summaries.append(
|
||
{
|
||
"assessment_id": a["id"],
|
||
"entered_count": len(vals),
|
||
"average_mark": _round(sum(vals) / len(vals)) if vals else None,
|
||
"average_percentage": _round((sum(vals) / len(vals) / max_marks) * 100) if vals and max_marks > 0 else None,
|
||
}
|
||
)
|
||
|
||
return {
|
||
"class": cls,
|
||
"students": students,
|
||
"assessments": assessments,
|
||
"assessment_summaries": assessment_summaries,
|
||
"summary": {
|
||
"student_count": len(roster),
|
||
"assessment_count": len(assessments),
|
||
"entered_mark_count": len(all_entered),
|
||
"class_average_mark": _round(sum(all_entered) / len(all_entered)) if all_entered else None,
|
||
},
|
||
}
|
||
|
||
|
||
@router.get("/classes/{class_id}/assessments")
|
||
async def list_assessments(class_id: str, ctx: ExamContext = Depends(get_exam_context)) -> Dict[str, Any]:
|
||
_fetch_class_or_404(ctx, class_id)
|
||
return {"assessments": _assessments(ctx, class_id)}
|
||
|
||
|
||
@router.post("/classes/{class_id}/assessments")
|
||
async def create_assessment(class_id: str, body: CreateAssessmentRequest, ctx: ExamContext = Depends(get_exam_context)) -> Dict[str, Any]:
|
||
cls = _fetch_class_or_404(ctx, class_id)
|
||
row = {
|
||
"class_id": class_id,
|
||
"tenant_id": cls["institute_id"],
|
||
"title": body.title.strip(),
|
||
"date": body.date.isoformat() if body.date else None,
|
||
"max_marks": body.max_marks,
|
||
}
|
||
created = _first(ctx.supabase.table("class_assessments").insert(row).execute())
|
||
if not created:
|
||
raise HTTPException(status_code=500, detail="Failed to create assessment")
|
||
logger.info(f"Markbook assessment {created.get('id')} created for class {class_id} by {ctx.user_id}")
|
||
return created
|
||
|
||
|
||
@router.get("/classes/{class_id}/grid")
|
||
async def get_grid(class_id: str, ctx: ExamContext = Depends(get_exam_context)) -> Dict[str, Any]:
|
||
return _assemble_grid(ctx, class_id)
|
||
|
||
|
||
@router.put("/classes/{class_id}/assessments/{assessment_id}/marks/{student_id}")
|
||
async def upsert_mark(
|
||
class_id: str,
|
||
assessment_id: str,
|
||
student_id: str,
|
||
body: MarkUpsertRequest,
|
||
ctx: ExamContext = Depends(get_exam_context),
|
||
) -> Dict[str, Any]:
|
||
cls = _fetch_class_or_404(ctx, class_id)
|
||
assessment = _first(
|
||
ctx.supabase.table("class_assessments")
|
||
.select("id, class_id, tenant_id, max_marks")
|
||
.eq("id", assessment_id)
|
||
.eq("class_id", class_id)
|
||
.limit(1)
|
||
.execute()
|
||
)
|
||
if not assessment:
|
||
raise HTTPException(status_code=404, detail="Assessment not found")
|
||
roster_row = _first(
|
||
ctx.supabase.table("class_students")
|
||
.select("student_id")
|
||
.eq("class_id", class_id)
|
||
.eq("student_id", student_id)
|
||
.eq("status", "active")
|
||
.limit(1)
|
||
.execute()
|
||
)
|
||
if not roster_row:
|
||
raise HTTPException(status_code=404, detail="Student is not active in this class")
|
||
if body.mark is not None and body.mark > float(assessment.get("max_marks") or 0):
|
||
raise HTTPException(status_code=422, detail="mark exceeds assessment max_marks")
|
||
|
||
row = {
|
||
"assessment_id": assessment_id,
|
||
"student_id": student_id,
|
||
"tenant_id": assessment.get("tenant_id") or cls["institute_id"],
|
||
"mark": body.mark,
|
||
"updated_by": ctx.user_id,
|
||
}
|
||
upserted = _first(ctx.supabase.table("assessment_marks").upsert(row, on_conflict="assessment_id,student_id").execute())
|
||
if not upserted:
|
||
raise HTTPException(status_code=500, detail="Failed to upsert mark")
|
||
return {"status": "ok", "mark": upserted}
|
||
|
||
|
||
@router.get("/classes/{class_id}/csv")
|
||
async def export_csv(class_id: str, ctx: ExamContext = Depends(get_exam_context)) -> Response:
|
||
data = _assemble_grid(ctx, class_id)
|
||
assessments = data["assessments"]
|
||
buf = io.StringIO()
|
||
writer = csv.writer(buf)
|
||
writer.writerow(["row", "student_name", "student_id"] + [a["title"] for a in assessments] + ["total", "percentage"])
|
||
for student in data["students"]:
|
||
writer.writerow(
|
||
[student["row_number"], student.get("student_name") or "", student.get("student_id") or ""]
|
||
+ ["" if student["marks"].get(a["id"]) is None else student["marks"].get(a["id"]) for a in assessments]
|
||
+ ["" if student["total"] is None else student["total"], "" if student["percentage"] is None else student["percentage"]]
|
||
)
|
||
filename = f"markbook-{class_id}.csv"
|
||
return Response(content=buf.getvalue(), media_type="text/csv", headers={"Content-Disposition": f'attachment; filename="{filename}"'})
|