| """Classes. A manager (or a developer) makes, changes and deletes them; a class's teacher, or a manager, adds and |
| removes its students; a teacher sees the classes they teach; a manager sees every student.""" |
| from __future__ import annotations |
|
|
| import re |
| from datetime import date |
|
|
| from fastapi import APIRouter, Depends, HTTPException, Request |
| from pydantic import BaseModel, Field |
|
|
| from . import reads |
| from .access import allowed, need |
| from .auth import EMAIL |
| from .store import ClassExists, MemberExists, TeacherIsStudent |
|
|
| MAX_EMAILS = 200 |
| MAX_NAME = 80 |
| router = APIRouter(prefix="/api/classes") |
| teaching = APIRouter(prefix="/api") |
|
|
|
|
| class ClassBody(BaseModel): |
| name: str = Field(max_length=200) |
| lessons: list[str] = Field(default_factory=list, max_length=200) |
| teacher_email: str | None = Field(default=None, max_length=300) |
| students: str | None = Field(default=None, max_length=40000) |
| ai_enabled: bool = False |
| ai_daily_checks: int = 150 |
| subjects: list[str] | None = Field(default=None, max_length=50) |
|
|
|
|
| class NewStudents(BaseModel): |
| emails: str = Field(max_length=20000) |
|
|
|
|
| def parse_emails(text: str) -> tuple[list[str], list[str]]: |
| """(valid, invalid) addresses from pasted text: lower-case, without repeats, in the order given.""" |
| valid: list[str] = [] |
| invalid: list[str] = [] |
| for part in re.split(r"[\s,;]+", text.strip().lower()): |
| if not part: |
| continue |
| bucket = valid if EMAIL.match(part) and len(part) <= 254 else invalid |
| if part not in bucket: |
| bucket.append(part) |
| return valid, invalid |
|
|
|
|
| def shown(request: Request, row: dict, with_students: bool = True) -> dict: |
| out = {k: row[k] for k in ("id", "name", "lessons", "teacher_email", "ai_enabled", "ai_daily_checks", "students")} |
| out["subjects"] = list(row.get("subjects") or ["math"]) |
| if with_students: |
| out["student_emails"] = [m["email"] for m in request.app.state.store.members(row["id"])] |
| return out |
|
|
|
|
| def checked(request: Request, body: ClassBody) -> dict: |
| """The class fields from a request, or an HTTP error that says what is wrong.""" |
| name = " ".join(body.name.split()) |
| if not 1 <= len(name) <= MAX_NAME: |
| raise HTTPException(422, f"砖诐 讛讻讬转讛 爪专讬讱 诇讛讬讜转 讘讬谉 转讜 讗讞讚 诇志{MAX_NAME} 转讜讜讬诐") |
| lessons = [lesson["id"] for lesson in request.app.state.bank.lessons] |
| teacher = (body.teacher_email or "").strip().lower() or None |
| if teacher and not EMAIL.match(teacher): |
| raise HTTPException(422, "讻转讜讘转 讛诪讜专讛 诇讗 谞专讗讬转 讻诪讜 诪讬讬诇") |
| if not 0 <= body.ai_daily_checks <= 10000: |
| raise HTTPException(422, "诪住驻专 讘讚讬拽讜转 讛志AI 讘讬讜诐 爪专讬讱 诇讛讬讜转 讘讬谉 0 诇志10000") |
| roster, invalid = parse_emails(body.students or "") |
| if len(roster) > 2 * MAX_EMAILS: |
| raise HTTPException(422, "讬讜转专 诪讚讬 转诇诪讬讚讬诐 讘讘转 讗讞转") |
| if teacher and teacher in roster: |
| raise HTTPException(422, "诪讜专讛 讛讻讬转讛 诇讗 讬讻讜诇 诇讛讬讜转 讙诐 转诇诪讬讚 讘讛") |
| subjects = None |
| if body.subjects is not None: |
| known = {s["id"] for s in request.app.state.bank.subjects} |
| subjects = list(dict.fromkeys(body.subjects)) |
| if not subjects: |
| raise HTTPException(422, "讻讬转讛 诇讜诪讚转 诇驻讞讜转 诪拽爪讜注 讗讞讚") |
| if any(s not in known for s in subjects): |
| raise HTTPException(422, "讗讞讚 讛诪拽爪讜注讜转 砖谞讘讞专讜 诇讗 拽讬讬诐") |
| return {"name": name, "lessons": lessons, "teacher_email": teacher, "ai_enabled": body.ai_enabled, |
| "ai_daily_checks": body.ai_daily_checks, "roster": roster, "invalid": invalid, "subjects": subjects} |
|
|
|
|
| def sync_roster(store, class_id: int, roster: list[str]) -> None: |
| """Make the class's students exactly `roster`. Removing a student never deletes their progress.""" |
| current = {m["email"] for m in store.members(class_id)} |
| for email in roster: |
| if email not in current: |
| store.add_member(class_id, email) |
| for email in current - set(roster): |
| store.remove_member(class_id, email) |
|
|
|
|
| def own_class_for_roster(request: Request, class_id: int, user: dict) -> dict: |
| """A manager (or developer) reaches any class; a teacher only the classes they teach.""" |
| row = request.app.state.store.get_class(class_id) |
| if not row: |
| raise HTTPException(404, "讛讻讬转讛 诇讗 谞诪爪讗讛") |
| if not allowed(user["roles"], "classes.manage") and row["teacher_email"] != user["email"]: |
| raise HTTPException(404, "讛讻讬转讛 诇讗 谞诪爪讗讛") |
| return row |
|
|
|
|
| |
|
|
| @router.get("") |
| def list_classes(request: Request, user: dict = Depends(need("classes.manage"))) -> list[dict]: |
| return [shown(request, c) for c in request.app.state.store.classes()] |
|
|
|
|
| @router.post("", status_code=201) |
| def create_class(body: ClassBody, request: Request, user: dict = Depends(need("classes.manage"))) -> dict: |
| fields = checked(request, body) |
| store = request.app.state.store |
| try: |
| row = store.create_class(user["email"], fields["name"], lessons=fields["lessons"], teacher_email=fields["teacher_email"], |
| ai_enabled=fields["ai_enabled"], ai_daily_checks=fields["ai_daily_checks"], |
| subjects=fields["subjects"] or ["math"]) |
| except ClassExists: |
| raise HTTPException(409, "讻讘专 讬砖 讻讬转讛 讘砖诐 讛讝讛") from None |
| sync_roster(store, row["id"], fields["roster"]) |
| request.app.state.roles.forget() |
| return {**shown(request, store.get_class(row["id"])), "invalid": fields["invalid"]} |
|
|
|
|
| @router.put("/{class_id}") |
| def change_class(class_id: int, body: ClassBody, request: Request, user: dict = Depends(need("classes.manage"))) -> dict: |
| fields = checked(request, body) |
| store = request.app.state.store |
| try: |
| store.update_class(class_id, name=fields["name"], lessons=fields["lessons"], teacher_email=fields["teacher_email"], |
| ai_enabled=fields["ai_enabled"], ai_daily_checks=fields["ai_daily_checks"], |
| **({"subjects": fields["subjects"]} if fields["subjects"] is not None else {})) |
| except KeyError: |
| raise HTTPException(404, "讛讻讬转讛 诇讗 谞诪爪讗讛") from None |
| except ClassExists: |
| raise HTTPException(409, "讻讘专 讬砖 讻讬转讛 讘砖诐 讛讝讛") from None |
| except TeacherIsStudent: |
| raise HTTPException(422, "诪讜专讛 讛讻讬转讛 诇讗 讬讻讜诇 诇讛讬讜转 讙诐 转诇诪讬讚 讘讛") from None |
| if body.students is not None: |
| sync_roster(store, class_id, fields["roster"]) |
| request.app.state.roles.forget() |
| return {**shown(request, store.get_class(class_id)), "invalid": fields["invalid"]} |
|
|
|
|
| @router.delete("/{class_id}", status_code=204) |
| def delete_class(class_id: int, request: Request, user: dict = Depends(need("classes.manage"))) -> None: |
| if not request.app.state.store.delete_class(class_id): |
| raise HTTPException(404, "讛讻讬转讛 诇讗 谞诪爪讗讛") |
| request.app.state.roles.forget() |
|
|
|
|
| @router.get("/{class_id}") |
| def show_class(class_id: int, request: Request, user: dict = Depends(need("classes.manage"))) -> dict: |
| row = request.app.state.store.get_class(class_id) |
| if not row: |
| raise HTTPException(404, "讛讻讬转讛 诇讗 谞诪爪讗讛") |
| return shown(request, row) |
|
|
|
|
| |
|
|
| @router.post("/{class_id}/students") |
| def add_students(class_id: int, body: NewStudents, request: Request, user: dict = Depends(need("class.roster"))) -> dict: |
| row = own_class_for_roster(request, class_id, user) |
| store = request.app.state.store |
| valid, invalid = parse_emails(body.emails) |
| if len(valid) > MAX_EMAILS: |
| raise HTTPException(422, f"讗驻砖专 诇讛讜住讬祝 注讚 {MAX_EMAILS} 转诇诪讬讚讬诐 讘讻诇 驻注诐") |
| result: dict = {"added": [], "already_here": [], "teacher": [], "invalid": invalid} |
| for email in valid: |
| try: |
| store.add_member(row["id"], email) |
| result["added"].append(email) |
| except MemberExists: |
| result["already_here"].append(email) |
| except TeacherIsStudent: |
| result["teacher"].append(email) |
| request.app.state.roles.forget() |
| return result |
|
|
|
|
| @router.delete("/{class_id}/students/{email:path}", status_code=204) |
| def remove_student(class_id: int, email: str, request: Request, user: dict = Depends(need("class.roster"))) -> None: |
| own_class_for_roster(request, class_id, user) |
| if not request.app.state.store.remove_member(class_id, email.strip().lower()): |
| raise HTTPException(404, "讛转诇诪讬讚 诇讗 谞诪爪讗 讘讻讬转讛") |
| request.app.state.roles.forget(email) |
|
|
|
|
| |
|
|
| @teaching.get("/teaching") |
| def my_classes(request: Request, user: dict = Depends(need("class.teach"))) -> list[dict]: |
| """The classes I teach, each with its students. A developer sees every class.""" |
| store = request.app.state.store |
| rows = store.classes() if allowed(user["roles"], "*") else store.classes_taught(user["email"]) |
| return [{"id": c["id"], "name": c["name"], "lessons": c["lessons"], "ai_enabled": c["ai_enabled"], "subjects": list(c.get("subjects") or ["math"]), |
| "students": [{"email": m["email"], "signed_in": bool(m["user_id"])} for m in store.members(c["id"])]} for c in rows] |
|
|
|
|
| @teaching.get("/students") |
| def all_students(request: Request, user: dict = Depends(need("students.list"))) -> list[dict]: |
| return request.app.state.store.all_students() |
|
|
|
|
| |
|
|
| curriculum = APIRouter(prefix="/api/curriculum") |
|
|
|
|
| class Dates(BaseModel): |
| dates: dict[str, str | None] = Field(max_length=200) |
|
|
|
|
| @curriculum.get("") |
| def read_curriculum(request: Request, user: dict = Depends(need("classes.manage"))) -> list[dict]: |
| """Every published lesson, grouped by subject in the subjects' order and each subject's lessons in order, with its date (or null) |
| and the bank's month as a hint.""" |
| dates, bank = reads.of(request).curriculum(), request.app.state.bank |
| return [{"id": lesson["id"], "name": lesson["name"], "subject": {"id": s["id"], "name": s["name"]}, "month": lesson.get("when", ""), |
| "due_date": dates[lesson["id"]].isoformat() if lesson["id"] in dates else None} |
| for s in bank.subjects for lesson in bank.lessons_of_subject(s["id"])] |
|
|
|
|
| @curriculum.put("") |
| def change_curriculum(body: Dates, request: Request, user: dict = Depends(need("classes.manage"))) -> list[dict]: |
| known = {lesson["id"] for lesson in request.app.state.bank.lessons} |
| parsed: dict[str, date | None] = {} |
| for lesson, text in body.dates.items(): |
| if lesson not in known: |
| raise HTTPException(422, "砖讬注讜专 诇讗 诪讜讻专") |
| try: |
| parsed[lesson] = date.fromisoformat(text) if text else None |
| except ValueError: |
| raise HTTPException(422, "转讗专讬讱 诇讗 转拽讬谉") from None |
| request.app.state.store.set_lesson_dates(parsed) |
| request.state.reads = None |
| return read_curriculum(request, user) |
|
|
