622 lines
19 KiB
Python
622 lines
19 KiB
Python
"""
|
|
Classes Router — Supabase-backed CRUD for the `classes` table.
|
|
Institute members can read their institute's classes.
|
|
School admins and teachers can create classes.
|
|
School admins manage teacher/student assignments; teachers can leave/enroll.
|
|
"""
|
|
import os
|
|
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
from modules.logger_tool import initialise_logger
|
|
from modules.auth.supabase_bearer import SupabaseBearer
|
|
from modules.database.supabase.utils.client import SupabaseServiceRoleClient
|
|
|
|
logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), 'default', True)
|
|
router = APIRouter()
|
|
|
|
|
|
def _sb() -> SupabaseServiceRoleClient:
|
|
return SupabaseServiceRoleClient()
|
|
|
|
|
|
def _resolve_institute_id(user_id: str) -> Optional[str]:
|
|
"""Return the Supabase institute UUID for this user via profiles.school_id."""
|
|
try:
|
|
sb = _sb()
|
|
p = sb.supabase.table("profiles").select("school_id").eq("id", user_id).single().execute()
|
|
return str((p.data or {}).get("school_id") or "")
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _require_institute(user_id: str) -> str:
|
|
"""Return institute_id or raise 400."""
|
|
institute_id = _resolve_institute_id(user_id)
|
|
if not institute_id:
|
|
raise HTTPException(status_code=400, detail="User is not linked to a school")
|
|
return institute_id
|
|
|
|
|
|
def _is_school_admin(user_id: str, institute_id: str) -> bool:
|
|
try:
|
|
sb = _sb()
|
|
r = (
|
|
sb.supabase.table("institute_memberships")
|
|
.select("role")
|
|
.eq("profile_id", user_id)
|
|
.eq("institute_id", institute_id)
|
|
.in_("role", ["school_admin", "department_head"])
|
|
.limit(1)
|
|
.execute()
|
|
)
|
|
return bool(r.data)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
# ─── Request models ───────────────────────────────────────────────────────────
|
|
|
|
class CreateClassRequest(BaseModel):
|
|
name: str
|
|
class_code: Optional[str] = None
|
|
subject: Optional[str] = None
|
|
key_stage: Optional[str] = None
|
|
year_group: Optional[str] = None
|
|
academic_year: Optional[str] = None
|
|
description: Optional[str] = None
|
|
|
|
|
|
class UpdateClassRequest(BaseModel):
|
|
name: Optional[str] = None
|
|
class_code: Optional[str] = None
|
|
subject: Optional[str] = None
|
|
key_stage: Optional[str] = None
|
|
year_group: Optional[str] = None
|
|
academic_year: Optional[str] = None
|
|
description: Optional[str] = None
|
|
is_active: Optional[bool] = None
|
|
|
|
|
|
class AddTeacherRequest(BaseModel):
|
|
teacher_id: str
|
|
is_primary: bool = False
|
|
|
|
|
|
class AddStudentRequest(BaseModel):
|
|
student_id: str
|
|
|
|
|
|
# ─── Endpoints ────────────────────────────────────────────────────────────────
|
|
|
|
@router.get("")
|
|
async def list_classes(
|
|
subject: Optional[str] = None,
|
|
school_year: Optional[str] = None,
|
|
academic_year: Optional[str] = None,
|
|
key_stage: Optional[str] = None,
|
|
year_group: Optional[str] = None,
|
|
search: Optional[str] = None,
|
|
active_only: bool = True,
|
|
skip: int = 0,
|
|
limit: int = 50,
|
|
credentials: dict = Depends(SupabaseBearer()),
|
|
) -> Dict[str, Any]:
|
|
user_id = credentials.get("sub", "")
|
|
institute_id = _require_institute(user_id)
|
|
sb = _sb()
|
|
|
|
q = sb.supabase.table("classes").select("*", count="exact").eq("institute_id", institute_id)
|
|
if active_only:
|
|
q = q.eq("is_active", True)
|
|
if subject:
|
|
q = q.eq("subject", subject)
|
|
# support both param names
|
|
yr = academic_year or school_year
|
|
if yr:
|
|
q = q.eq("academic_year", yr)
|
|
if key_stage:
|
|
q = q.eq("key_stage", key_stage)
|
|
if year_group:
|
|
q = q.eq("year_group", year_group)
|
|
if search:
|
|
q = q.ilike("name", f"%{search}%")
|
|
|
|
q = q.order("name").range(skip, skip + limit - 1)
|
|
res = q.execute()
|
|
|
|
classes = res.data or []
|
|
total = res.count or 0
|
|
|
|
# Attach student/teacher counts
|
|
class_ids = [c["id"] for c in classes]
|
|
teacher_counts: Dict[str, int] = {}
|
|
student_counts: Dict[str, int] = {}
|
|
if class_ids:
|
|
try:
|
|
tc = (
|
|
sb.supabase.table("class_teachers")
|
|
.select("class_id", count="exact")
|
|
.in_("class_id", class_ids)
|
|
.execute()
|
|
)
|
|
for row in (tc.data or []):
|
|
teacher_counts[row["class_id"]] = teacher_counts.get(row["class_id"], 0) + 1
|
|
|
|
sc = (
|
|
sb.supabase.table("class_students")
|
|
.select("class_id")
|
|
.in_("class_id", class_ids)
|
|
.eq("status", "active")
|
|
.execute()
|
|
)
|
|
for row in (sc.data or []):
|
|
student_counts[row["class_id"]] = student_counts.get(row["class_id"], 0) + 1
|
|
except Exception:
|
|
pass
|
|
|
|
enriched = [
|
|
{**c, "teacher_count": teacher_counts.get(c["id"], 0), "student_count": student_counts.get(c["id"], 0)}
|
|
for c in classes
|
|
]
|
|
return {"classes": enriched, "total": total}
|
|
|
|
|
|
@router.get("/me/teacher")
|
|
async def my_teaching_classes(
|
|
credentials: dict = Depends(SupabaseBearer()),
|
|
) -> Dict[str, Any]:
|
|
user_id = credentials.get("sub", "")
|
|
institute_id = _require_institute(user_id)
|
|
sb = _sb()
|
|
|
|
assigned = (
|
|
sb.supabase.table("class_teachers")
|
|
.select("class_id, is_primary")
|
|
.eq("teacher_id", user_id)
|
|
.execute()
|
|
.data or []
|
|
)
|
|
if not assigned:
|
|
return {"classes": []}
|
|
|
|
class_ids = [a["class_id"] for a in assigned]
|
|
is_primary_map = {a["class_id"]: a["is_primary"] for a in assigned}
|
|
|
|
res = (
|
|
sb.supabase.table("classes")
|
|
.select("*")
|
|
.in_("id", class_ids)
|
|
.eq("institute_id", institute_id)
|
|
.eq("is_active", True)
|
|
.order("name")
|
|
.execute()
|
|
.data or []
|
|
)
|
|
enriched = [{**c, "is_primary_teacher": is_primary_map.get(c["id"], False)} for c in res]
|
|
return {"classes": enriched}
|
|
|
|
|
|
@router.get("/me/student")
|
|
async def my_student_classes(
|
|
credentials: dict = Depends(SupabaseBearer()),
|
|
) -> Dict[str, Any]:
|
|
user_id = credentials.get("sub", "")
|
|
institute_id = _require_institute(user_id)
|
|
sb = _sb()
|
|
|
|
enrolled = (
|
|
sb.supabase.table("class_students")
|
|
.select("class_id")
|
|
.eq("student_id", user_id)
|
|
.eq("status", "active")
|
|
.execute()
|
|
.data or []
|
|
)
|
|
if not enrolled:
|
|
return {"classes": []}
|
|
|
|
class_ids = [e["class_id"] for e in enrolled]
|
|
res = (
|
|
sb.supabase.table("classes")
|
|
.select("*")
|
|
.in_("id", class_ids)
|
|
.eq("institute_id", institute_id)
|
|
.eq("is_active", True)
|
|
.order("name")
|
|
.execute()
|
|
.data or []
|
|
)
|
|
return {"classes": res}
|
|
|
|
@router.get("/school/students")
|
|
async def list_school_students(
|
|
credentials: dict = Depends(SupabaseBearer()),
|
|
) -> Dict[str, Any]:
|
|
"""List all students in the caller's school. Used by admin to add students to a class."""
|
|
user_id = credentials.get("sub", "")
|
|
institute_id = _require_institute(user_id)
|
|
sb = _sb()
|
|
members = (
|
|
sb.supabase.table("institute_memberships")
|
|
.select("profile_id")
|
|
.eq("institute_id", institute_id)
|
|
.eq("role", "student")
|
|
.execute()
|
|
.data or []
|
|
)
|
|
student_ids = [m["profile_id"] for m in members]
|
|
if not student_ids:
|
|
return {"students": []}
|
|
profiles = (
|
|
sb.supabase.table("profiles")
|
|
.select("id, full_name, display_name, email, user_type")
|
|
.in_("id", student_ids)
|
|
.order("full_name")
|
|
.execute()
|
|
.data or []
|
|
)
|
|
return {"students": profiles}
|
|
|
|
|
|
|
|
@router.get("/{class_id}")
|
|
async def get_class(
|
|
class_id: str,
|
|
credentials: dict = Depends(SupabaseBearer()),
|
|
) -> Dict[str, Any]:
|
|
user_id = credentials.get("sub", "")
|
|
institute_id = _require_institute(user_id)
|
|
sb = _sb()
|
|
|
|
cls_res = (
|
|
sb.supabase.table("classes")
|
|
.select("*")
|
|
.eq("id", class_id)
|
|
.eq("institute_id", institute_id)
|
|
.single()
|
|
.execute()
|
|
)
|
|
if not cls_res.data:
|
|
raise HTTPException(status_code=404, detail="Class not found")
|
|
|
|
teachers = (
|
|
sb.supabase.table("class_teachers")
|
|
.select("teacher_id, is_primary, can_edit, assigned_at")
|
|
.eq("class_id", class_id)
|
|
.execute()
|
|
.data or []
|
|
)
|
|
students = (
|
|
sb.supabase.table("class_students")
|
|
.select("student_id, status, enrolled_at")
|
|
.eq("class_id", class_id)
|
|
.eq("status", "active")
|
|
.execute()
|
|
.data or []
|
|
)
|
|
|
|
# Enrich with profile data
|
|
all_ids = [t["teacher_id"] for t in teachers] + [s["student_id"] for s in students]
|
|
profile_map: Dict[str, Dict] = {}
|
|
if all_ids:
|
|
profiles = (
|
|
sb.supabase.table("profiles")
|
|
.select("id, full_name, display_name, email, user_type")
|
|
.in_("id", list(set(all_ids)))
|
|
.execute()
|
|
.data or []
|
|
)
|
|
profile_map = {p["id"]: p for p in profiles}
|
|
|
|
for t in teachers:
|
|
t["profile"] = profile_map.get(t["teacher_id"], {})
|
|
for s in students:
|
|
s["profile"] = profile_map.get(s["student_id"], {})
|
|
|
|
# Enrollment requests (pending)
|
|
reqs = (
|
|
sb.supabase.table("enrollment_requests")
|
|
.select("id, student_id, status, created_at")
|
|
.eq("class_id", class_id)
|
|
.eq("status", "pending")
|
|
.execute()
|
|
.data or []
|
|
)
|
|
req_student_ids = [r["student_id"] for r in reqs if r.get("student_id")]
|
|
req_profiles: Dict[str, Dict] = {}
|
|
if req_student_ids:
|
|
rp = (
|
|
sb.supabase.table("profiles")
|
|
.select("id, full_name, email")
|
|
.in_("id", req_student_ids)
|
|
.execute()
|
|
.data or []
|
|
)
|
|
req_profiles = {p["id"]: p for p in rp}
|
|
for r in reqs:
|
|
r["profile"] = req_profiles.get(r.get("student_id", ""), {})
|
|
|
|
return {
|
|
**cls_res.data,
|
|
"teachers": teachers,
|
|
"students": students,
|
|
"enrollment_requests": reqs,
|
|
"student_count": len(students),
|
|
}
|
|
|
|
|
|
@router.post("")
|
|
async def create_class(
|
|
body: CreateClassRequest,
|
|
credentials: dict = Depends(SupabaseBearer()),
|
|
) -> Dict[str, Any]:
|
|
user_id = credentials.get("sub", "")
|
|
institute_id = _require_institute(user_id)
|
|
sb = _sb()
|
|
|
|
row = {
|
|
"institute_id": institute_id,
|
|
"name": body.name,
|
|
"created_by": user_id,
|
|
}
|
|
for field in ("class_code", "subject", "key_stage", "year_group", "academic_year", "description"):
|
|
val = getattr(body, field)
|
|
if val is not None:
|
|
row[field] = val
|
|
|
|
res = sb.supabase.table("classes").insert(row).execute()
|
|
new_class = (res.data or [{}])[0]
|
|
class_id = new_class.get("id")
|
|
|
|
# Auto-assign creator as primary teacher if they have teacher role
|
|
if class_id:
|
|
try:
|
|
mem = (
|
|
sb.supabase.table("institute_memberships")
|
|
.select("role")
|
|
.eq("profile_id", user_id)
|
|
.eq("institute_id", institute_id)
|
|
.single()
|
|
.execute()
|
|
)
|
|
if (mem.data or {}).get("role") in ("teacher", "department_head"):
|
|
sb.supabase.table("class_teachers").insert({
|
|
"class_id": class_id,
|
|
"teacher_id": user_id,
|
|
"is_primary": True,
|
|
"assigned_by": user_id,
|
|
}).execute()
|
|
except Exception:
|
|
pass
|
|
|
|
logger.info(f"Class created: {class_id} '{body.name}' by {user_id}")
|
|
return new_class
|
|
|
|
|
|
@router.patch("/{class_id}")
|
|
async def update_class(
|
|
class_id: str,
|
|
body: UpdateClassRequest,
|
|
credentials: dict = Depends(SupabaseBearer()),
|
|
) -> Dict[str, Any]:
|
|
user_id = credentials.get("sub", "")
|
|
institute_id = _require_institute(user_id)
|
|
sb = _sb()
|
|
|
|
# Must be school admin OR primary teacher of this class
|
|
is_admin = _is_school_admin(user_id, institute_id)
|
|
if not is_admin:
|
|
ct = (
|
|
sb.supabase.table("class_teachers")
|
|
.select("is_primary")
|
|
.eq("class_id", class_id)
|
|
.eq("teacher_id", user_id)
|
|
.single()
|
|
.execute()
|
|
)
|
|
if not (ct.data or {}).get("is_primary"):
|
|
raise HTTPException(status_code=403, detail="Only school admins or the primary teacher can update this class")
|
|
|
|
updates = {k: v for k, v in body.dict().items() if v is not None}
|
|
if not updates:
|
|
raise HTTPException(status_code=400, detail="No fields to update")
|
|
updates["updated_at"] = datetime.utcnow().isoformat()
|
|
|
|
res = (
|
|
sb.supabase.table("classes")
|
|
.update(updates)
|
|
.eq("id", class_id)
|
|
.eq("institute_id", institute_id)
|
|
.execute()
|
|
)
|
|
return (res.data or [{}])[0]
|
|
|
|
|
|
@router.delete("/{class_id}")
|
|
async def delete_class(
|
|
class_id: str,
|
|
credentials: dict = Depends(SupabaseBearer()),
|
|
) -> Dict[str, Any]:
|
|
user_id = credentials.get("sub", "")
|
|
institute_id = _require_institute(user_id)
|
|
|
|
if not _is_school_admin(user_id, institute_id):
|
|
raise HTTPException(status_code=403, detail="Only school admins can delete classes")
|
|
|
|
sb = _sb()
|
|
sb.supabase.table("classes").update({"is_active": False}).eq("id", class_id).eq("institute_id", institute_id).execute()
|
|
return {"status": "ok"}
|
|
|
|
|
|
@router.post("/{class_id}/teachers")
|
|
async def add_teacher(
|
|
class_id: str,
|
|
body: AddTeacherRequest,
|
|
credentials: dict = Depends(SupabaseBearer()),
|
|
) -> Dict[str, Any]:
|
|
user_id = credentials.get("sub", "")
|
|
institute_id = _require_institute(user_id)
|
|
|
|
if not _is_school_admin(user_id, institute_id):
|
|
raise HTTPException(status_code=403, detail="Only school admins can assign teachers")
|
|
|
|
sb = _sb()
|
|
res = sb.supabase.table("class_teachers").upsert({
|
|
"class_id": class_id,
|
|
"teacher_id": body.teacher_id,
|
|
"is_primary": body.is_primary,
|
|
"assigned_by": user_id,
|
|
}, on_conflict="class_id,teacher_id").execute()
|
|
return {"status": "ok", "row": (res.data or [{}])[0]}
|
|
|
|
|
|
@router.delete("/{class_id}/teachers/{teacher_id}")
|
|
async def remove_teacher(
|
|
class_id: str,
|
|
teacher_id: str,
|
|
credentials: dict = Depends(SupabaseBearer()),
|
|
) -> Dict[str, Any]:
|
|
user_id = credentials.get("sub", "")
|
|
institute_id = _require_institute(user_id)
|
|
|
|
if not _is_school_admin(user_id, institute_id):
|
|
raise HTTPException(status_code=403, detail="Only school admins can remove teachers")
|
|
|
|
sb = _sb()
|
|
sb.supabase.table("class_teachers").delete().eq("class_id", class_id).eq("teacher_id", teacher_id).execute()
|
|
return {"status": "ok"}
|
|
|
|
|
|
|
|
@router.post("/{class_id}/students")
|
|
async def add_student(
|
|
class_id: str,
|
|
body: AddStudentRequest,
|
|
credentials: dict = Depends(SupabaseBearer()),
|
|
) -> Dict[str, Any]:
|
|
user_id = credentials.get("sub", "")
|
|
institute_id = _require_institute(user_id)
|
|
|
|
if not _is_school_admin(user_id, institute_id):
|
|
raise HTTPException(status_code=403, detail="Only school admins can enroll students")
|
|
|
|
sb = _sb()
|
|
res = sb.supabase.table("class_students").upsert({
|
|
"class_id": class_id,
|
|
"student_id": body.student_id,
|
|
"status": "active",
|
|
"enrolled_by": user_id,
|
|
}, on_conflict="class_id,student_id").execute()
|
|
return {"status": "ok", "row": (res.data or [{}])[0]}
|
|
|
|
|
|
@router.delete("/{class_id}/students/{student_id}")
|
|
async def remove_student(
|
|
class_id: str,
|
|
student_id: str,
|
|
credentials: dict = Depends(SupabaseBearer()),
|
|
) -> Dict[str, Any]:
|
|
user_id = credentials.get("sub", "")
|
|
institute_id = _require_institute(user_id)
|
|
|
|
if not _is_school_admin(user_id, institute_id):
|
|
raise HTTPException(status_code=403, detail="Only school admins can remove students")
|
|
|
|
sb = _sb()
|
|
sb.supabase.table("class_students").delete().eq("class_id", class_id).eq("student_id", student_id).execute()
|
|
return {"status": "ok"}
|
|
|
|
|
|
@router.post("/{class_id}/leave")
|
|
async def leave_class(
|
|
class_id: str,
|
|
credentials: dict = Depends(SupabaseBearer()),
|
|
) -> Dict[str, Any]:
|
|
user_id = credentials.get("sub", "")
|
|
sb = _sb()
|
|
sb.supabase.table("class_students").update({"status": "inactive"}).eq("class_id", class_id).eq("student_id", user_id).execute()
|
|
return {"status": "ok"}
|
|
|
|
|
|
@router.get("/{class_id}/enrollment-requests")
|
|
async def list_enrollment_requests(
|
|
class_id: str,
|
|
credentials: dict = Depends(SupabaseBearer()),
|
|
) -> Dict[str, Any]:
|
|
user_id = credentials.get("sub", "")
|
|
institute_id = _require_institute(user_id)
|
|
|
|
sb = _sb()
|
|
reqs = (
|
|
sb.supabase.table("enrollment_requests")
|
|
.select("*")
|
|
.eq("class_id", class_id)
|
|
.eq("status", "pending")
|
|
.execute()
|
|
.data or []
|
|
)
|
|
return {"requests": reqs}
|
|
|
|
|
|
@router.post("/{class_id}/enroll")
|
|
async def request_enrollment(
|
|
class_id: str,
|
|
credentials: dict = Depends(SupabaseBearer()),
|
|
) -> Dict[str, Any]:
|
|
user_id = credentials.get("sub", "")
|
|
sb = _sb()
|
|
res = sb.supabase.table("enrollment_requests").insert({
|
|
"class_id": class_id,
|
|
"student_id": user_id,
|
|
"status": "pending",
|
|
}).execute()
|
|
return {"status": "ok", "request": (res.data or [{}])[0]}
|
|
|
|
|
|
@router.patch("/{class_id}/enrollment-requests/{request_id}")
|
|
async def respond_enrollment_request(
|
|
class_id: str,
|
|
request_id: str,
|
|
body: dict,
|
|
credentials: dict = Depends(SupabaseBearer()),
|
|
) -> Dict[str, Any]:
|
|
"""Approve or reject a pending enrollment request. School admin only."""
|
|
user_id = credentials.get("sub", "")
|
|
institute_id = _require_institute(user_id)
|
|
if not _is_school_admin(user_id, institute_id):
|
|
raise HTTPException(status_code=403, detail="Only school admins can respond to enrollment requests")
|
|
|
|
action = body.get("action")
|
|
if action not in ("approve", "reject"):
|
|
raise HTTPException(status_code=400, detail="action must be 'approve' or 'reject'")
|
|
|
|
sb = _sb()
|
|
req = (
|
|
sb.supabase.table("enrollment_requests")
|
|
.select("student_id, status")
|
|
.eq("id", request_id)
|
|
.eq("class_id", class_id)
|
|
.single()
|
|
.execute()
|
|
)
|
|
if not req.data:
|
|
raise HTTPException(status_code=404, detail="Request not found")
|
|
if req.data["status"] != "pending":
|
|
raise HTTPException(status_code=400, detail="Request is not pending")
|
|
|
|
student_id = req.data["student_id"]
|
|
new_status = "accepted" if action == "approve" else "rejected"
|
|
sb.supabase.table("enrollment_requests").update({"status": new_status}).eq("id", request_id).execute()
|
|
|
|
if action == "approve":
|
|
sb.supabase.table("class_students").upsert({
|
|
"class_id": class_id,
|
|
"student_id": student_id,
|
|
"status": "active",
|
|
"enrolled_by": user_id,
|
|
}, on_conflict="class_id,student_id").execute()
|
|
|
|
return {"status": "ok", "action": action, "student_id": student_id}
|