Commit a4260231 by Afzal Wali Committed by Muhammad Shoaib

refactored the model code into a new custom model manager for Attempts.

implemented the search query for attempts.
parent 34a613d1
...@@ -160,7 +160,7 @@ def get_exam_attempt(exam_id, user_id): ...@@ -160,7 +160,7 @@ def get_exam_attempt(exam_id, user_id):
""" """
Return an existing exam attempt for the given student Return an existing exam attempt for the given student
""" """
exam_attempt_obj = ProctoredExamStudentAttempt.get_exam_attempt(exam_id, user_id) exam_attempt_obj = ProctoredExamStudentAttempt.objects.get_exam_attempt(exam_id, user_id)
serialized_attempt_obj = ProctoredExamStudentAttemptSerializer(exam_attempt_obj) serialized_attempt_obj = ProctoredExamStudentAttemptSerializer(exam_attempt_obj)
return serialized_attempt_obj.data if exam_attempt_obj else None return serialized_attempt_obj.data if exam_attempt_obj else None
...@@ -169,7 +169,7 @@ def get_exam_attempt_by_id(attempt_id): ...@@ -169,7 +169,7 @@ def get_exam_attempt_by_id(attempt_id):
""" """
Return an existing exam attempt for the given student Return an existing exam attempt for the given student
""" """
exam_attempt_obj = ProctoredExamStudentAttempt.get_exam_attempt_by_id(attempt_id) exam_attempt_obj = ProctoredExamStudentAttempt.objects.get_exam_attempt_by_id(attempt_id)
serialized_attempt_obj = ProctoredExamStudentAttemptSerializer(exam_attempt_obj) serialized_attempt_obj = ProctoredExamStudentAttemptSerializer(exam_attempt_obj)
return serialized_attempt_obj.data if exam_attempt_obj else None return serialized_attempt_obj.data if exam_attempt_obj else None
...@@ -180,7 +180,7 @@ def create_exam_attempt(exam_id, user_id, taking_as_proctored=False): ...@@ -180,7 +180,7 @@ def create_exam_attempt(exam_id, user_id, taking_as_proctored=False):
one exam_attempt per user per exam. Multiple attempts by user will be archived one exam_attempt per user per exam. Multiple attempts by user will be archived
in a separate table in a separate table
""" """
if ProctoredExamStudentAttempt.get_exam_attempt(exam_id, user_id): if ProctoredExamStudentAttempt.objects.get_exam_attempt(exam_id, user_id):
err_msg = ( err_msg = (
'Cannot create new exam attempt for exam_id = {exam_id} and ' 'Cannot create new exam attempt for exam_id = {exam_id} and '
'user_id = {user_id} because it already exists!' 'user_id = {user_id} because it already exists!'
...@@ -245,7 +245,7 @@ def start_exam_attempt(exam_id, user_id): ...@@ -245,7 +245,7 @@ def start_exam_attempt(exam_id, user_id):
Returns: exam_attempt_id (PK) Returns: exam_attempt_id (PK)
""" """
existing_attempt = ProctoredExamStudentAttempt.get_exam_attempt(exam_id, user_id) existing_attempt = ProctoredExamStudentAttempt.objects.get_exam_attempt(exam_id, user_id)
if not existing_attempt: if not existing_attempt:
err_msg = ( err_msg = (
...@@ -264,7 +264,7 @@ def start_exam_attempt_by_code(attempt_code): ...@@ -264,7 +264,7 @@ def start_exam_attempt_by_code(attempt_code):
an attempt code an attempt code
""" """
existing_attempt = ProctoredExamStudentAttempt.get_exam_attempt_by_code(attempt_code) existing_attempt = ProctoredExamStudentAttempt.objects.get_exam_attempt_by_code(attempt_code)
if not existing_attempt: if not existing_attempt:
err_msg = ( err_msg = (
...@@ -298,7 +298,7 @@ def stop_exam_attempt(exam_id, user_id): ...@@ -298,7 +298,7 @@ def stop_exam_attempt(exam_id, user_id):
""" """
Marks the exam attempt as completed (sets the completed_at field and updates the record) Marks the exam attempt as completed (sets the completed_at field and updates the record)
""" """
exam_attempt_obj = ProctoredExamStudentAttempt.get_exam_attempt(exam_id, user_id) exam_attempt_obj = ProctoredExamStudentAttempt.objects.get_exam_attempt(exam_id, user_id)
if exam_attempt_obj is None: if exam_attempt_obj is None:
raise StudentExamAttemptDoesNotExistsException('Error. Trying to stop an exam that is not in progress.') raise StudentExamAttemptDoesNotExistsException('Error. Trying to stop an exam that is not in progress.')
else: else:
...@@ -312,7 +312,7 @@ def remove_exam_attempt_by_id(attempt_id): ...@@ -312,7 +312,7 @@ def remove_exam_attempt_by_id(attempt_id):
Removes an exam attempt given the attempt id. Removes an exam attempt given the attempt id.
""" """
existing_attempt = ProctoredExamStudentAttempt.get_exam_attempt_by_id(attempt_id) existing_attempt = ProctoredExamStudentAttempt.objects.get_exam_attempt_by_id(attempt_id)
if not existing_attempt: if not existing_attempt:
err_msg = ( err_msg = (
...@@ -358,7 +358,7 @@ def get_all_exam_attempts(course_id): ...@@ -358,7 +358,7 @@ def get_all_exam_attempts(course_id):
""" """
Returns all the exam attempts for the course id. Returns all the exam attempts for the course id.
""" """
exam_attempts = ProctoredExamStudentAttempt.get_all_exam_attempts(course_id) exam_attempts = ProctoredExamStudentAttempt.objects.get_all_exam_attempts(course_id)
return [ProctoredExamStudentAttemptSerializer(active_exam).data for active_exam in exam_attempts] return [ProctoredExamStudentAttemptSerializer(active_exam).data for active_exam in exam_attempts]
...@@ -366,7 +366,7 @@ def get_filtered_exam_attempts(course_id, search_by): ...@@ -366,7 +366,7 @@ def get_filtered_exam_attempts(course_id, search_by):
""" """
returns all exam attempts for a course id filtered by the search_by string in user names and emails. returns all exam attempts for a course id filtered by the search_by string in user names and emails.
""" """
exam_attempts = ProctoredExamStudentAttempt.get_all_exam_attempts(course_id) exam_attempts = ProctoredExamStudentAttempt.objects.get_filtered_exam_attempts(course_id, search_by)
return [ProctoredExamStudentAttemptSerializer(active_exam).data for active_exam in exam_attempts] return [ProctoredExamStudentAttemptSerializer(active_exam).data for active_exam in exam_attempts]
...@@ -391,7 +391,7 @@ def get_active_exams_for_user(user_id, course_id=None): ...@@ -391,7 +391,7 @@ def get_active_exams_for_user(user_id, course_id=None):
""" """
result = [] result = []
student_active_exams = ProctoredExamStudentAttempt.get_active_student_attempts(user_id, course_id) student_active_exams = ProctoredExamStudentAttempt.objects.get_active_student_attempts(user_id, course_id)
for active_exam in student_active_exams: for active_exam in student_active_exams:
# convert the django orm objects # convert the django orm objects
# into the serialized form. # into the serialized form.
......
...@@ -11,7 +11,7 @@ from model_utils.models import TimeStampedModel ...@@ -11,7 +11,7 @@ from model_utils.models import TimeStampedModel
from django.contrib.auth.models import User from django.contrib.auth.models import User
from edx_proctoring.exceptions import UserNotFoundException from edx_proctoring.exceptions import UserNotFoundException
from django.db.models.base import ObjectDoesNotExist
class ProctoredExam(TimeStampedModel): class ProctoredExam(TimeStampedModel):
""" """
...@@ -76,11 +76,79 @@ class ProctoredExam(TimeStampedModel): ...@@ -76,11 +76,79 @@ class ProctoredExam(TimeStampedModel):
return cls.objects.filter(course_id=course_id) return cls.objects.filter(course_id=course_id)
class ProctoredExamStudentAttemptManager(models.Manager):
"""
Custom manager
"""
def get_exam_attempt(self, exam_id, user_id):
"""
Returns the Student Exam Attempt object if found
else Returns None.
"""
try:
exam_attempt_obj = self.get(proctored_exam_id=exam_id, user_id=user_id)
except ObjectDoesNotExist: # pylint: disable=no-member
exam_attempt_obj = None
return exam_attempt_obj
def get_exam_attempt_by_id(self, attempt_id):
"""
Returns the Student Exam Attempt by the attempt_id else return None
"""
try:
exam_attempt_obj = self.get(id=attempt_id)
except ObjectDoesNotExist: # pylint: disable=no-member
exam_attempt_obj = None
return exam_attempt_obj
def get_exam_attempt_by_code(self, attempt_code):
"""
Returns the Student Exam Attempt object if found
else Returns None.
"""
try:
exam_attempt_obj = self.get(attempt_code=attempt_code)
except ObjectDoesNotExist: # pylint: disable=no-member
exam_attempt_obj = None
return exam_attempt_obj
def get_all_exam_attempts(self, course_id):
"""
Returns the Student Exam Attempts for the given course_id.
"""
return self.filter(proctored_exam__course_id=course_id)
def get_filtered_exam_attempts(self, course_id, search_by):
"""
Returns the Student Exam Attempts for the given course_id filtered by search_by.
"""
filtered_query = Q(proctored_exam__course_id=course_id) & (
Q(user__username__contains=search_by) | Q(user__email__contains=search_by)
)
return self.filter(filtered_query)
def get_active_student_attempts(self, user_id, course_id=None):
"""
Returns the active student exams (user in-progress exams)
"""
filtered_query = Q(user_id=user_id) & Q(started_at__isnull=False) & Q(completed_at__isnull=True)
if course_id is not None:
filtered_query = filtered_query & Q(proctored_exam__course_id=course_id)
return self.filter(filtered_query)
class ProctoredExamStudentAttempt(TimeStampedModel): class ProctoredExamStudentAttempt(TimeStampedModel):
""" """
Information about the Student Attempt on a Information about the Student Attempt on a
Proctored Exam. Proctored Exam.
""" """
objects = ProctoredExamStudentAttemptManager()
user = models.ForeignKey(User, db_index=True) user = models.ForeignKey(User, db_index=True)
proctored_exam = models.ForeignKey(ProctoredExam, db_index=True) proctored_exam = models.ForeignKey(ProctoredExam, db_index=True)
...@@ -148,60 +216,6 @@ class ProctoredExamStudentAttempt(TimeStampedModel): ...@@ -148,60 +216,6 @@ class ProctoredExamStudentAttempt(TimeStampedModel):
self.started_at = datetime.now(pytz.UTC) self.started_at = datetime.now(pytz.UTC)
self.save() self.save()
@classmethod
def get_exam_attempt(cls, exam_id, user_id):
"""
Returns the Student Exam Attempt object if found
else Returns None.
"""
try:
exam_attempt_obj = cls.objects.get(proctored_exam_id=exam_id, user_id=user_id)
except cls.DoesNotExist: # pylint: disable=no-member
exam_attempt_obj = None
return exam_attempt_obj
@classmethod
def get_exam_attempt_by_id(cls, attempt_id):
"""
Returns the Student Exam Attempt by the attempt_id else return None
"""
try:
exam_attempt_obj = cls.objects.get(id=attempt_id)
except cls.DoesNotExist: # pylint: disable=no-member
exam_attempt_obj = None
return exam_attempt_obj
@classmethod
def get_exam_attempt_by_code(cls, attempt_code):
"""
Returns the Student Exam Attempt object if found
else Returns None.
"""
try:
exam_attempt_obj = cls.objects.get(attempt_code=attempt_code)
except cls.DoesNotExist: # pylint: disable=no-member
exam_attempt_obj = None
return exam_attempt_obj
@classmethod
def get_all_exam_attempts(cls, course_id):
"""
Returns the Student Exam Attempts for the given course_id.
"""
return cls.objects.filter(proctored_exam__course_id=course_id)
@classmethod
def get_active_student_attempts(cls, user_id, course_id=None):
"""
Returns the active student exams (user in-progress exams)
"""
filtered_query = Q(user_id=user_id) & Q(started_at__isnull=False) & Q(completed_at__isnull=True)
if course_id is not None:
filtered_query = filtered_query & Q(proctored_exam__course_id=course_id)
return cls.objects.filter(filtered_query)
def delete_exam_attempt(self): def delete_exam_attempt(self):
""" """
deletes the exam attempt object. deletes the exam attempt object.
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment