"""Where classes, students, roles and learning progress are kept. Two stores with the same methods:
InMemoryStore (tests and local development, lost on restart) and SupabaseStore (PostgREST with the secret key)."""
from __future__ import annotations
import copy
import re
import itertools
import json
import threading
import urllib.error
import urllib.parse
import uuid
from datetime import date, datetime, timezone
from . import pooled
TIMEOUT = 10
class StoreUnavailable(RuntimeError):
"""The database could not be reached or refused the request."""
class ClassExists(ValueError):
"""The teacher already has a class with this name."""
class MemberExists(ValueError):
"""The email already belongs to this class."""
class SubjectExists(ValueError):
"""A subject with this name (ignoring case) already exists."""
class TeacherIsStudent(ValueError):
"""A class's teacher cannot also be one of its students."""
def _now() -> str:
return datetime.now(timezone.utc).isoformat()
# What the server keeps about one student and one skill (table m2_skill_progress). `mastered_at` is set once.
PROGRESS_FIELDS = ("current_key", "current_helped", "started_at", "mastered_at", "round_start")
PROGRESS_DEFAULTS = {"current_key": None, "current_helped": False, "started_at": None, "mastered_at": None, "round_start": None}
ATTEMPT_FIELDS = {"answers", "submitted_at", "timed_out", "grade", "credit", "checked_at", "checked_by", "review"}
def _copy_attempt(row: dict) -> dict:
return {**row, "questions": [dict(q) for q in row["questions"]], "answers": dict(row["answers"]), "credit": dict(row.get("credit") or {}),
"review": copy.deepcopy(row.get("review"))}
def _when(value):
if value is None or isinstance(value, datetime):
return value
# Python 3.10 reads a fraction of exactly 3 or 6 digits only, and Postgres drops trailing zeros ("...57.63003+00:00")
text = re.sub(r"\.(\d+)", lambda m: "." + m.group(1)[:6].ljust(6, "0"), str(value).replace("Z", "+00:00"), count=1)
return datetime.fromisoformat(text)
def _attempt_row(row: dict) -> dict:
"""A database row as the in-memory store gives it: real datetimes, plain lists and dicts."""
return {**row, "started_at": _when(row["started_at"]), "due_at": _when(row["due_at"]), "submitted_at": _when(row.get("submitted_at")),
"questions": list(row.get("questions") or []), "answers": dict(row.get("answers") or {}), "credit": dict(row.get("credit") or {}),
"checked_at": _when(row.get("checked_at")), "checked_by": row.get("checked_by"), "review": row.get("review")}
TEST_FIELDS = {"date", "lessons", "note", "questions", "minutes", "subject_id"}
CLASS_FIELDS = {"name", "lessons", "teacher_email", "ai_enabled", "ai_daily_checks", "subjects"}
SUBJECT_FIELDS = {"name", "icon", "position"}
LESSON_FIELDS = {"subject_id", "position", "draft", "published", "status", "failure", "files", "published_at"}
AI_FEATURES = ("photo", "check", "worksheet", "prompter") # m2_ai_usage.feature (the database checks the same list)
PROMPTER_FIELDS = {"title", "body", "outline"}
MATH = {"id": "math", "name": "מתמטיקה", "icon": "∑", "position": 0, "created_by": "setup"}
def _lesson_row(row: dict) -> dict:
"""A lesson row as both stores give it: real datetimes, the draft and published documents as plain dicts (published may be None)."""
return {**row, "draft": dict(row.get("draft") or {}), "published": dict(row["published"]) if row.get("published") else None,
"files": list(row.get("files") or []), "created_at": _when(row.get("created_at")), "updated_at": _when(row.get("updated_at")),
"published_at": _when(row.get("published_at"))}
def _check_test(fields: dict) -> None:
"""The limits the database enforces too: 1-40 questions, 5-180 minutes, at least one lesson."""
if "questions" in fields and not 1 <= fields["questions"] <= 40:
raise ValueError("questions must be 1-40")
if "minutes" in fields and not 5 <= fields["minutes"] <= 180:
raise ValueError("minutes must be 5-180")
if "lessons" in fields and not fields["lessons"]:
raise ValueError("a test covers at least one lesson")
def _test_row(row: dict) -> dict:
"""A database row as the in-memory store gives it: a real date and datetimes, a plain list."""
day = row["date"] if isinstance(row["date"], date) else date.fromisoformat(str(row["date"])[:10])
return {**row, "date": day, "lessons": list(row.get("lessons") or []), "subject_id": row.get("subject_id") or "math",
"created_at": _when(row["created_at"]), "removed_at": _when(row.get("removed_at"))}
PICTURE_KINDS = ("page", "answer", "help", "work")
SUBMISSION_FIELDS = {"pictures", "state", "late", "handed_at", "grade", "teacher_note", "checked_at", "checked_by"}
REQUEST_FIELDS = {"state", "picture_id"}
AI_FIELDS = {"state", "picture_id", "skill_id", "snapshot", "result", "teacher_note", "failure", "budget", "sent_by", "sent_at", "done_at"}
AI_STAMPS = ("created_at", "sent_at", "done_at")
REVIEW_SOURCES = ("practice", "quiz", "exercise_test", "simulation")
def _ai_row(row: dict) -> dict:
return {**_stamped(row, *AI_STAMPS)}
def _stamped(row: dict, *names: str) -> dict:
"""A database row with the named timestamp columns as datetimes (the in-memory store keeps datetimes already)."""
return {**row, **{n: _when(row.get(n)) for n in names}}
def _assignment_row(row: dict) -> dict:
day = row["due_date"] if isinstance(row["due_date"], date) else date.fromisoformat(str(row["due_date"])[:10])
return {**_stamped(row, "created_at", "removed_at"), "due_date": day, "pages": [str(x) for x in (row.get("pages") or [])]}
def _submission_row(row: dict) -> dict:
return {**_stamped(row, "handed_at", "checked_at"), "pictures": [str(x) for x in (row.get("pictures") or [])]}
class InMemoryStore:
def __init__(self) -> None:
self._lock = threading.Lock()
self._ids = itertools.count(1)
self._classes: dict[int, dict] = {}
self._members: dict[tuple[int, str], dict] = {} # (class_id, email) -> row
self._roles: dict[tuple[str, str], dict] = {} # (email, role) -> row
self._progress: dict[tuple[str, str], dict] = {} # (user_id, skill_id) -> row
self._answers: list[dict] = []
self._comets: dict[str, dict[str, str]] = {} # user_id -> {comet_id: caught_at}
self._attempts: dict[str, dict] = {} # attempt id -> row
self._work: dict[tuple, dict] = {} # (user id, ref, question key) -> the board's picture for that answer
self._reward: dict[str, dict] = {} # user_id -> {last_comet_day, last_catch_day}
self._tests: dict[int, dict] = {} # test id -> row
self._pictures: dict[str, dict] = {}
self._curriculum: dict[str, date] = {} # lesson id -> the date it should be completed by
self._assignments: dict[int, dict] = {}
self._submissions: dict[int, dict] = {}
self._requests: dict[str, dict] = {}
self._ai: dict[str, dict] = {} # AI requests by id
self._review: dict[tuple[str, str], dict] = {} # (user_id, question_key) -> a wrong answer waiting in the review tab
self._review_done: list[dict] = [] # right answers given in the review tab (the daily goal counts them)
self._allowance: dict[tuple[str, date], int] = {} # (scope, day) -> used
self._ai_usage: list[dict] = []
self._subs: dict[int, dict] = {} # push subscriptions
self._subjects: dict[str, dict] = {"math": {**MATH, "created_at": datetime.now(timezone.utc)}}
self._lessons: dict[int, dict] = {} # lessons kept in the database (drafts and published)
self._prompter: dict[str, dict] = {} # the developers' prompter texts by id
# --- classes ---
def _students(self, class_id: int) -> int:
return sum(1 for (c, _) in self._members if c == class_id)
def _row(self, row: dict) -> dict:
return {**row, "lessons": list(row["lessons"]), "subjects": list(row.get("subjects") or ["math"]), "students": self._students(row["id"])}
def _taken(self, name: str, except_id: int | None = None) -> bool:
return any(c["name"].lower() == name.lower() and c["id"] != except_id for c in self._classes.values())
def create_class(self, created_by: str, name: str, *, lessons=(), teacher_email: str | None = None,
ai_enabled: bool = False, ai_daily_checks: int = 150, subjects=("math",)) -> dict:
with self._lock:
if self._taken(name):
raise ClassExists(name)
row = {"id": next(self._ids), "name": name, "created_by": created_by, "teacher_email": teacher_email,
"lessons": list(lessons), "ai_enabled": ai_enabled, "ai_daily_checks": ai_daily_checks,
"subjects": list(subjects), "created_at": _now()}
self._classes[row["id"]] = row
return self._row(row)
def update_class(self, class_id: int, **fields) -> dict:
unknown = set(fields) - CLASS_FIELDS
if unknown:
raise ValueError(f"unknown class fields {sorted(unknown)}")
with self._lock:
row = self._classes.get(class_id)
if row is None:
raise KeyError(class_id)
if "name" in fields and self._taken(fields["name"], class_id):
raise ClassExists(fields["name"])
teacher = fields.get("teacher_email", row["teacher_email"])
if teacher and (class_id, teacher) in self._members:
raise TeacherIsStudent(teacher)
row.update({k: (list(v) if k in ("lessons", "subjects") else v) for k, v in fields.items()})
return self._row(row)
def delete_class(self, class_id: int) -> bool:
with self._lock:
if self._classes.pop(class_id, None) is None:
return False
for key in [k for k in self._members if k[0] == class_id]:
del self._members[key]
return True
def get_class(self, class_id: int) -> dict | None:
with self._lock:
row = self._classes.get(class_id)
return self._row(row) if row else None
def classes(self) -> list[dict]:
with self._lock:
return [self._row(c) for c in self._classes.values()]
def classes_taught(self, email: str) -> list[dict]:
with self._lock:
return [self._row(c) for c in self._classes.values() if c["teacher_email"] == email]
# --- students ---
def members(self, class_id: int) -> list[dict]:
with self._lock:
rows = [m for (c, _), m in self._members.items() if c == class_id]
return [{"email": m["email"], "user_id": m["user_id"], "added_at": m["added_at"]}
for m in sorted(rows, key=lambda m: (m["added_at"], m["email"]))]
def classes_of_student(self, email: str) -> list[dict]:
with self._lock:
return [{"id": c["id"], "name": c["name"], "lessons": list(c["lessons"]), "ai_enabled": c["ai_enabled"], "ai_daily_checks": c["ai_daily_checks"],
"subjects": list(c.get("subjects") or ["math"])}
for (cid, e), _ in self._members.items() if e == email for c in [self._classes[cid]]]
def add_member(self, class_id: int, email: str) -> None:
with self._lock:
if self._classes[class_id]["teacher_email"] == email:
raise TeacherIsStudent(email)
if (class_id, email) in self._members:
raise MemberExists(email)
user_id = next((m["user_id"] for (_, e), m in self._members.items() if e == email and m["user_id"]), None)
self._members[(class_id, email)] = {"class_id": class_id, "email": email, "user_id": user_id, "added_at": _now()}
def remove_member(self, class_id: int, email: str) -> bool:
with self._lock:
return self._members.pop((class_id, email), None) is not None
def link_user(self, email: str, user_id: str) -> None:
with self._lock:
for (_, e), row in self._members.items():
if e == email and not row["user_id"]:
row["user_id"] = user_id
def all_students(self) -> list[dict]:
with self._lock:
found: dict[str, list[str]] = {}
for (cid, email) in self._members:
found.setdefault(email, []).append(self._classes[cid]["name"])
return [{"email": e, "classes": sorted(names)} for e, names in sorted(found.items())]
# --- roles (given by an admin; the settings add developers and the first admins) ---
def roles_of(self, email: str) -> set[str]:
with self._lock:
return {role for (e, role) in self._roles if e == email}
def give_role(self, email: str, role: str, granted_by: str) -> None:
with self._lock:
self._roles.setdefault((email, role), {"email": email, "role": role, "granted_by": granted_by, "granted_at": _now()})
def take_role(self, email: str, role: str, settings_admins: int = 0) -> str:
""""removed", "absent", or "last_admin" (nothing was removed): the last admin keeps the role.
`settings_admins` counts the admins named in the app's settings, who stay admins whatever the table says."""
with self._lock:
if (email, role) not in self._roles:
return "absent"
if role == "admin":
others = sum(1 for (e, r) in self._roles if r == "admin" and e != email)
if others + settings_admins == 0:
return "last_admin"
del self._roles[(email, role)]
return "removed"
def accounts(self) -> list[dict]:
"""Every email the app knows (a role, a class entry or a class's teacher) with the roles given in the table."""
with self._lock:
known = {e for (e, _) in self._roles} | {e for (_, e) in self._members}
known |= {c["teacher_email"] for c in self._classes.values() if c["teacher_email"]}
return [{"email": e, "roles": sorted(r for (x, r) in self._roles if x == e)} for e in sorted(known)]
# --- skill learning ---
def get_progress(self, user_id: str, skill_id: str) -> dict | None:
with self._lock:
row = self._progress.get((user_id, skill_id))
return dict(row) if row else None
def save_progress(self, user_id: str, skill_id: str, **fields) -> dict:
"""Create the row if needed, then set the given fields; the others stay as they are."""
unknown = set(fields) - set(PROGRESS_FIELDS)
if unknown:
raise ValueError(f"unknown progress fields {sorted(unknown)}")
with self._lock:
row = self._progress.setdefault((user_id, skill_id), {"user_id": user_id, "skill_id": skill_id,
**PROGRESS_DEFAULTS})
row.update(fields)
return dict(row)
def progress_of(self, user_id: str) -> list[dict]:
with self._lock:
return [dict(r) for (u, _), r in self._progress.items() if u == user_id]
# --- many students at once (a class's progress screen: one call each, not one per student) ---
def progress_of_users(self, user_ids) -> dict[str, list[dict]]:
wanted = set(user_ids)
with self._lock:
found: dict[str, list[dict]] = {u: [] for u in dict.fromkeys(user_ids)}
for (u, _), r in self._progress.items():
if u in wanted:
found[u].append(dict(r))
return found
def last_answered_users(self, user_ids) -> dict[str, dict[str, date]]:
wanted = set(user_ids)
with self._lock:
found: dict[str, dict[str, date]] = {u: {} for u in dict.fromkeys(user_ids)}
for a in self._answers:
days = found.get(a["user_id"])
if days is not None and (a["skill_id"] not in days or a["day"] > days[a["skill_id"]]):
days[a["skill_id"]] = a["day"]
return found
def attempts_of_users(self, user_ids, kind: str) -> dict[str, list[dict]]:
wanted = set(user_ids)
with self._lock:
found: dict[str, list[dict]] = {u: [] for u in dict.fromkeys(user_ids)}
for r in sorted(self._attempts.values(), key=lambda r: r["started_at"]):
if r["user_id"] in wanted and r["kind"] == kind:
found[r["user_id"]].append(_copy_attempt(r))
return found
def add_answer(self, user_id: str, skill_id: str, question_key: str, correct: bool, counted: bool,
mistake_id: str | None, day: date) -> None:
with self._lock:
self._answers.append({"id": len(self._answers) + 1, "user_id": user_id, "skill_id": skill_id,
"question_key": question_key, "correct": correct, "counted": counted,
"mistake_id": mistake_id, "day": day})
def last_counted_answers(self, user_id: str, skill_id: str, n: int) -> list[dict]:
"""The last `n` answers that were not given with help, oldest first."""
with self._lock:
rows = [a for a in self._answers if a["user_id"] == user_id and a["skill_id"] == skill_id and a["counted"]]
return [{"question_key": a["question_key"], "correct": a["correct"], "day": a["day"]} for a in rows[-n:]]
def answer_counts(self, user_id: str, skill_id: str) -> dict[str, int]:
with self._lock:
counts: dict[str, int] = {}
for a in self._answers:
if a["user_id"] == user_id and a["skill_id"] == skill_id:
counts[a["question_key"]] = counts.get(a["question_key"], 0) + 1
return counts
def answers_of(self, user_id: str, skill_id: str) -> list[dict]:
"""All the student's answers on a skill, oldest first: question_key, correct, counted, day. One read gives what mastery,
the next question and the progress need."""
with self._lock:
return [{k: a[k] for k in ("question_key", "correct", "counted", "day", "mistake_id")} for a in self._answers
if a["user_id"] == user_id and a["skill_id"] == skill_id]
def recent_skills(self, user_id: str, n: int = 20) -> list[str]:
"""The skills the student answered most recently, newest first, each once (helped and wrong answers count)."""
with self._lock:
found: list[str] = []
for a in reversed(self._answers):
if a["user_id"] == user_id and a["skill_id"] not in found:
found.append(a["skill_id"])
if len(found) == n:
break
return found
def solved_keys(self, user_id: str, skill_id: str) -> set[str]:
"""The questions of a skill the student has answered correctly at least once (with or without help)."""
with self._lock:
return {a["question_key"] for a in self._answers
if a["user_id"] == user_id and a["skill_id"] == skill_id and a["correct"]}
# --- real tests (a teacher sets them for a class) ---
def create_test(self, class_id: int, test_day: date, lessons: list[str], created_by: str, *, note: str = "", questions: int = 12,
minutes: int = 45, created_at: datetime | None = None, subject_id: str = "math") -> dict:
fields = {"lessons": list(lessons), "questions": questions, "minutes": minutes, "subject_id": subject_id}
_check_test(fields)
with self._lock:
row = {"id": next(self._ids), "class_id": class_id, "date": test_day, **fields, "note": note, "created_by": created_by,
"created_at": created_at or datetime.now(timezone.utc), "removed_at": None}
self._tests[row["id"]] = row
return dict(row, lessons=list(row["lessons"]))
def get_test(self, test_id: int) -> dict | None:
with self._lock:
row = self._tests.get(test_id)
return dict(row, lessons=list(row["lessons"])) if row else None
def update_test(self, test_id: int, **fields) -> dict:
unknown = set(fields) - TEST_FIELDS
if unknown:
raise ValueError(f"unknown test fields {sorted(unknown)}")
_check_test(fields)
with self._lock:
row = self._tests.get(test_id)
if row is None or row["removed_at"]:
raise KeyError(test_id)
row.update({k: (list(v) if k == "lessons" else v) for k, v in fields.items()})
return dict(row, lessons=list(row["lessons"]))
def remove_test(self, test_id: int, at: datetime | None = None) -> bool:
"""Soft delete: the test disappears from lists, grades already earned keep its name."""
with self._lock:
row = self._tests.get(test_id)
if row is None or row["removed_at"]:
return False
row["removed_at"] = at or datetime.now(timezone.utc)
return True
def tests_of_classes(self, class_ids, include_removed: bool = False) -> list[dict]:
"""The tests of these classes, the nearest date first."""
wanted = set(class_ids)
with self._lock:
rows = [r for r in self._tests.values() if r["class_id"] in wanted and (include_removed or not r["removed_at"])]
return [dict(r, lessons=list(r["lessons"])) for r in sorted(rows, key=lambda r: (r["date"], r["id"]))]
# --- pictures, assignments and submissions (written work) ---
def create_picture(self, owner_id: str, kind: str, path: str, content_type: str, size: int, created_at: datetime | None = None) -> dict:
if kind not in PICTURE_KINDS:
raise ValueError(f"unknown picture kind {kind!r}")
row = {"id": str(uuid.uuid4()), "owner_id": owner_id, "kind": kind, "path": path, "content_type": content_type, "size": size,
"created_at": created_at or datetime.now(timezone.utc), "deleted_at": None}
with self._lock:
self._pictures[row["id"]] = row
return dict(row)
def get_picture(self, picture_id: str) -> dict | None:
with self._lock:
row = self._pictures.get(str(picture_id))
return dict(row) if row else None
def mark_picture_deleted(self, picture_id: str, at: datetime | None = None) -> bool:
with self._lock:
row = self._pictures.get(str(picture_id))
if row is None or row["deleted_at"]:
return False
row["deleted_at"] = at or datetime.now(timezone.utc)
return True
def pictures(self, kind: str | None = None, owner_id: str | None = None, include_deleted: bool = False) -> list[dict]:
with self._lock:
return [dict(r) for r in sorted(self._pictures.values(), key=lambda r: r["created_at"])
if (kind is None or r["kind"] == kind) and (owner_id is None or r["owner_id"] == owner_id) and (include_deleted or not r["deleted_at"])]
def picture_refs(self, picture_id: str) -> dict:
"""What refers to a picture: assignments (as a page), submissions (as an answer) and photo requests."""
pid = str(picture_id)
with self._lock:
return {"assignments": sorted(a["id"] for a in self._assignments.values() if pid in a["pages"]),
"submissions": sorted(s["id"] for s in self._submissions.values() if pid in s["pictures"]),
"requests": sorted(r["id"] for r in self._requests.values() if r["picture_id"] == pid),
"ai_requests": sorted(r["id"] for r in self._ai.values() if r["picture_id"] == pid),
"work": sorted(f"{w['ref']}:{w['question_key']}" for w in self._work.values() if w["picture_id"] == pid)}
# --- the work on the board: one picture per student, attempt (or "practice") and question ---
def save_work(self, user_id: str, ref: str, question_key: str, skill_id: str, picture_id: str, at: datetime | None = None) -> str | None:
"""Keep this picture as the work for the answer. Returns the picture it replaced (nothing refers to it any more), or None."""
with self._lock:
old = self._work.get((user_id, ref, question_key))
self._work[(user_id, ref, question_key)] = {"user_id": user_id, "ref": ref, "question_key": question_key, "skill_id": skill_id,
"picture_id": str(picture_id), "saved_at": at or datetime.now(timezone.utc)}
return old["picture_id"] if old and old["picture_id"] != str(picture_id) else None
def work_of(self, user_id: str, ref: str) -> dict[str, str]:
"""The question keys of an attempt (or "practice") that have work, each with its picture id."""
with self._lock:
return {w["question_key"]: w["picture_id"] for (u, r, _), w in self._work.items() if u == user_id and r == ref}
def create_assignment(self, class_id: int, title: str, due_date: date, created_by: str, kind: str, *, note: str = "", pages=(),
quiz_lesson: str | None = None, test_id: int | None = None, exercise_number: int | None = None,
created_at: datetime | None = None) -> dict:
ok = {"pages": bool(pages), "quiz": bool(quiz_lesson), "exercise_test": test_id is not None and exercise_number in (1, 2)}
if not ok.get(kind) or not 1 <= len(title) <= 200:
raise ValueError("an assignment needs a title and its content")
with self._lock:
row = {"id": next(self._ids), "class_id": class_id, "title": title, "note": note, "due_date": due_date, "kind": kind,
"pages": [str(p) for p in pages], "quiz_lesson": quiz_lesson, "test_id": test_id, "exercise_number": exercise_number,
"created_by": created_by, "created_at": created_at or datetime.now(timezone.utc), "removed_at": None}
self._assignments[row["id"]] = row
return dict(row, pages=list(row["pages"]))
def get_assignment(self, assignment_id: int) -> dict | None:
with self._lock:
row = self._assignments.get(assignment_id)
return dict(row, pages=list(row["pages"])) if row else None
def assignments_of_classes(self, class_ids, include_removed: bool = False) -> list[dict]:
"""The assignments of these classes, the nearest due date first."""
wanted = set(class_ids)
with self._lock:
rows = [r for r in self._assignments.values() if r["class_id"] in wanted and (include_removed or not r["removed_at"])]
return [dict(r, pages=list(r["pages"])) for r in sorted(rows, key=lambda r: (r["due_date"], r["id"]))]
def remove_assignment(self, assignment_id: int, at: datetime | None = None) -> bool:
with self._lock:
row = self._assignments.get(assignment_id)
if row is None or row["removed_at"]:
return False
row["removed_at"] = at or datetime.now(timezone.utc)
return True
def get_submission(self, assignment_id: int, user_id: str) -> dict | None:
with self._lock:
row = next((r for r in self._submissions.values() if r["assignment_id"] == assignment_id and r["user_id"] == user_id), None)
return dict(row, pictures=list(row["pictures"])) if row else None
def get_submission_by_id(self, submission_id: int) -> dict | None:
with self._lock:
row = self._submissions.get(submission_id)
return dict(row, pictures=list(row["pictures"])) if row else None
def save_submission(self, assignment_id: int, user_id: str, **fields) -> dict:
"""Create the student's submission or change the given fields of it."""
unknown = set(fields) - SUBMISSION_FIELDS
if unknown:
raise ValueError(f"unknown submission fields {sorted(unknown)}")
with self._lock:
row = next((r for r in self._submissions.values() if r["assignment_id"] == assignment_id and r["user_id"] == user_id), None)
candidate = dict(row) if row else {"id": next(self._ids), "assignment_id": assignment_id, "user_id": user_id, "pictures": [], "state": "handed_in",
"late": False, "handed_at": datetime.now(timezone.utc), "grade": None, "teacher_note": "", "checked_at": None,
"checked_by": None}
candidate.update({k: ([str(x) for x in v] if k == "pictures" else v) for k, v in fields.items()})
if candidate["state"] == "graded" and candidate["grade"] is None:
raise ValueError("a graded submission has a grade") # refused before anything is stored
if candidate["grade"] is not None and not 0 <= candidate["grade"] <= 100:
raise ValueError("a grade is 0 to 100")
self._submissions[candidate["id"]] = candidate
return dict(candidate, pictures=list(candidate["pictures"]))
def submissions_of_assignments(self, assignment_ids) -> list[dict]:
wanted = set(assignment_ids)
with self._lock:
return [dict(r, pictures=list(r["pictures"])) for r in sorted(self._submissions.values(), key=lambda r: (r["handed_at"], r["id"]))
if r["assignment_id"] in wanted]
def submissions_of_user(self, user_id: str) -> list[dict]:
with self._lock:
return [dict(r, pictures=list(r["pictures"])) for r in self._submissions.values() if r["user_id"] == user_id]
# --- the review list (חזרה): wrong answers to redo ---
def add_review(self, user_id: str, items, at: datetime | None = None) -> int:
"""Add wrong answers, each `(skill_id, question_key, source)`; one already in the list is not added again. Returns how many were added."""
added = 0
with self._lock:
for skill_id, key, source in items:
if source not in REVIEW_SOURCES:
raise ValueError(f"unknown review source {source!r}")
if (user_id, key) not in self._review:
self._review[(user_id, key)] = {"user_id": user_id, "question_key": key, "skill_id": skill_id, "source": source,
"wrong_at": at or datetime.now(timezone.utc)}
added += 1
return added
def review_of(self, user_id: str) -> list[dict]:
"""The student's wrong answers waiting to be redone, the oldest first."""
with self._lock:
return [dict(r) for r in sorted((r for (u, _), r in self._review.items() if u == user_id), key=lambda r: (r["wrong_at"], r["question_key"]))]
def remove_review(self, user_id: str, question_key: str) -> bool:
with self._lock:
return self._review.pop((user_id, question_key), None) is not None
def add_review_done(self, user_id: str, question_key: str, day: date) -> None:
with self._lock:
self._review_done.append({"user_id": user_id, "question_key": question_key, "day": day})
# --- AI help: requests, allowances (budgets) and usage ---
def create_ai_request(self, kind: str, user_id: str, class_id: int | None, *, picture_id: str | None = None, assignment_id: int | None = None,
skill_id: str | None = None, snapshot: dict | None = None, created_at: datetime | None = None) -> dict:
if kind not in ("photo", "check", "again") or (kind == "photo" and not picture_id) or (kind != "photo" and assignment_id is None):
raise ValueError("an AI request needs its picture (a photo) or its assignment (a check, a look again)")
with self._lock:
row = {"id": str(uuid.uuid4()), "kind": kind, "user_id": user_id, "class_id": class_id, "state": "waiting", "picture_id": picture_id,
"assignment_id": assignment_id, "skill_id": skill_id, "snapshot": copy.deepcopy(snapshot), "result": None, "teacher_note": "",
"failure": None, "budget": None, "sent_by": None, "created_at": created_at or datetime.now(timezone.utc), "sent_at": None, "done_at": None}
self._ai[row["id"]] = row
return copy.deepcopy(row)
def get_ai_request(self, request_id: str) -> dict | None:
with self._lock:
row = self._ai.get(str(request_id))
return copy.deepcopy(row) if row else None
def save_ai_request(self, request_id: str, **fields) -> dict:
unknown = set(fields) - AI_FIELDS
if unknown:
raise ValueError(f"unknown AI request fields {sorted(unknown)}")
with self._lock:
row = self._ai.get(str(request_id))
if row is None:
raise KeyError(request_id)
row.update(copy.deepcopy(fields))
return copy.deepcopy(row)
def ai_requests(self, *, user_id: str | None = None, class_ids=None, states=None, kinds=None, assignment_id: int | None = None,
picture_id: str | None = None) -> list[dict]:
"""Requests, oldest first, narrowed by who asked, the classes, the states, the kinds and the assignment."""
with self._lock:
rows = [r for r in self._ai.values()
if (user_id is None or r["user_id"] == user_id) and (class_ids is None or r["class_id"] in set(class_ids))
and (states is None or r["state"] in set(states)) and (kinds is None or r["kind"] in set(kinds))
and (assignment_id is None or r["assignment_id"] == assignment_id) and (picture_id is None or r["picture_id"] == str(picture_id))]
return [copy.deepcopy(r) for r in sorted(rows, key=lambda r: (r["created_at"], r["id"]))]
def start_ai_request(self, request_id: str, sent_by: str, at: datetime) -> dict | None:
"""waiting (or failed, for a retry) -> processing in one step; None when it was neither (someone else sent or declined it first)."""
with self._lock:
row = self._ai.get(str(request_id))
if row is None or row["state"] not in ("waiting", "failed"):
return None
row.update(state="processing", failure=None, sent_by=sent_by, sent_at=at)
return copy.deepcopy(row)
def requeue_processing_ai_requests(self) -> int:
"""After a restart: requests that were being processed go back to waiting."""
with self._lock:
rows = [r for r in self._ai.values() if r["state"] == "processing"]
for r in rows:
r["state"] = "waiting"
return len(rows)
def take_ai_allowance(self, scope: str, day: date, n: int, cap: int) -> bool:
"""Take n of the day's allowance only if it stays within the cap, in one step."""
with self._lock:
used = self._allowance.get((scope, day), 0)
if used + n > cap:
return False
self._allowance[(scope, day)] = used + n
return True
def give_back_ai_allowance(self, scope: str, day: date, n: int) -> None:
with self._lock:
self._allowance[(scope, day)] = max(self._allowance.get((scope, day), 0) - n, 0)
def ai_allowance_used(self, scope: str, day: date) -> int:
with self._lock:
return self._allowance.get((scope, day), 0)
def add_ai_usage(self, row: dict) -> None:
if row.get("feature") not in AI_FEATURES:
raise ValueError(f"unknown AI usage feature {row.get('feature')!r}") # the database refuses it too
with self._lock:
self._ai_usage.append({**row, "created_at": row.get("created_at") or datetime.now(timezone.utc)})
def ai_usage_rows(self) -> list[dict]:
with self._lock:
return [dict(r) for r in self._ai_usage]
# --- photo requests and push subscriptions ---
def create_photo_request(self, user_id: str, assignment_id: int, device: str, code: str, created_at: datetime, expires_at: datetime) -> dict:
"""A new request; the student's earlier waiting request for the same assignment becomes superseded."""
with self._lock:
for r in self._requests.values():
if r["user_id"] == user_id and r["assignment_id"] == assignment_id and r["state"] == "waiting":
r["state"] = "superseded"
row = {"id": str(uuid.uuid4()), "user_id": user_id, "assignment_id": assignment_id, "device": device, "code": code, "state": "waiting",
"created_at": created_at, "expires_at": expires_at, "picture_id": None}
self._requests[row["id"]] = row
return dict(row)
def get_photo_request(self, request_id: str) -> dict | None:
with self._lock:
row = self._requests.get(str(request_id))
return dict(row) if row else None
def save_photo_request(self, request_id: str, **fields) -> dict:
unknown = set(fields) - REQUEST_FIELDS
if unknown:
raise ValueError(f"unknown photo request fields {sorted(unknown)}")
with self._lock:
row = self._requests.get(str(request_id))
if row is None:
raise KeyError(request_id)
row.update(fields)
return dict(row)
def open_photo_requests(self, user_id: str) -> list[dict]:
"""The student's requests still waiting for or receiving a picture, the newest first (expiry is decided by the caller)."""
with self._lock:
return [dict(r) for r in sorted(self._requests.values(), key=lambda r: r["created_at"], reverse=True)
if r["user_id"] == user_id and r["state"] in ("waiting", "uploading")]
def add_push_subscription(self, user_id: str, endpoint: str, p256dh: str, auth: str, key_id: str) -> dict:
with self._lock:
row = next((r for r in self._subs.values() if r["endpoint"] == endpoint), None)
if row is None:
row = {"id": next(self._ids), "endpoint": endpoint, "created_at": datetime.now(timezone.utc)}
self._subs[row["id"]] = row
row.update({"user_id": user_id, "p256dh": p256dh, "auth": auth, "key_id": key_id})
return dict(row)
def push_subscriptions_of(self, user_id: str) -> list[dict]:
with self._lock:
return [dict(r) for r in self._subs.values() if r["user_id"] == user_id]
def delete_push_subscription(self, endpoint: str) -> bool:
with self._lock:
found = [i for i, r in self._subs.items() if r["endpoint"] == endpoint]
for i in found:
del self._subs[i]
return bool(found)
def drop_push_subscriptions_not_of(self, key_id: str) -> int:
"""The server's push key changed: subscriptions made under another key are dropped, and each phone turns notifications on again."""
with self._lock:
gone = [i for i, r in self._subs.items() if r["key_id"] != key_id]
for i in gone:
del self._subs[i]
return len(gone)
# --- timed attempts (quizzes; later exercise tests and the simulation) ---
def create_attempt(self, user_id: str, kind: str, scope: str, started_at: datetime, due_at: datetime, questions: list[dict]) -> dict:
row = {"id": str(uuid.uuid4()), "user_id": user_id, "kind": kind, "scope": scope, "started_at": started_at, "due_at": due_at,
"submitted_at": None, "timed_out": False, "grade": None, "questions": [dict(q) for q in questions], "answers": {},
"credit": {}, "checked_at": None, "checked_by": None, "review": None}
with self._lock:
self._attempts[row["id"]] = row
return _copy_attempt(row)
def get_attempt(self, attempt_id: str) -> dict | None:
with self._lock:
row = self._attempts.get(str(attempt_id))
return _copy_attempt(row) if row else None
def save_attempt(self, attempt_id: str, **fields) -> dict:
"""Update `answers`, `submitted_at`, `timed_out` and/or `grade`."""
unknown = set(fields) - ATTEMPT_FIELDS
if unknown:
raise ValueError(f"unknown attempt fields {sorted(unknown)}")
with self._lock:
row = self._attempts.get(str(attempt_id))
if row is None:
raise KeyError(attempt_id)
row.update({k: (dict(v) if k in ("answers", "credit") else v) for k, v in fields.items()})
return _copy_attempt(row)
def save_open_answers(self, attempt_id: str, answers: dict) -> dict | None:
"""Save the answers only while the attempt is still open, in one step; None once it was submitted (the grade stands)."""
with self._lock:
row = self._attempts.get(str(attempt_id))
if row is None:
raise KeyError(attempt_id)
if row["submitted_at"] is not None:
return None
row["answers"] = dict(answers)
return _copy_attempt(row)
def attempts_of(self, user_id: str, kind: str, scope: str | None = None) -> list[dict]:
"""A student's attempts of one kind (of one scope, when given), oldest first."""
with self._lock:
rows = [r for r in self._attempts.values() if r["user_id"] == user_id and r["kind"] == kind and (scope is None or r["scope"] == scope)]
return [_copy_attempt(r) for r in sorted(rows, key=lambda r: r["started_at"])]
def comets_of(self, user_id: str) -> set[str]:
with self._lock:
return set(self._comets.get(user_id, {}))
def add_comet(self, user_id: str, comet_id: str) -> None:
with self._lock:
self._comets.setdefault(user_id, {}).setdefault(comet_id, datetime.now(timezone.utc).isoformat())
def reward_state(self, user_id: str) -> dict | None:
with self._lock:
row = self._reward.get(user_id)
return dict(row) if row else None
def save_reward_state(self, user_id: str, **fields) -> dict:
"""Set `last_comet_day` and/or `last_catch_day`; the other field stays as it was."""
with self._lock:
row = self._reward.setdefault(user_id, {"user_id": user_id, "last_comet_day": None, "last_catch_day": None})
row.update(fields)
return dict(row)
# --- the curriculum: the date by which each lesson should be completed ---
def curriculum(self) -> dict[str, date]:
with self._lock:
return dict(self._curriculum)
def set_lesson_dates(self, dates: dict[str, date | None]) -> None:
"""Set each lesson's date; None clears it."""
with self._lock:
for lesson, day in dates.items():
if day is None:
self._curriculum.pop(lesson, None)
else:
self._curriculum[lesson] = day
def last_answered(self, user_id: str) -> dict[str, date]:
"""The day of the student's newest answer on each skill (helped and wrong answers count): what fading reads."""
with self._lock:
days: dict[str, date] = {}
for a in self._answers:
if a["user_id"] == user_id and (a["skill_id"] not in days or a["day"] > days[a["skill_id"]]):
days[a["skill_id"]] = a["day"]
return days
# --- subjects and the lessons kept in the database ---
def subjects(self) -> list[dict]:
with self._lock:
return [dict(s) for s in sorted(self._subjects.values(), key=lambda s: (s["position"], s["created_at"]))]
def create_subject(self, subject_id: str, name: str, created_by: str, *, icon: str = "∑") -> dict:
with self._lock:
if subject_id in self._subjects or any(s["name"].lower() == name.lower() for s in self._subjects.values()):
raise SubjectExists(name)
row = {"id": subject_id, "name": name, "icon": icon, "created_by": created_by, "created_at": datetime.now(timezone.utc),
"position": max((s["position"] for s in self._subjects.values()), default=0) + 1}
self._subjects[subject_id] = row
return dict(row)
def update_subject(self, subject_id: str, **fields) -> dict:
unknown = set(fields) - SUBJECT_FIELDS
if unknown:
raise ValueError(f"unknown subject fields {sorted(unknown)}")
with self._lock:
row = self._subjects.get(subject_id)
if row is None:
raise KeyError(subject_id)
if "name" in fields and any(s["name"].lower() == fields["name"].lower() and s["id"] != subject_id for s in self._subjects.values()):
raise SubjectExists(fields["name"])
row.update(fields)
return dict(row)
def delete_subject(self, subject_id: str) -> bool:
with self._lock:
return self._subjects.pop(subject_id, None) is not None
def create_lesson(self, subject_id: str, created_by: str, draft: dict, *, source: str = "json", status: str = "ready",
files: list | None = None) -> dict:
with self._lock:
if subject_id not in self._subjects:
raise KeyError(subject_id)
now = datetime.now(timezone.utc)
position = max((l["position"] for l in self._lessons.values() if l["subject_id"] == subject_id), default=0) + 1
row = {"id": next(self._ids), "subject_id": subject_id, "position": position, "draft": copy.deepcopy(draft), "published": None,
"status": status, "failure": None, "files": list(files or []), "source": source, "created_by": created_by,
"created_at": now, "updated_at": now, "published_at": None}
self._lessons[row["id"]] = row
return _lesson_row(copy.deepcopy(row))
def get_lesson(self, lesson_id: int) -> dict | None:
with self._lock:
row = self._lessons.get(lesson_id)
return _lesson_row(copy.deepcopy(row)) if row else None
def lessons_of(self, subject_id: str | None = None) -> list[dict]:
"""The lessons kept in the database (drafts and published), of one subject or all, in subject and position order."""
with self._lock:
rows = [l for l in self._lessons.values() if subject_id is None or l["subject_id"] == subject_id]
return [_lesson_row(copy.deepcopy(l)) for l in sorted(rows, key=lambda l: (l["subject_id"], l["position"], l["id"]))]
def published_lessons(self) -> list[dict]:
return [l for l in self.lessons_of() if l["published"] is not None]
def save_lesson(self, lesson_id: int, **fields) -> dict:
unknown = set(fields) - LESSON_FIELDS
if unknown:
raise ValueError(f"unknown lesson fields {sorted(unknown)}")
with self._lock:
row = self._lessons.get(lesson_id)
if row is None:
raise KeyError(lesson_id)
row.update(copy.deepcopy(fields))
row["updated_at"] = datetime.now(timezone.utc)
return _lesson_row(copy.deepcopy(row))
def delete_lesson(self, lesson_id: int) -> bool:
with self._lock:
return self._lessons.pop(lesson_id, None) is not None
# --- the developers' prompter texts (each sees only their own) ---
def prompter_texts(self, user_id: str) -> list[dict]:
"""The account's texts, the last changed first."""
with self._lock:
mine = [dict(t) for t in self._prompter.values() if t["user_id"] == user_id]
return sorted(mine, key=lambda t: t["updated_at"], reverse=True)
def get_prompter_text(self, user_id: str, text_id: str) -> dict | None:
with self._lock:
row = self._prompter.get(text_id)
return dict(row) if row and row["user_id"] == user_id else None
def add_prompter_text(self, user_id: str, title: str, body: str) -> dict:
with self._lock:
row = {"id": str(uuid.uuid4()), "user_id": user_id, "title": title, "body": body, "outline": None, "updated_at": datetime.now(timezone.utc)}
self._prompter[row["id"]] = row
return dict(row)
def save_prompter_text(self, user_id: str, text_id: str, *, touch: bool = True, **fields) -> dict | None:
"""Change the given fields of the account's own text (None when it is not theirs). `touch`: the text changed (not only its outline)."""
unknown = set(fields) - PROMPTER_FIELDS
if unknown:
raise ValueError(f"unknown prompter fields {sorted(unknown)}")
with self._lock:
row = self._prompter.get(text_id)
if row is None or row["user_id"] != user_id:
return None
row.update(fields)
if touch:
row["updated_at"] = datetime.now(timezone.utc)
return dict(row)
def delete_prompter_text(self, user_id: str, text_id: str) -> bool:
with self._lock:
row = self._prompter.get(text_id)
if row is None or row["user_id"] != user_id:
return False
del self._prompter[text_id]
return True
def activity_since(self, first: date) -> list[dict]:
"""One row per person and Israel day from `first` on, with their practice answers and started attempts (view m2_activity)."""
from .clock import school_day
with self._lock:
found: dict[tuple, dict] = {}
for a in self._answers:
if a["day"] >= first:
found.setdefault((a["day"], a["user_id"]), {"day": a["day"], "user_id": a["user_id"], "answers": 0, "attempts": 0})["answers"] += 1
for r in self._attempts.values():
day = school_day(r["started_at"])
if day >= first:
found.setdefault((day, r["user_id"]), {"day": day, "user_id": r["user_id"], "answers": 0, "attempts": 0})["attempts"] += 1
return sorted(found.values(), key=lambda r: (r["day"], r["user_id"]))
def daily_correct(self, user_id: str) -> dict[date, int]:
"""Correct practice answers per day (helped answers included): the daily goal and the streak."""
with self._lock:
counts: dict[date, int] = {}
for a in self._answers:
if a["user_id"] == user_id and a["correct"]:
counts[a["day"]] = counts.get(a["day"], 0) + 1
for r in self._review_done: # a right answer in the review tab counts toward the daily goal too
if r["user_id"] == user_id:
counts[r["day"]] = counts.get(r["day"], 0) + 1
return counts
class SupabaseStore:
def __init__(self, url: str, secret: str) -> None:
self.url, self.secret = url.rstrip("/"), secret
def _request(self, method: str, table: str, params: dict[str, str] | None = None, body: dict | None = None,
returning: bool = False, upsert: bool = False):
query = "&".join(f"{k}={urllib.parse.quote(v, safe='(),.*:')}" for k, v in (params or {}).items())
headers = {"apikey": self.secret, "Authorization": f"Bearer {self.secret}", "User-Agent": "mathly2/1.0",
"Content-Type": "application/json"}
prefer = []
if upsert:
prefer.append("resolution=ignore-duplicates" if upsert == "ignore" else "resolution=merge-duplicates")
if returning:
prefer.append("return=representation")
if prefer:
headers["Prefer"] = ",".join(prefer)
data = json.dumps(body).encode() if body is not None else None
try:
_, raw = pooled.request(method, f"{self.url}/rest/v1/{table}" + (f"?{query}" if query else ""), headers, data, TIMEOUT)
except urllib.error.HTTPError as exc:
if exc.code in (400, 409): # a conflict, or one of the database's own refusals (see _why)
raise
raise StoreUnavailable(f"supabase answered {exc.code} for {method} {table}") from exc
except (urllib.error.URLError, TimeoutError) as exc:
raise StoreUnavailable(str(exc)) from exc
return json.loads(raw) if raw else None
@staticmethod
def _eq(value) -> str:
return f"eq.{value}"
@staticmethod
def _why(exc: urllib.error.HTTPError) -> str:
"""The database's own message for a refusal, or empty if it sent none. The body can be read only once, so it is kept."""
if not hasattr(exc, "_mathly_why"):
try:
exc._mathly_why = str(json.loads(exc.read() or b"{}").get("message", ""))
except (ValueError, AttributeError, OSError):
exc._mathly_why = ""
return exc._mathly_why
def _refused(self, exc: urllib.error.HTTPError, conflict: Exception) -> Exception:
"""A 409 is the conflict the caller expects (a name or an entry that already exists). Anything else the database refused
(a missing column because a migration was not applied, bad data) is a failure to report, not a name clash."""
if exc.code == 409:
return conflict
return StoreUnavailable(f"supabase refused ({exc.code}): {self._why(exc) or exc.reason}")
_CLASS_BASE = "id,name,created_by,teacher_email,lessons,ai_enabled,ai_daily_checks,created_at"
_has_subjects: bool | None = None # does m2_classes have the subjects column? (None: not asked yet)
def _class_select(self) -> str:
return self._CLASS_BASE + (",subjects" if self._has_subjects is not False else "")
def _class_rows(self, params: dict) -> list[dict]:
"""Classes with their subjects. Before the subjects migration is applied the column is missing: then every class studies math,
and the classes still load (the subject screens say the migration is needed)."""
try:
rows = self._request("GET", "m2_classes", {**params, "select": f"{self._class_select()},m2_class_members(count)"})
except urllib.error.HTTPError as exc:
if self._has_subjects is False or exc.code != 400 or "subjects" not in self._why(exc):
raise self._refused(exc, StoreUnavailable("classes")) from exc
self._has_subjects = False
rows = self._request("GET", "m2_classes", {**params, "select": f"{self._class_select()},m2_class_members(count)"})
else:
if self._has_subjects is None:
self._has_subjects = True
return [self._class_row(r) for r in rows]
def _class_row(self, row: dict) -> dict:
count = (row.get("m2_class_members") or [{"count": 0}])[0]["count"]
return {**{k: row[k] for k in self._CLASS_BASE.split(",")}, "subjects": list(row.get("subjects") or ["math"]), "students": count}
# --- classes ---
def create_class(self, created_by: str, name: str, *, lessons=(), teacher_email: str | None = None,
ai_enabled: bool = False, ai_daily_checks: int = 150, subjects=("math",)) -> dict:
body = {"created_by": created_by, "name": name, "lessons": list(lessons), "teacher_email": teacher_email,
"ai_enabled": ai_enabled, "ai_daily_checks": ai_daily_checks}
if list(subjects) != ["math"]: # math is the column's default, so a class works before the migration too
body["subjects"] = list(subjects)
try:
rows = self._request("POST", "m2_classes", body=body, returning=True)
except urllib.error.HTTPError as exc:
raise self._refused(exc, ClassExists(name)) from exc
return {**rows[0], "subjects": list(rows[0].get("subjects") or ["math"]), "students": 0}
def update_class(self, class_id: int, **fields) -> dict:
unknown = set(fields) - CLASS_FIELDS
if unknown:
raise ValueError(f"unknown class fields {sorted(unknown)}")
try:
rows = self._request("PATCH", "m2_classes", {"id": self._eq(class_id)}, fields, returning=True)
except urllib.error.HTTPError as exc:
if "teacher_is_student" in self._why(exc):
raise TeacherIsStudent(str(fields.get("teacher_email"))) from exc
raise self._refused(exc, ClassExists(str(fields.get("name")))) from exc
if not rows:
raise KeyError(class_id)
return self.get_class(class_id)
def delete_class(self, class_id: int) -> bool:
return bool(self._request("DELETE", "m2_classes", {"id": self._eq(class_id)}, returning=True))
def get_class(self, class_id: int) -> dict | None:
rows = self._class_rows({"id": self._eq(class_id)})
return rows[0] if rows else None
def classes(self) -> list[dict]:
return self._class_rows({"order": "id"})
def classes_taught(self, email: str) -> list[dict]:
return self._class_rows({"teacher_email": self._eq(email), "order": "id"})
# --- students ---
def members(self, class_id: int) -> list[dict]:
return self._request("GET", "m2_class_members", {"class_id": self._eq(class_id),
"select": "email,user_id,added_at", "order": "added_at,email"})
def classes_of_student(self, email: str) -> list[dict]:
def read(columns: str) -> list[dict]:
rows = self._request("GET", "m2_class_members", {"email": self._eq(email), "select": f"m2_classes({columns})"})
return [{**r["m2_classes"], "subjects": list(r["m2_classes"].get("subjects") or ["math"])} for r in rows]
base = "id,name,lessons,ai_enabled,ai_daily_checks"
if self._has_subjects is False:
return read(base)
try:
return read(base + ",subjects")
except urllib.error.HTTPError as exc: # before the subjects migration: every class studies math
if exc.code != 400 or "subjects" not in self._why(exc):
raise self._refused(exc, StoreUnavailable("classes")) from exc
self._has_subjects = False
return read(base)
def add_member(self, class_id: int, email: str) -> None:
try:
self._request("POST", "m2_class_members", body={"class_id": class_id, "email": email})
except urllib.error.HTTPError as exc:
if "teacher_is_student" in self._why(exc):
raise TeacherIsStudent(email) from exc
raise self._refused(exc, MemberExists(email)) from exc
def remove_member(self, class_id: int, email: str) -> bool:
rows = self._request("DELETE", "m2_class_members", {"class_id": self._eq(class_id), "email": self._eq(email)},
returning=True)
return bool(rows)
def link_user(self, email: str, user_id: str) -> None:
self._request("PATCH", "m2_class_members", {"email": self._eq(email), "user_id": "is.null"}, {"user_id": user_id})
def all_students(self) -> list[dict]:
rows = self._request("GET", "m2_class_members", {"select": "email,m2_classes(name)", "order": "email"})
found: dict[str, list[str]] = {}
for r in rows:
found.setdefault(r["email"], []).append(r["m2_classes"]["name"])
return [{"email": e, "classes": sorted(names)} for e, names in found.items()]
# --- roles ---
def roles_of(self, email: str) -> set[str]:
return {r["role"] for r in self._request("GET", "m2_roles", {"email": self._eq(email), "select": "role"})}
def give_role(self, email: str, role: str, granted_by: str) -> None:
self._request("POST", "m2_roles", {"on_conflict": "email,role"},
{"email": email, "role": role, "granted_by": granted_by}, upsert=True)
def take_role(self, email: str, role: str, settings_admins: int = 0) -> str:
"""One call in the database counts the admins and deletes together (function m2_take_role), so two admins
removing each other at once cannot leave none."""
return self._request("POST", "rpc/m2_take_role", body={"p_email": email, "p_role": role, "p_settings_admins": settings_admins})
def accounts(self) -> list[dict]:
roles: dict[str, list[str]] = {}
for r in self._request("GET", "m2_roles", {"select": "email,role", "order": "email"}):
roles.setdefault(r["email"], []).append(r["role"])
known = set(roles)
known |= {r["email"] for r in self._request("GET", "m2_class_members", {"select": "email"})}
known |= {r["teacher_email"] for r in self._request("GET", "m2_classes", {"select": "teacher_email", "teacher_email": "not.is.null"})}
return [{"email": e, "roles": sorted(roles.get(e, []))} for e in sorted(known)]
# --- skill learning ---
def get_progress(self, user_id: str, skill_id: str) -> dict | None:
rows = self._request("GET", "m2_skill_progress", {"user_id": self._eq(user_id), "skill_id": self._eq(skill_id),
"select": "*"})
return rows[0] if rows else None
def save_progress(self, user_id: str, skill_id: str, **fields) -> dict:
unknown = set(fields) - set(PROGRESS_FIELDS)
if unknown:
raise ValueError(f"unknown progress fields {sorted(unknown)}")
# merge-duplicates updates only the columns sent, so the other fields stay as they are
rows = self._request("POST", "m2_skill_progress", {"on_conflict": "user_id,skill_id"},
{"user_id": user_id, "skill_id": skill_id, **fields}, returning=True, upsert=True)
return rows[0]
def progress_of(self, user_id: str) -> list[dict]:
return self._request("GET", "m2_skill_progress", {"user_id": self._eq(user_id), "select": "*"})
# --- many students at once ---
@staticmethod
def _in(ids) -> str:
return "in.(" + ",".join(dict.fromkeys(ids)) + ")"
def progress_of_users(self, user_ids) -> dict[str, list[dict]]:
ids = list(dict.fromkeys(user_ids))
found: dict[str, list[dict]] = {u: [] for u in ids}
if ids:
for r in self._request("GET", "m2_skill_progress", {"user_id": self._in(ids), "select": "*"}):
found[str(r["user_id"])].append(r)
return found
def last_answered_users(self, user_ids) -> dict[str, dict[str, date]]:
ids = list(dict.fromkeys(user_ids))
found: dict[str, dict[str, date]] = {u: {} for u in ids}
if ids:
for r in self._request("GET", "m2_last_answered", {"user_id": self._in(ids), "select": "user_id,skill_id,day"}):
found[str(r["user_id"])][r["skill_id"]] = date.fromisoformat(r["day"])
return found
def attempts_of_users(self, user_ids, kind: str) -> dict[str, list[dict]]:
ids = list(dict.fromkeys(user_ids))
found: dict[str, list[dict]] = {u: [] for u in ids}
if ids:
for r in self._request("GET", "m2_attempts", {"user_id": self._in(ids), "kind": self._eq(kind), "select": "*", "order": "started_at"}):
found[str(r["user_id"])].append(_attempt_row(r))
return found
def add_answer(self, user_id: str, skill_id: str, question_key: str, correct: bool, counted: bool,
mistake_id: str | None, day: date) -> None:
self._request("POST", "m2_answers", body={"user_id": user_id, "skill_id": skill_id, "question_key": question_key,
"correct": correct, "counted": counted, "mistake_id": mistake_id,
"day": day.isoformat()})
def last_counted_answers(self, user_id: str, skill_id: str, n: int) -> list[dict]:
rows = self._request("GET", "m2_answers", {"user_id": self._eq(user_id), "skill_id": self._eq(skill_id),
"counted": "eq.true", "select": "question_key,correct,day", "order": "id.desc",
"limit": str(n)})
return [{**r, "day": date.fromisoformat(r["day"])} for r in reversed(rows)]
def answer_counts(self, user_id: str, skill_id: str) -> dict[str, int]:
rows = self._request("GET", "m2_answers", {"user_id": self._eq(user_id), "skill_id": self._eq(skill_id),
"select": "question_key"})
counts: dict[str, int] = {}
for r in rows:
counts[r["question_key"]] = counts.get(r["question_key"], 0) + 1
return counts
# --- real tests ---
def create_test(self, class_id: int, test_day: date, lessons: list[str], created_by: str, *, note: str = "", questions: int = 12,
minutes: int = 45, created_at: datetime | None = None, subject_id: str = "math") -> dict:
fields = {"lessons": list(lessons), "questions": questions, "minutes": minutes}
_check_test(fields)
body = {"class_id": class_id, "date": test_day.isoformat(), **fields, "note": note, "created_by": created_by}
if subject_id != "math": # math is the column's default (tests work before the migration too)
body["subject_id"] = subject_id
if created_at:
body["created_at"] = created_at.isoformat()
return _test_row(self._request("POST", "m2_tests", body=body, returning=True)[0])
def get_test(self, test_id: int) -> dict | None:
rows = self._request("GET", "m2_tests", {"id": self._eq(test_id), "select": "*"})
return _test_row(rows[0]) if rows else None
def update_test(self, test_id: int, **fields) -> dict:
unknown = set(fields) - TEST_FIELDS
if unknown:
raise ValueError(f"unknown test fields {sorted(unknown)}")
_check_test(fields)
body = {k: (v.isoformat() if isinstance(v, date) else v) for k, v in fields.items()}
rows = self._request("PATCH", "m2_tests", {"id": self._eq(test_id), "removed_at": "is.null"}, body, returning=True)
if not rows:
raise KeyError(test_id)
return _test_row(rows[0])
def remove_test(self, test_id: int, at: datetime | None = None) -> bool:
stamp = (at or datetime.now(timezone.utc)).isoformat()
return bool(self._request("PATCH", "m2_tests", {"id": self._eq(test_id), "removed_at": "is.null"}, {"removed_at": stamp}, returning=True))
def tests_of_classes(self, class_ids, include_removed: bool = False) -> list[dict]:
ids = sorted(set(class_ids))
if not ids:
return []
params = {"class_id": f"in.({','.join(str(i) for i in ids)})", "select": "*", "order": "date,id"}
if not include_removed:
params["removed_at"] = "is.null"
return [_test_row(r) for r in self._request("GET", "m2_tests", params)]
# --- pictures, assignments and submissions (written work) ---
def create_picture(self, owner_id: str, kind: str, path: str, content_type: str, size: int, created_at: datetime | None = None) -> dict:
if kind not in PICTURE_KINDS:
raise ValueError(f"unknown picture kind {kind!r}")
body = {"owner_id": owner_id, "kind": kind, "path": path, "content_type": content_type, "size": size}
if created_at:
body["created_at"] = created_at.isoformat()
return _stamped(self._request("POST", "m2_pictures", body=body, returning=True)[0], "created_at", "deleted_at")
def get_picture(self, picture_id: str) -> dict | None:
try:
rows = self._request("GET", "m2_pictures", {"id": self._eq(picture_id), "select": "*"})
except urllib.error.HTTPError as exc:
if exc.code == 400: # not a valid id
return None
raise
return _stamped(rows[0], "created_at", "deleted_at") if rows else None
def mark_picture_deleted(self, picture_id: str, at: datetime | None = None) -> bool:
stamp = (at or datetime.now(timezone.utc)).isoformat()
return bool(self._request("PATCH", "m2_pictures", {"id": self._eq(picture_id), "deleted_at": "is.null"}, {"deleted_at": stamp}, returning=True))
def pictures(self, kind: str | None = None, owner_id: str | None = None, include_deleted: bool = False) -> list[dict]:
params = {"select": "*", "order": "created_at"}
if kind:
params["kind"] = self._eq(kind)
if owner_id:
params["owner_id"] = self._eq(owner_id)
if not include_deleted:
params["deleted_at"] = "is.null"
return [_stamped(r, "created_at", "deleted_at") for r in self._request("GET", "m2_pictures", params)]
def picture_refs(self, picture_id: str) -> dict:
pid = str(picture_id)
return {"assignments": sorted(r["id"] for r in self._request("GET", "m2_assignments", {"pages": f"cs.{{{pid}}}", "select": "id"})),
"submissions": sorted(r["id"] for r in self._request("GET", "m2_submissions", {"pictures": f"cs.{{{pid}}}", "select": "id"})),
"requests": sorted(r["id"] for r in self._request("GET", "m2_photo_requests", {"picture_id": self._eq(pid), "select": "id"})),
"ai_requests": sorted(r["id"] for r in self._request("GET", "m2_ai_requests", {"picture_id": self._eq(pid), "select": "id"})),
"work": sorted(f"{r['ref']}:{r['question_key']}" for r in self._request("GET", "m2_work", {"picture_id": self._eq(pid), "select": "ref,question_key"}))}
def save_work(self, user_id: str, ref: str, question_key: str, skill_id: str, picture_id: str, at: datetime | None = None) -> str | None:
where = {"user_id": self._eq(user_id), "ref": self._eq(ref), "question_key": self._eq(question_key), "select": "picture_id"}
old = self._request("GET", "m2_work", where)
body = {"user_id": user_id, "ref": ref, "question_key": question_key, "skill_id": skill_id, "picture_id": str(picture_id),
"saved_at": (at or datetime.now(timezone.utc)).isoformat()}
self._request("POST", "m2_work", {"on_conflict": "user_id,ref,question_key"}, body, returning=False, upsert=True)
before = str(old[0]["picture_id"]) if old else None
return before if before and before != str(picture_id) else None
def work_of(self, user_id: str, ref: str) -> dict[str, str]:
rows = self._request("GET", "m2_work", {"user_id": self._eq(user_id), "ref": self._eq(ref), "select": "question_key,picture_id"})
return {r["question_key"]: str(r["picture_id"]) for r in rows}
def create_assignment(self, class_id: int, title: str, due_date: date, created_by: str, kind: str, *, note: str = "", pages=(),
quiz_lesson: str | None = None, test_id: int | None = None, exercise_number: int | None = None,
created_at: datetime | None = None) -> dict:
ok = {"pages": bool(pages), "quiz": bool(quiz_lesson), "exercise_test": test_id is not None and exercise_number in (1, 2)}
if not ok.get(kind) or not 1 <= len(title) <= 200:
raise ValueError("an assignment needs a title and its content")
body = {"class_id": class_id, "title": title, "note": note, "due_date": due_date.isoformat(), "kind": kind, "pages": [str(p) for p in pages],
"quiz_lesson": quiz_lesson, "test_id": test_id, "exercise_number": exercise_number, "created_by": created_by}
if created_at:
body["created_at"] = created_at.isoformat()
return _assignment_row(self._request("POST", "m2_assignments", body=body, returning=True)[0])
def get_assignment(self, assignment_id: int) -> dict | None:
rows = self._request("GET", "m2_assignments", {"id": self._eq(assignment_id), "select": "*"})
return _assignment_row(rows[0]) if rows else None
def assignments_of_classes(self, class_ids, include_removed: bool = False) -> list[dict]:
ids = sorted(set(class_ids))
if not ids:
return []
params = {"class_id": f"in.({','.join(str(i) for i in ids)})", "select": "*", "order": "due_date,id"}
if not include_removed:
params["removed_at"] = "is.null"
return [_assignment_row(r) for r in self._request("GET", "m2_assignments", params)]
def remove_assignment(self, assignment_id: int, at: datetime | None = None) -> bool:
stamp = (at or datetime.now(timezone.utc)).isoformat()
return bool(self._request("PATCH", "m2_assignments", {"id": self._eq(assignment_id), "removed_at": "is.null"}, {"removed_at": stamp}, returning=True))
def get_submission(self, assignment_id: int, user_id: str) -> dict | None:
rows = self._request("GET", "m2_submissions", {"assignment_id": self._eq(assignment_id), "user_id": self._eq(user_id), "select": "*"})
return _submission_row(rows[0]) if rows else None
def get_submission_by_id(self, submission_id: int) -> dict | None:
rows = self._request("GET", "m2_submissions", {"id": self._eq(submission_id), "select": "*"})
return _submission_row(rows[0]) if rows else None
def save_submission(self, assignment_id: int, user_id: str, **fields) -> dict:
unknown = set(fields) - SUBMISSION_FIELDS
if unknown:
raise ValueError(f"unknown submission fields {sorted(unknown)}")
body = {k: (v.isoformat() if isinstance(v, datetime) else ([str(x) for x in v] if k == "pictures" else v)) for k, v in fields.items()}
rows = self._request("POST", "m2_submissions", {"on_conflict": "assignment_id,user_id"},
{"assignment_id": assignment_id, "user_id": user_id, **body}, returning=True, upsert=True)
return _submission_row(rows[0])
def submissions_of_assignments(self, assignment_ids) -> list[dict]:
ids = sorted(set(assignment_ids))
if not ids:
return []
rows = self._request("GET", "m2_submissions", {"assignment_id": f"in.({','.join(str(i) for i in ids)})", "select": "*", "order": "handed_at,id"})
return [_submission_row(r) for r in rows]
def submissions_of_user(self, user_id: str) -> list[dict]:
return [_submission_row(r) for r in self._request("GET", "m2_submissions", {"user_id": self._eq(user_id), "select": "*"})]
# --- the review list (חזרה) ---
def add_review(self, user_id: str, items, at: datetime | None = None) -> int:
rows = []
for skill_id, key, source in items:
if source not in REVIEW_SOURCES:
raise ValueError(f"unknown review source {source!r}")
rows.append({"user_id": user_id, "question_key": key, "skill_id": skill_id, "source": source, **({"wrong_at": at.isoformat()} if at else {})})
if not rows:
return 0
got = self._request("POST", "m2_review", {"on_conflict": "user_id,question_key"}, rows, returning=True, upsert="ignore") # a row already there is kept
return len(got or [])
def review_of(self, user_id: str) -> list[dict]:
rows = self._request("GET", "m2_review", {"user_id": self._eq(user_id), "select": "*", "order": "wrong_at,question_key"})
return [_stamped(r, "wrong_at") for r in rows]
def remove_review(self, user_id: str, question_key: str) -> bool:
return bool(self._request("DELETE", "m2_review", {"user_id": self._eq(user_id), "question_key": self._eq(question_key)}, returning=True))
def add_review_done(self, user_id: str, question_key: str, day: date) -> None:
self._request("POST", "m2_review_done", body={"user_id": user_id, "question_key": question_key, "day": day.isoformat()})
# --- AI help: requests, allowances (budgets) and usage ---
def create_ai_request(self, kind: str, user_id: str, class_id: int | None, *, picture_id: str | None = None, assignment_id: int | None = None,
skill_id: str | None = None, snapshot: dict | None = None, created_at: datetime | None = None) -> dict:
if kind not in ("photo", "check", "again") or (kind == "photo" and not picture_id) or (kind != "photo" and assignment_id is None):
raise ValueError("an AI request needs its picture (a photo) or its assignment (a check, a look again)")
body = {"kind": kind, "user_id": user_id, "class_id": class_id, "picture_id": picture_id, "assignment_id": assignment_id,
"skill_id": skill_id, "snapshot": snapshot}
if created_at:
body["created_at"] = created_at.isoformat()
return _ai_row(self._request("POST", "m2_ai_requests", body=body, returning=True)[0])
def get_ai_request(self, request_id: str) -> dict | None:
try:
rows = self._request("GET", "m2_ai_requests", {"id": self._eq(request_id), "select": "*"})
except urllib.error.HTTPError as exc:
if exc.code == 400:
return None
raise
return _ai_row(rows[0]) if rows else None
def save_ai_request(self, request_id: str, **fields) -> dict:
unknown = set(fields) - AI_FIELDS
if unknown:
raise ValueError(f"unknown AI request fields {sorted(unknown)}")
body = {k: (v.isoformat() if isinstance(v, datetime) else v) for k, v in fields.items()}
rows = self._request("PATCH", "m2_ai_requests", {"id": self._eq(request_id)}, body, returning=True)
if not rows:
raise KeyError(request_id)
return _ai_row(rows[0])
def ai_requests(self, *, user_id: str | None = None, class_ids=None, states=None, kinds=None, assignment_id: int | None = None,
picture_id: str | None = None) -> list[dict]:
params = {"select": "*", "order": "created_at,id"}
if user_id is not None:
params["user_id"] = self._eq(user_id)
if class_ids is not None:
ids = sorted(set(class_ids))
if not ids:
return []
params["class_id"] = f"in.({','.join(str(i) for i in ids)})"
if states is not None:
params["state"] = f"in.({','.join(sorted(set(states)))})"
if kinds is not None:
params["kind"] = f"in.({','.join(sorted(set(kinds)))})"
if assignment_id is not None:
params["assignment_id"] = self._eq(assignment_id)
if picture_id is not None:
params["picture_id"] = self._eq(picture_id)
return [_ai_row(r) for r in self._request("GET", "m2_ai_requests", params)]
def start_ai_request(self, request_id: str, sent_by: str, at: datetime) -> dict | None:
rows = self._request("PATCH", "m2_ai_requests", {"id": self._eq(request_id), "state": "in.(waiting,failed)"},
{"state": "processing", "failure": None, "sent_by": sent_by, "sent_at": at.isoformat()}, returning=True)
return _ai_row(rows[0]) if rows else None
def requeue_processing_ai_requests(self) -> int:
return len(self._request("PATCH", "m2_ai_requests", {"state": "eq.processing"}, {"state": "waiting"}, returning=True) or [])
def take_ai_allowance(self, scope: str, day: date, n: int, cap: int) -> bool:
return bool(self._request("POST", "rpc/m2_ai_take", body={"p_scope": scope, "p_day": day.isoformat(), "p_n": n, "p_cap": cap}))
def give_back_ai_allowance(self, scope: str, day: date, n: int) -> None:
self._request("POST", "rpc/m2_ai_give_back", body={"p_scope": scope, "p_day": day.isoformat(), "p_n": n})
def ai_allowance_used(self, scope: str, day: date) -> int:
rows = self._request("GET", "m2_ai_allowance", {"scope": self._eq(scope), "day": self._eq(day.isoformat()), "select": "used"})
return rows[0]["used"] if rows else 0
def add_ai_usage(self, row: dict) -> None:
self._request("POST", "m2_ai_usage", body={k: v for k, v in row.items() if k != "created_at"})
def ai_usage_rows(self) -> list[dict]:
return self._request("GET", "m2_ai_usage", {"select": "*", "order": "created_at"})
# --- photo requests and push subscriptions ---
def create_photo_request(self, user_id: str, assignment_id: int, device: str, code: str, created_at: datetime, expires_at: datetime) -> dict:
self._request("PATCH", "m2_photo_requests", {"user_id": self._eq(user_id), "assignment_id": self._eq(assignment_id), "state": "eq.waiting"},
{"state": "superseded"})
rows = self._request("POST", "m2_photo_requests", body={"user_id": user_id, "assignment_id": assignment_id, "device": device, "code": code,
"created_at": created_at.isoformat(), "expires_at": expires_at.isoformat()}, returning=True)
return _stamped(rows[0], "created_at", "expires_at")
def get_photo_request(self, request_id: str) -> dict | None:
try:
rows = self._request("GET", "m2_photo_requests", {"id": self._eq(request_id), "select": "*"})
except urllib.error.HTTPError as exc:
if exc.code == 400:
return None
raise
return _stamped(rows[0], "created_at", "expires_at") if rows else None
def save_photo_request(self, request_id: str, **fields) -> dict:
unknown = set(fields) - REQUEST_FIELDS
if unknown:
raise ValueError(f"unknown photo request fields {sorted(unknown)}")
rows = self._request("PATCH", "m2_photo_requests", {"id": self._eq(request_id)}, fields, returning=True)
if not rows:
raise KeyError(request_id)
return _stamped(rows[0], "created_at", "expires_at")
def open_photo_requests(self, user_id: str) -> list[dict]:
rows = self._request("GET", "m2_photo_requests", {"user_id": self._eq(user_id), "state": "in.(waiting,uploading)", "select": "*", "order": "created_at.desc"})
return [_stamped(r, "created_at", "expires_at") for r in rows]
def add_push_subscription(self, user_id: str, endpoint: str, p256dh: str, auth: str, key_id: str) -> dict:
rows = self._request("POST", "m2_push_subscriptions", {"on_conflict": "endpoint"},
{"user_id": user_id, "endpoint": endpoint, "p256dh": p256dh, "auth": auth, "key_id": key_id}, returning=True, upsert=True)
return _stamped(rows[0], "created_at")
def push_subscriptions_of(self, user_id: str) -> list[dict]:
return [_stamped(r, "created_at") for r in self._request("GET", "m2_push_subscriptions", {"user_id": self._eq(user_id), "select": "*"})]
def delete_push_subscription(self, endpoint: str) -> bool:
return bool(self._request("DELETE", "m2_push_subscriptions", {"endpoint": self._eq(endpoint)}, returning=True))
def drop_push_subscriptions_not_of(self, key_id: str) -> int:
return len(self._request("DELETE", "m2_push_subscriptions", {"key_id": f"neq.{key_id}"}, returning=True) or [])
# --- timed attempts ---
def create_attempt(self, user_id: str, kind: str, scope: str, started_at: datetime, due_at: datetime, questions: list[dict]) -> dict:
rows = self._request("POST", "m2_attempts", body={"user_id": user_id, "kind": kind, "scope": scope, "started_at": started_at.isoformat(),
"due_at": due_at.isoformat(), "questions": questions}, returning=True)
return _attempt_row(rows[0])
def get_attempt(self, attempt_id: str) -> dict | None:
try:
rows = self._request("GET", "m2_attempts", {"id": self._eq(attempt_id), "select": "*"})
except urllib.error.HTTPError as exc:
if exc.code == 400: # not a valid id
return None
raise
return _attempt_row(rows[0]) if rows else None
def save_attempt(self, attempt_id: str, **fields) -> dict:
unknown = set(fields) - ATTEMPT_FIELDS
if unknown:
raise ValueError(f"unknown attempt fields {sorted(unknown)}")
body = {k: (v.isoformat() if isinstance(v, datetime) else v) for k, v in fields.items()}
rows = self._request("PATCH", "m2_attempts", {"id": self._eq(attempt_id)}, body, returning=True)
if not rows:
raise KeyError(attempt_id)
return _attempt_row(rows[0])
def save_open_answers(self, attempt_id: str, answers: dict) -> dict | None:
rows = self._request("PATCH", "m2_attempts", {"id": self._eq(attempt_id), "submitted_at": "is.null"}, {"answers": answers}, returning=True)
if rows:
return _attempt_row(rows[0])
if self.get_attempt(attempt_id) is None:
raise KeyError(attempt_id)
return None
def attempts_of(self, user_id: str, kind: str, scope: str | None = None) -> list[dict]:
params = {"user_id": self._eq(user_id), "kind": self._eq(kind), "select": "*", "order": "started_at"}
if scope is not None:
params["scope"] = self._eq(scope)
return [_attempt_row(r) for r in self._request("GET", "m2_attempts", params)]
def comets_of(self, user_id: str) -> set[str]:
rows = self._request("GET", "m2_comets", {"user_id": self._eq(user_id), "select": "comet_id"})
return {r["comet_id"] for r in rows}
def add_comet(self, user_id: str, comet_id: str) -> None:
self._request("POST", "m2_comets", {"on_conflict": "user_id,comet_id"}, {"user_id": user_id, "comet_id": comet_id}, upsert=True)
def reward_state(self, user_id: str) -> dict | None:
rows = self._request("GET", "m2_reward_state", {"user_id": self._eq(user_id), "select": "*"})
if not rows:
return None
return self.reward_state_row(rows[0])
def save_reward_state(self, user_id: str, **fields) -> dict:
body = {k: v.isoformat() if isinstance(v, date) else v for k, v in fields.items()}
rows = self._request("POST", "m2_reward_state", {"on_conflict": "user_id"}, {"user_id": user_id, **body}, returning=True, upsert=True)
return self.reward_state_row(rows[0])
@staticmethod
def reward_state_row(row: dict) -> dict:
return {**row, **{k: date.fromisoformat(row[k]) if row.get(k) else None for k in ("last_comet_day", "last_catch_day")}}
# --- the curriculum ---
def curriculum(self) -> dict[str, date]:
return {r["lesson_id"]: date.fromisoformat(str(r["due_date"])[:10]) for r in self._request("GET", "m2_curriculum", {"select": "lesson_id,due_date"})}
def set_lesson_dates(self, dates: dict[str, date | None]) -> None:
cleared = [lesson for lesson, day in dates.items() if day is None]
if cleared:
self._request("DELETE", "m2_curriculum", {"lesson_id": "in.(" + ",".join(f'"{x}"' for x in cleared) + ")"})
rows = [{"lesson_id": lesson, "due_date": day.isoformat()} for lesson, day in dates.items() if day is not None]
if rows:
self._request("POST", "m2_curriculum", {"on_conflict": "lesson_id"}, rows, upsert=True)
def last_answered(self, user_id: str) -> dict[str, date]:
rows = self._request("GET", "m2_last_answered", {"user_id": self._eq(user_id), "select": "skill_id,day"})
return {r["skill_id"]: date.fromisoformat(r["day"]) for r in rows}
# --- subjects and the lessons kept in the database ---
def subjects(self) -> list[dict]:
rows = self._request("GET", "m2_subjects", {"select": "*", "order": "position,created_at"})
return [{**r, "created_at": _when(r.get("created_at"))} for r in rows]
def create_subject(self, subject_id: str, name: str, created_by: str, *, icon: str = "∑") -> dict:
position = max((s["position"] for s in self.subjects()), default=0) + 1
body = {"id": subject_id, "name": name, "icon": icon, "created_by": created_by, "position": position}
try:
rows = self._request("POST", "m2_subjects", body=body, returning=True)
except urllib.error.HTTPError as exc:
raise self._refused(exc, SubjectExists(name)) from exc
return {**rows[0], "created_at": _when(rows[0].get("created_at"))}
def update_subject(self, subject_id: str, **fields) -> dict:
unknown = set(fields) - SUBJECT_FIELDS
if unknown:
raise ValueError(f"unknown subject fields {sorted(unknown)}")
try:
rows = self._request("PATCH", "m2_subjects", {"id": self._eq(subject_id)}, fields, returning=True)
except urllib.error.HTTPError as exc:
raise self._refused(exc, SubjectExists(str(fields.get("name")))) from exc
if not rows:
raise KeyError(subject_id)
return {**rows[0], "created_at": _when(rows[0].get("created_at"))}
def delete_subject(self, subject_id: str) -> bool:
return bool(self._request("DELETE", "m2_subjects", {"id": self._eq(subject_id)}, returning=True))
def create_lesson(self, subject_id: str, created_by: str, draft: dict, *, source: str = "json", status: str = "ready",
files: list | None = None) -> dict:
last = self._request("GET", "m2_lessons", {"subject_id": self._eq(subject_id), "select": "position", "order": "position.desc", "limit": "1"})
body = {"subject_id": subject_id, "created_by": created_by, "draft": draft, "source": source, "status": status,
"files": list(files or []), "position": (last[0]["position"] if last else 0) + 1}
try:
rows = self._request("POST", "m2_lessons", body=body, returning=True)
except urllib.error.HTTPError as exc:
raise self._refused(exc, KeyError(subject_id)) from exc
return _lesson_row(rows[0])
def get_lesson(self, lesson_id: int) -> dict | None:
rows = self._request("GET", "m2_lessons", {"id": self._eq(lesson_id), "select": "*"})
return _lesson_row(rows[0]) if rows else None
def lessons_of(self, subject_id: str | None = None) -> list[dict]:
params = {"select": "*", "order": "subject_id,position,id"}
if subject_id is not None:
params["subject_id"] = self._eq(subject_id)
return [_lesson_row(r) for r in self._request("GET", "m2_lessons", params)]
def published_lessons(self) -> list[dict]:
rows = self._request("GET", "m2_lessons", {"published": "not.is.null", "select": "*", "order": "subject_id,position,id"})
return [_lesson_row(r) for r in rows]
def save_lesson(self, lesson_id: int, **fields) -> dict:
unknown = set(fields) - LESSON_FIELDS
if unknown:
raise ValueError(f"unknown lesson fields {sorted(unknown)}")
body = {k: (v.isoformat() if isinstance(v, datetime) else v) for k, v in fields.items()}
body["updated_at"] = datetime.now(timezone.utc).isoformat()
rows = self._request("PATCH", "m2_lessons", {"id": self._eq(lesson_id)}, body, returning=True)
if not rows:
raise KeyError(lesson_id)
return _lesson_row(rows[0])
def delete_lesson(self, lesson_id: int) -> bool:
return bool(self._request("DELETE", "m2_lessons", {"id": self._eq(lesson_id)}, returning=True))
# --- the developers' prompter texts ---
_PROMPTER = "id,user_id,title,body,outline,updated_at"
@staticmethod
def _prompter_row(row: dict) -> dict:
return {**row, "updated_at": _when(row.get("updated_at"))}
def prompter_texts(self, user_id: str) -> list[dict]:
rows = self._request("GET", "m2_prompter_texts", {"user_id": self._eq(user_id), "select": self._PROMPTER, "order": "updated_at.desc", "limit": "200"})
return [self._prompter_row(r) for r in rows]
def get_prompter_text(self, user_id: str, text_id: str) -> dict | None:
rows = self._request("GET", "m2_prompter_texts", {"id": self._eq(text_id), "user_id": self._eq(user_id), "select": self._PROMPTER})
return self._prompter_row(rows[0]) if rows else None
def add_prompter_text(self, user_id: str, title: str, body: str) -> dict:
rows = self._request("POST", "m2_prompter_texts", body={"user_id": user_id, "title": title, "body": body,
"updated_at": datetime.now(timezone.utc).isoformat()}, returning=True)
return self._prompter_row(rows[0])
def save_prompter_text(self, user_id: str, text_id: str, *, touch: bool = True, **fields) -> dict | None:
unknown = set(fields) - PROMPTER_FIELDS
if unknown:
raise ValueError(f"unknown prompter fields {sorted(unknown)}")
body = dict(fields, **({"updated_at": datetime.now(timezone.utc).isoformat()} if touch else {}))
rows = self._request("PATCH", "m2_prompter_texts", {"id": self._eq(text_id), "user_id": self._eq(user_id)}, body, returning=True)
return self._prompter_row(rows[0]) if rows else None
def delete_prompter_text(self, user_id: str, text_id: str) -> bool:
return bool(self._request("DELETE", "m2_prompter_texts", {"id": self._eq(text_id), "user_id": self._eq(user_id)}, returning=True))
ACTIVITY_PAGE = 1000
def activity_since(self, first: date) -> list[dict]:
"""Read in pages (the API returns at most a page at a time). Each page starts after the last (day, user_id) read, not at an
offset, so a row added meanwhile cannot shift a row into two pages or out of both."""
rows: list[dict] = []
params = {"day": f"gte.{first.isoformat()}", "select": "day,user_id,answers,attempts", "order": "day,user_id",
"limit": str(self.ACTIVITY_PAGE)}
while True:
page = self._request("GET", "m2_activity", params)
rows += page
if len(page) < self.ACTIVITY_PAGE:
break
last = page[-1]
params = {**params, "or": f"(day.gt.{last['day']},and(day.eq.{last['day']},user_id.gt.{last['user_id']}))"}
return [{**r, "day": date.fromisoformat(r["day"])} for r in rows]
def daily_correct(self, user_id: str) -> dict[date, int]:
rows = self._request("GET", "m2_daily_correct", {"user_id": self._eq(user_id), "select": "day,correct"})
return {date.fromisoformat(r["day"]): r["correct"] for r in rows}
def answers_of(self, user_id: str, skill_id: str) -> list[dict]:
rows = self._request("GET", "m2_answers", {"user_id": self._eq(user_id), "skill_id": self._eq(skill_id),
"select": "question_key,correct,counted,day,mistake_id", "order": "id"})
return [{**r, "day": date.fromisoformat(r["day"])} for r in rows]
def recent_skills(self, user_id: str, n: int = 20) -> list[str]:
rows = self._request("GET", "m2_answers", {"user_id": self._eq(user_id), "select": "skill_id", "order": "id.desc",
"limit": str(max(200, n * 20))})
found: list[str] = []
for r in rows:
if r["skill_id"] not in found:
found.append(r["skill_id"])
if len(found) == n:
break
return found
def solved_keys(self, user_id: str, skill_id: str) -> set[str]:
rows = self._request("GET", "m2_answers", {"user_id": self._eq(user_id), "skill_id": self._eq(skill_id),
"correct": "eq.true", "select": "question_key"})
return {r["question_key"] for r in rows}
"""Where classes, students, roles and learning progress are kept. Two stores with the same methods:
InMemoryStore (tests and local development, lost on restart) and SupabaseStore (PostgREST with the secret key)."""
from __future__ import annotations
import copy
import re
import itertools
import json
import threading
import urllib.error
import urllib.parse
import uuid
from datetime import date, datetime, timezone
from . import pooled
TIMEOUT = 10
class StoreUnavailable(RuntimeError):
"""The database could not be reached or refused the request."""
class ClassExists(ValueError):
"""The teacher already has a class with this name."""
class MemberExists(ValueError):
"""The email already belongs to this class."""
class SubjectExists(ValueError):
"""A subject with this name (ignoring case) already exists."""
class TeacherIsStudent(ValueError):
"""A class's teacher cannot also be one of its students."""
def _now() -> str:
return datetime.now(timezone.utc).isoformat()
# What the server keeps about one student and one skill (table m2_skill_progress). `mastered_at` is set once.
PROGRESS_FIELDS = ("current_key", "current_helped", "started_at", "mastered_at", "round_start")
PROGRESS_DEFAULTS = {"current_key": None, "current_helped": False, "started_at": None, "mastered_at": None, "round_start": None}
ATTEMPT_FIELDS = {"answers", "submitted_at", "timed_out", "grade", "credit", "checked_at", "checked_by", "review"}
def _copy_attempt(row: dict) -> dict:
return {**row, "questions": [dict(q) for q in row["questions"]], "answers": dict(row["answers"]), "credit": dict(row.get("credit") or {}),
"review": copy.deepcopy(row.get("review"))}
def _when(value):
if value is None or isinstance(value, datetime):
return value
# Python 3.10 reads a fraction of exactly 3 or 6 digits only, and Postgres drops trailing zeros ("...57.63003+00:00")
text = re.sub(r"\.(\d+)", lambda m: "." + m.group(1)[:6].ljust(6, "0"), str(value).replace("Z", "+00:00"), count=1)
return datetime.fromisoformat(text)
def _attempt_row(row: dict) -> dict:
"""A database row as the in-memory store gives it: real datetimes, plain lists and dicts."""
return {**row, "started_at": _when(row["started_at"]), "due_at": _when(row["due_at"]), "submitted_at": _when(row.get("submitted_at")),
"questions": list(row.get("questions") or []), "answers": dict(row.get("answers") or {}), "credit": dict(row.get("credit") or {}),
"checked_at": _when(row.get("checked_at")), "checked_by": row.get("checked_by"), "review": row.get("review")}
TEST_FIELDS = {"date", "lessons", "note", "questions", "minutes", "subject_id"}
CLASS_FIELDS = {"name", "lessons", "teacher_email", "ai_enabled", "ai_daily_checks", "subjects"}
SUBJECT_FIELDS = {"name", "icon", "position"}
LESSON_FIELDS = {"subject_id", "position", "draft", "published", "status", "failure", "files", "published_at"}
AI_FEATURES = ("photo", "check", "worksheet", "prompter") # m2_ai_usage.feature (the database checks the same list)
PROMPTER_FIELDS = {"title", "body", "outline"}
MATH = {"id": "math", "name": "מתמטיקה", "icon": "∑", "position": 0, "created_by": "setup"}
def _lesson_row(row: dict) -> dict:
"""A lesson row as both stores give it: real datetimes, the draft and published documents as plain dicts (published may be None)."""
return {**row, "draft": dict(row.get("draft") or {}), "published": dict(row["published"]) if row.get("published") else None,
"files": list(row.get("files") or []), "created_at": _when(row.get("created_at")), "updated_at": _when(row.get("updated_at")),
"published_at": _when(row.get("published_at"))}
def _check_test(fields: dict) -> None:
"""The limits the database enforces too: 1-40 questions, 5-180 minutes, at least one lesson."""
if "questions" in fields and not 1 <= fields["questions"] <= 40:
raise ValueError("questions must be 1-40")
if "minutes" in fields and not 5 <= fields["minutes"] <= 180:
raise ValueError("minutes must be 5-180")
if "lessons" in fields and not fields["lessons"]:
raise ValueError("a test covers at least one lesson")
def _test_row(row: dict) -> dict:
"""A database row as the in-memory store gives it: a real date and datetimes, a plain list."""
day = row["date"] if isinstance(row["date"], date) else date.fromisoformat(str(row["date"])[:10])