Andal.LND/edx-platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: upgrading students_update_enrollment api to DRF (#37074)

Browse Source 
* feat: upgrading students_update_enrollment api to DRF
This commit is contained in:
Hunzlah Malik
2025-07-29 19:36:52 +05:00
committed by GitHub
parent 4e1ba65183
commit cdf5083544
4 changed files with 153 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
lms/djangoapps/bulk_enroll/views.py
View File

@@ -3,8 +3,6 @@ API views for Bulk Enrollment
"""
import json
from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
from opaque_keys import InvalidKeyError
from opaque_keys.edx.keys import CourseKey
@@ -15,7 +13,7 @@ from six.moves import zip_longest
from common.djangoapps.util.disable_rate_limit import can_disable_rate_limit
from lms.djangoapps.bulk_enroll.serializers import BulkEnrollmentSerializer
from lms.djangoapps.instructor.views.api import students_update_enrollment
from lms.djangoapps.instructor.views.api import StudentsUpdateEnrollmentView
from openedx.core.djangoapps.course_groups.cohorts import add_user_to_cohort, get_cohort_by_name
from openedx.core.djangoapps.course_groups.models import CourseUserGroup
from openedx.core.djangoapps.enrollments.views import EnrollmentUserThrottle
@@ -89,8 +87,14 @@ class BulkEnrollView(APIView):
}
for course_id, cohort_name in zip_longest(serializer.data.get('courses'),
serializer.data.get('cohorts', [])):
response = students_update_enrollment(self.request, course_id=course_id)
response_content = json.loads(response.content.decode('utf-8'))
# Internal request to DRF view
view = StudentsUpdateEnrollmentView()
response_content = view._process_student_enrollment( # pylint: disable=protected-access
user=request.user,
course_id=course_id,
data=request.data,
secure=request.is_secure()
)
if cohort_name:
try:

272
lms/djangoapps/instructor/views/api.py
View File

@@ -114,9 +114,10 @@ from lms.djangoapps.instructor.views.serializer import (
UserSerializer,
UniqueStudentIdentifierSerializer,
ProblemResetSerializer,
RescoreEntranceExamSerializer,
UpdateForumRoleMembershipSerializer,
OverrideProblemScoreSerializer
RescoreEntranceExamSerializer,
OverrideProblemScoreSerializer,
StudentsUpdateEnrollmentSerializer
)
from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
from openedx.core.djangoapps.course_groups.cohorts import add_user_to_cohort, is_course_cohorted
@@ -755,161 +756,156 @@ def create_and_enroll_user(
return errors
@require_POST
@ensure_csrf_cookie
@cache_control(no_cache=True, no_store=True, must_revalidate=True)
@require_course_permission(permissions.CAN_ENROLL)
@require_post_params(action="enroll or unenroll", identifiers="stringified list of emails and/or usernames")
def students_update_enrollment(request, course_id): # lint-amnesty, pylint: disable=too-many-statements
@method_decorator(cache_control(no_cache=True, no_store=True, must_revalidate=True), name='dispatch')
class StudentsUpdateEnrollmentView(APIView):
"""
Enroll or unenroll students by email.
Requires staff access.
Query Parameters:
- action in ['enroll', 'unenroll']
- identifiers is string containing a list of emails and/or usernames separated by anything split_input_list can handle. # lint-amnesty, pylint: disable=line-too-long
- auto_enroll is a boolean (defaults to false)
If auto_enroll is false, students will be allowed to enroll.
If auto_enroll is true, students will be enrolled as soon as they register.
- email_students is a boolean (defaults to false)
If email_students is true, students will be sent email notification
If email_students is false, students will not be sent email notification
Returns an analog to this JSON structure: {
"action": "enroll",
"auto_enroll": false,
"results": [
{
"email": "testemail@test.org",
"before": {
"enrollment": false,
"auto_enroll": false,
"user": true,
"allowed": false
},
"after": {
"enrollment": true,
"auto_enroll": false,
"user": true,
"allowed": false
}
}
]
}
API view to enroll or unenroll students in a course.
"""
course_id = CourseKey.from_string(course_id)
action = request.POST.get('action')
identifiers_raw = request.POST.get('identifiers')
identifiers = _split_input_list(identifiers_raw)
auto_enroll = _get_boolean_param(request, 'auto_enroll')
email_students = _get_boolean_param(request, 'email_students')
reason = request.POST.get('reason')
enrollment_obj = None
state_transition = DEFAULT_TRANSITION_STATE
permission_classes = (IsAuthenticated, permissions.InstructorPermission)
permission_name = permissions.CAN_ENROLL
email_params = {}
if email_students:
course = get_course_by_id(course_id)
email_params = get_email_params(course, auto_enroll, secure=request.is_secure())
@method_decorator(ensure_csrf_cookie)
def post(self, request, course_id):
"""
Handle POST request to enroll or unenroll students.
results = []
for identifier in identifiers: # lint-amnesty, pylint: disable=too-many-nested-blocks
# First try to get a user object from the identifier
user = None
email = None
language = None
try:
user = get_student_from_identifier(identifier)
except User.DoesNotExist:
email = identifier
else:
email = user.email
language = get_user_email_language(user)
Parameters:
- action (str): 'enroll' or 'unenroll'
- identifiers (str): comma/newline separated emails or usernames
- auto_enroll (bool): auto-enroll in verified track if applicable
- email_students (bool): whether to send enrollment emails
- reason (str, optional): reason for enrollment change
try:
# Use django.core.validators.validate_email to check email address
# validity (obviously, cannot check if email actually /exists/,
# simply that it is plausibly valid)
validate_email(email) # Raises ValidationError if invalid
if action == 'enroll':
before, after, enrollment_obj = enroll_email(
course_id, email, auto_enroll, email_students, {**email_params}, language=language
)
before_enrollment = before.to_dict()['enrollment']
before_user_registered = before.to_dict()['user']
before_allowed = before.to_dict()['allowed']
after_enrollment = after.to_dict()['enrollment']
after_allowed = after.to_dict()['allowed']
Returns:
- JSON response with action, auto_enroll flag, and enrollment results.
"""
response_payload = self._process_student_enrollment(
user=request.user,
course_id=course_id,
data=request.data,
secure=request.is_secure()
)
return JsonResponse(response_payload)
if before_user_registered:
if after_enrollment:
if before_enrollment:
state_transition = ENROLLED_TO_ENROLLED
else:
if before_allowed:
def _process_student_enrollment(self, user, course_id, data, secure): # pylint: disable=too-many-statements
"""
Core logic for enrolling or unenrolling students.
:param user: User making the request
:param course_id: Course identifier
:param data: Request data containing action, identifiers, etc.
:param secure: Whether the request is secure (HTTPS)
"""
# Validate request data with serializer
serializer = StudentsUpdateEnrollmentSerializer(data=data)
serializer.is_valid(raise_exception=True)
# Extract validated data
action = serializer.validated_data['action']
identifiers_raw = serializer.validated_data['identifiers']
auto_enroll = serializer.validated_data['auto_enroll']
email_students = serializer.validated_data['email_students']
reason = serializer.validated_data.get('reason')
# Parse identifiers
identifiers = _split_input_list(identifiers_raw)
course_key = CourseKey.from_string(course_id)
enrollment_obj = None
state_transition = DEFAULT_TRANSITION_STATE
email_params = {}
if email_students:
course = get_course_by_id(course_key)
email_params = get_email_params(course, auto_enroll, secure=secure)
results = []
for identifier in identifiers: # pylint: disable=too-many-nested-blocks
identified_user = None
email = None
language = None
try:
identified_user = get_student_from_identifier(identifier)
except User.DoesNotExist:
email = identifier
else:
email = identified_user.email
language = get_user_email_language(identified_user)
try:
validate_email(email) # Raises ValidationError if invalid
if action == 'enroll':
before, after, enrollment_obj = enroll_email(
course_key, email, auto_enroll, email_students, {**email_params}, language=language
)
before_enrollment = before.to_dict()['enrollment']
before_user_registered = before.to_dict()['user']
before_allowed = before.to_dict()['allowed']
after_enrollment = after.to_dict()['enrollment']
after_allowed = after.to_dict()['allowed']
if before_user_registered:
if after_enrollment:
if before_enrollment:
state_transition = ENROLLED_TO_ENROLLED
elif before_allowed:
state_transition = ALLOWEDTOENROLL_TO_ENROLLED
else:
state_transition = UNENROLLED_TO_ENROLLED
else:
if after_allowed:
elif after_allowed:
state_transition = UNENROLLED_TO_ALLOWEDTOENROLL
elif action == 'unenroll':
before, after = unenroll_email(
course_id, email, email_students, {**email_params}, language=language
)
before_enrollment = before.to_dict()['enrollment']
before_allowed = before.to_dict()['allowed']
enrollment_obj = CourseEnrollment.get_enrollment(user, course_id) if user else None
elif action == 'unenroll':
before, after = unenroll_email(
course_key, email, email_students, {**email_params}, language=language
)
before_enrollment = before.to_dict()['enrollment']
before_allowed = before.to_dict()['allowed']
enrollment_obj = (
CourseEnrollment.get_enrollment(identified_user, course_key)
if identified_user else None
)
if before_enrollment:
state_transition = ENROLLED_TO_UNENROLLED
else:
if before_allowed:
if before_enrollment:
state_transition = ENROLLED_TO_UNENROLLED
elif before_allowed:
state_transition = ALLOWEDTOENROLL_TO_UNENROLLED
else:
state_transition = UNENROLLED_TO_UNENROLLED
except ValidationError:
results.append({
'identifier': identifier,
'invalidIdentifier': True,
})
except Exception as exc: # pylint: disable=broad-except
log.exception("Error while processing student")
log.exception(exc)
results.append({
'identifier': identifier,
'error': True,
})
else:
return HttpResponseBadRequest(strip_tags(
f"Unrecognized action '{action}'"
))
ManualEnrollmentAudit.create_manual_enrollment_audit(
identified_user, email, state_transition, reason, enrollment_obj
)
results.append({
'identifier': identifier,
'before': before.to_dict(),
'after': after.to_dict(),
})
except ValidationError:
# Flag this email as an error if invalid, but continue checking
# the remaining in the list
results.append({
'identifier': identifier,
'invalidIdentifier': True,
})
except Exception as exc: # pylint: disable=broad-except
# catch and log any exceptions
# so that one error doesn't cause a 500.
log.exception("Error while #{}ing student")
log.exception(exc)
results.append({
'identifier': identifier,
'error': True,
})
else:
ManualEnrollmentAudit.create_manual_enrollment_audit(
request.user, email, state_transition, reason, enrollment_obj
)
results.append({
'identifier': identifier,
'before': before.to_dict(),
'after': after.to_dict(),
})
response_payload = {
'action': action,
'results': results,
'auto_enroll': auto_enroll,
}
return JsonResponse(response_payload)
return {
'action': action,
'auto_enroll': auto_enroll,
'results': results,
}
@method_decorator(cache_control(no_cache=True, no_store=True, must_revalidate=True), name='dispatch')

2
lms/djangoapps/instructor/views/api_urls.py
View File

@@ -21,7 +21,7 @@ v1_api_urls = [
]
urlpatterns = [
path('students_update_enrollment', api.students_update_enrollment, name='students_update_enrollment'),
path('students_update_enrollment', api.StudentsUpdateEnrollmentView.as_view(), name='students_update_enrollment'),
path('register_and_enroll_students', api.RegisterAndEnrollStudents.as_view(), name='register_and_enroll_students'),
path('list_course_role_members', api.ListCourseRoleMembersView.as_view(), name='list_course_role_members'),
path('modify_access', api.ModifyAccess.as_view(), name='modify_access'),

9
lms/djangoapps/instructor/views/serializer.py
View File

@@ -485,6 +485,15 @@ class RescoreEntranceExamSerializer(serializers.Serializer):
only_if_higher = serializers.BooleanField(required=False, allow_null=True)
class StudentsUpdateEnrollmentSerializer(serializers.Serializer):
"""Serializer for student enroll/unenroll actions."""
action = serializers.ChoiceField(choices=["enroll", "unenroll"])
identifiers = serializers.CharField()
auto_enroll = serializers.BooleanField(default=False)
email_students = serializers.BooleanField(default=False)
reason = serializers.CharField(required=False, allow_blank=True)
class OverrideProblemScoreSerializer(UniqueStudentIdentifierSerializer):
"""
Serializer for overriding a student's score for a specific problem.