From 429adfc3763877f053be950b06e1fe34485b0bf8 Mon Sep 17 00:00:00 2001 From: Ross Date: Sun, 4 Jan 2026 13:15:15 +0000 Subject: [PATCH] Optimize exam retrieval by prefetching related data to reduce N+1 queries and enhance access checks for user permissions --- generic/views.py | 104 +++++++++++++++++++++++++++++++---------------- 1 file changed, 70 insertions(+), 34 deletions(-) diff --git a/generic/views.py b/generic/views.py index cdceb9ea..047d445c 100644 --- a/generic/views.py +++ b/generic/views.py @@ -3044,7 +3044,12 @@ class ExamViews(View, LoginRequiredMixin): cid: int | None = None, passcode: str | None = None, ): - exams = self.Exam.objects.filter(archive=False) + # Prefetch related relations used during the active_exams check to + # avoid N+1 queries when iterating many exams. + exams = ( + self.Exam.objects.filter(archive=False) + .prefetch_related("valid_cid_users", "valid_user_users", "exam_questions") + ) active_exams = {"exams": []} @@ -3056,45 +3061,76 @@ class ExamViews(View, LoginRequiredMixin): return JsonResponse({"status": "invalid"}, status=401) exam: ExamBase + # Prepare cid/user quick lookup when cid/user provided to avoid DB hits + user_id = request.user.pk if hasattr(request, "user") and request.user.is_authenticated else None + for exam in exams: - if exam.active or self.check_user_access(request.user, exam.pk): - if exam.exam_mode and not exam.check_cid_user( - cid, passcode, user=request.user, user_id=request.user.pk - ): + # Skip exams the user cannot see unless it's active + if not (exam.active or self.check_user_access(request.user, exam.pk)): + continue + + # If exam is in exam_mode, ensure the cid/user is allowed. Use the + # prefetched relations to perform in-memory checks rather than extra + # DB queries via `check_cid_user` which issues its own queries. + if exam.exam_mode: + allowed = False + # superusers always allowed + if hasattr(request, "user") and getattr(request.user, "is_superuser", False): + allowed = True + + # open access allows any logged-in user + if not allowed and exam.exam_open_access and user_id is not None: + allowed = True + + # check valid_user_users (prefetched) + if not allowed and user_id is not None: + try: + valid_user_ids = {u.pk for u in exam.valid_user_users.all()} + if user_id in valid_user_ids: + allowed = True + except Exception: + allowed = False + + # check CID users when cid provided + if not allowed and cid is not None: + try: + for cu in exam.valid_cid_users.all(): + if getattr(cu, "cid", None) == cid and getattr(cu, "passcode", None) == passcode: + allowed = True + break + except Exception: + allowed = False + + if not allowed: continue - if exam.json_creation_time: - creation_time = exam.json_creation_time.isoformat() - else: - creation_time = "None" + creation_time = exam.json_creation_time.isoformat() if exam.json_creation_time else "None" - url = request.build_absolute_uri( - exam.get_json_url(cid=cid, passcode=passcode) - ) - # hacky - if not based: - url = url + "/unbased" + url = request.build_absolute_uri(exam.get_json_url(cid=cid, passcode=passcode)) + if not based: + url = url + "/unbased" - obj = { - "name": exam.get_exam_name(), - "url": url, - "type": self.question_type, - "eid": "{}/{}".format(self.question_type, exam.pk), - "json_creation_time": creation_time, - "exam_json_id": exam.exam_json_id, - "exam_active": exam.active, - "exam_mode": exam.exam_mode, - } + obj = { + "name": exam.get_exam_name(), + "url": url, + "type": self.question_type, + "eid": "{}/{}".format(self.question_type, exam.pk), + "json_creation_time": creation_time, + "exam_json_id": exam.exam_json_id, + "exam_active": exam.active, + "exam_mode": exam.exam_mode, + } - if self.question_type == "long": - h = {} - for question in exam.exam_questions.all(): - # Generate json if needed - if question.json_creation_time is None: - question.get_question_json() - h[question.pk] = question.question_json_id - obj["multi_question_json"] = h - active_exams["exams"].append(obj) + # Avoid generating question JSON in this endpoint; return existing ids + # to keep CPU usage low. Generation should be handled asynchronously + # or on-demand by the request that needs it. + if self.question_type == "long": + h = {} + for question in exam.exam_questions.all(): + h[question.pk] = getattr(question, "question_json_id", None) + obj["multi_question_json"] = h + + active_exams["exams"].append(obj) if json is False: return active_exams["exams"]