feat: adds course enrollments endpoint for admin view

CHANGELOG.rst

@@ -17,6 +17,10 @@ Unreleased
 ----------
 * nothing unreleased
 
+[5.9.0]
+--------
+* feat: creates an endpoint for admin to view a learner's course enrollments.
+
 [5.8.2]
 --------
 * feat: add index to `user_fk` in `EnterpriseCustomerUser` model. Part 2 of adding auth_user as a foreign key.

enterprise/__init__.py

@@ -2,4 +2,4 @@
 Your project description goes here.
 """
 
-__version__ = "5.8.2"
+__version__ = "5.9.0"

enterprise/api/utils.py

@@ -1,7 +1,6 @@
 """
 Utility functions for the Enterprise API.
 """
-
 from django.conf import settings
 from django.contrib import auth
 from django.utils.translation import gettext as _
@@ -29,6 +28,17 @@
 )
 
 
+class CourseRunProgressStatuses:
+    """
+    Class to group statuses that a course run can be in with respect to user progress.
+    """
+
+    IN_PROGRESS = 'in_progress'
+    UPCOMING = 'upcoming'
+    COMPLETED = 'completed'
+    SAVED_FOR_LATER = 'saved_for_later'
+
+
 def get_service_usernames():
     """
     Return the set of service usernames that are given extended permissions in the API.

enterprise/api/v1/serializers.py

@@ -21,12 +21,15 @@
 from django.utils.translation import gettext_lazy as _
 
 from enterprise import models, utils  # pylint: disable=cyclic-import
+from enterprise.api.utils import CourseRunProgressStatuses  # pylint: disable=cyclic-import
 from enterprise.api.v1.fields import Base64EmailCSVField
 from enterprise.api_client.lms import ThirdPartyAuthApiClient
 from enterprise.constants import (
     ENTERPRISE_ADMIN_ROLE,
     ENTERPRISE_PERMISSION_GROUPS,
+    EXEC_ED_COURSE_TYPE,
     GROUP_MEMBERSHIP_ACCEPTED_STATUS,
+    PRODUCT_SOURCE_2U,
     DefaultColors,
 )
 from enterprise.logging import getEnterpriseLogger
@@ -51,6 +54,18 @@
 )
 from enterprise.validators import validate_pgp_key
 
+try:
+    from federated_content_connector.models import CourseDetails
+except ImportError:
+    CourseDetails = None
+
+try:
+    from lms.djangoapps.certificates.api import get_certificate_for_user
+except ImportError:
+    get_certificate_for_user = None
+    get_course_run_url = None
+    get_emails_enabled = None
+
 LOGGER = getEnterpriseLogger(__name__)
 User = auth.get_user_model()
 
@@ -430,6 +445,144 @@ class Meta:
     course_end = serializers.DateTimeField()
 
 
+class EnterpriseCourseEnrollmentAdminViewSerializer(serializers.ModelSerializer):
+    """
+    Serializer for EnterpriseCourseEnrollment model.
+    """
+    class Meta:
+        model = models.EnterpriseCourseEnrollment
+        fields = '__all__'
+
+    def to_representation(self, instance):
+        """
+        Convert the `EnterpriseCourseEnrollment` instance into a dictionary representation.
+
+        Args:
+            instance (EnterpriseCourseEnrollment): The enrollment instance being serialized.
+
+        Returns:
+            dict: A dictionary representation of the enrollment data.
+        """
+        representation = super().to_representation(instance)
+        course_run_id = instance.course_id
+        user = self.context['enterprise_customer_user']
+        course_overview = self._get_course_overview(course_run_id)
+
+        certificate_info = get_certificate_for_user(user.username, course_run_id) or {}
+
+        representation['course_run_id'] = course_run_id
+        representation['course_run_status'] = self._get_course_run_status(
+            course_overview,
+            certificate_info,
+            instance
+        )
+        representation['created'] = instance.created.isoformat()
+        representation['start_date'] = course_overview['start']
+        representation['end_date'] = course_overview['end']
+        representation['display_name'] = course_overview['display_name_with_default']
+        representation['org_name'] = course_overview['display_org_with_default']
+        representation['pacing'] = course_overview['pacing']
+        representation['is_revoked'] = instance.license.is_revoked if instance.license else False
+        representation['is_enrollment_active'] = instance.is_active
+        representation['mode'] = instance.mode
+
+        if CourseDetails:
+            course_details = CourseDetails.objects.filter(id=course_run_id).first()
+            if course_details:
+                representation['course_key'] = course_details.course_key
+                representation['course_type'] = course_details.course_type
+                representation['product_source'] = course_details.product_source
+                representation['start_date'] = course_details.start_date or representation['start_date']
+                representation['end_date'] = course_details.end_date or representation['end_date']
+                representation['enroll_by'] = course_details.enroll_by
+
+                if (course_details.product_source == PRODUCT_SOURCE_2U and
+                        course_details.course_type == EXEC_ED_COURSE_TYPE):
+                    representation['course_run_status'] = self._get_exec_ed_course_run_status(
+                        course_details,
+                        certificate_info,
+                        instance
+                    )
+        return representation
+
+    def _get_course_overview(self, course_run_id):
+        """
+        Get the appropriate course overview from the context.
+        """
+        for overview in self.context['course_overviews']:
+            if overview['id'] == course_run_id:
+                return overview
+
+        return None
+
+    def _get_exec_ed_course_run_status(self, course_details, certificate_info, enterprise_enrollment):
+        """
+        Get the status of a exec ed course run, given the state of a user's certificate in the course.
+
+        A run is considered "complete" when either the course run has ended OR the user has earned a
+        passing certificate.
+
+        Arguments:
+            course_details : the details for the exececutive education course run
+            certificate_info: A dict containing the following key:
+                ``is_passing``: whether the  user has a passing certificate in the course run
+
+        Returns:
+            status: one of (
+                CourseRunProgressStatuses.SAVED_FOR_LATER,
+                CourseRunProgressStatuses.COMPLETE,
+                CourseRunProgressStatuses.IN_PROGRESS,
+                CourseRunProgressStatuses.UPCOMING,
+            )
+        """
+        if enterprise_enrollment and enterprise_enrollment.saved_for_later:
+            return CourseRunProgressStatuses.SAVED_FOR_LATER
+
+        is_certificate_passing = certificate_info.get('is_passing', False)
+        start_date = course_details.start_date
+        end_date = course_details.end_date
+
+        has_started = datetime.now(pytz.utc) > start_date if start_date is not None else True
+        has_ended = datetime.now(pytz.utc) > end_date if end_date is not None else False
+
+        if has_ended or is_certificate_passing:
+            return CourseRunProgressStatuses.COMPLETED
+        if has_started:
+            return CourseRunProgressStatuses.IN_PROGRESS
+        return CourseRunProgressStatuses.UPCOMING
+
+    def _get_course_run_status(self, course_overview, certificate_info, enterprise_enrollment):
+        """
+        Get the status of a course run, given the state of a user's certificate in the course.
+
+        A run is considered "complete" when either the course run has ended OR the user has earned a
+        passing certificate.
+
+        Arguments:
+            course_overview (CourseOverview): the overview for the course run
+            certificate_info: A dict containing the following key:
+                ``is_passing``: whether the  user has a passing certificate in the course run
+
+        Returns:
+            status: one of (
+                CourseRunProgressStatuses.SAVED_FOR_LATER,
+                CourseRunProgressStatuses.COMPLETE,
+                CourseRunProgressStatuses.IN_PROGRESS,
+                CourseRunProgressStatuses.UPCOMING,
+            )
+        """
+        if enterprise_enrollment and enterprise_enrollment.saved_for_later:
+            return CourseRunProgressStatuses.SAVED_FOR_LATER
+
+        is_certificate_passing = certificate_info.get('is_passing', False)
+
+        if course_overview['has_ended'] or is_certificate_passing:
+            return CourseRunProgressStatuses.COMPLETED
+        if course_overview['has_started']:
+            return CourseRunProgressStatuses.IN_PROGRESS
+        return CourseRunProgressStatuses.UPCOMING
+
+
 class EnterpriseCourseEnrollmentWriteSerializer(serializers.ModelSerializer):
     """
     Serializer for writing to the EnterpriseCourseEnrollment model.

enterprise/api/v1/urls.py

@@ -224,6 +224,13 @@
         enterprise_customer_members.EnterpriseCustomerMembersViewSet.as_view({'get': 'get_members'}),
         name='enterprise-customer-members'
     ),
+    re_path(
+        r'^enterprise-course-enrollment-admin/?$',
+        enterprise_course_enrollment.EnterpriseCourseEnrollmentAdminViewSet.as_view(
+            {'get': 'get_enterprise_course_enrollment'}
+        ),
+        name='enterprise-course-enrollment-admin'
+    ),
 ]
 
 urlpatterns += router.urls

enterprise/api/v1/views/enterprise_course_enrollment.py

@@ -3,13 +3,19 @@
 """
 from django_filters.rest_framework import DjangoFilterBackend
 from edx_rest_framework_extensions.paginators import DefaultPagination
-from rest_framework import filters
+from rest_framework import filters, permissions
+from rest_framework.decorators import action
+from rest_framework.pagination import PageNumberPagination
+from rest_framework.response import Response
+from rest_framework.status import HTTP_400_BAD_REQUEST
 
 from django.core.paginator import Paginator
+from django.shortcuts import get_object_or_404
 from django.utils.functional import cached_property
 
 from enterprise import models
 from enterprise.api.filters import EnterpriseCourseEnrollmentFilterBackend
+from enterprise.api.utils import CourseRunProgressStatuses  # pylint: disable=cyclic-import
 from enterprise.api.v1 import serializers
 from enterprise.api.v1.views.base_views import EnterpriseReadWriteModelViewSet
 
@@ -18,6 +24,10 @@
 except ImportError:
     def read_replica_or_default():
         return None
+try:
+    from openedx.core.djangoapps.content.course_overviews.api import get_course_overviews
+except ImportError:
+    get_course_overviews = None
 
 
 class PaginatorWithOptimizedCount(Paginator):
@@ -86,3 +96,89 @@ def get_serializer_class(self):
         if self.request.method in ('GET',):
             return serializers.EnterpriseCourseEnrollmentWithAdditionalFieldsReadOnlySerializer
         return serializers.EnterpriseCourseEnrollmentWriteSerializer
+
+
+class EnterpriseCourseEnrollmentAdminPagination(PageNumberPagination):
+    """
+    Custom pagination class for Enterprise Course Enrollment API.
+    """
+    page_size = 10
+    page_size_query_param = 'page_size'
+    max_page_size = 100
+
+
+class EnterpriseCourseEnrollmentAdminViewSet(EnterpriseReadWriteModelViewSet):
+    """
+    API views for the ``enterprise-course-enrollment-admin`` API endpoint.
+    """
+
+    queryset = models.EnterpriseCourseEnrollment.with_additional_fields.all()
+    serializer_class = serializers.EnterpriseCourseEnrollmentAdminViewSerializer
+    permission_classes = (permissions.IsAuthenticated,)
+    pagination_class = EnterpriseCourseEnrollmentAdminPagination
+
+    @action(detail=False, methods=['get'])
+    def get_enterprise_course_enrollment(self, request):
+        """
+        Endpoint to get enrollments for a learner by `lms_user_id` and `enterprise_uuid` viewed
+        by an admin of that enterprise.
+
+        Parameters:
+        - `lms_user_id` (str): Filter results by the LMS user ID.
+        - `enterprise_uuid` (str): Filter results by the Enterprise UUID.
+        """
+        lms_user_id = request.query_params.get('lms_user_id')
+        enterprise_uuid = request.query_params.get('enterprise_uuid')
+        if not lms_user_id or not enterprise_uuid:
+            return Response(
+                {"error": "Both 'lms_user_id' and 'enterprise_uuid' are required parameters."},
+                status=HTTP_400_BAD_REQUEST
+            )
+        enterprise_customer_user = get_object_or_404(
+            models.EnterpriseCustomerUser,
+            user_id=lms_user_id,
+            enterprise_customer__uuid=enterprise_uuid
+        )
+        enterprise_enrollments = models.EnterpriseCourseEnrollment.objects.filter(
+            enterprise_customer_user=enterprise_customer_user
+        )
+        filtered_enterprise_enrollments = [record for record in enterprise_enrollments if record.course_enrollment]
+        course_overviews = get_course_overviews([record.course_id for record in filtered_enterprise_enrollments])
+        serialized_data = serializers.EnterpriseCourseEnrollmentAdminViewSerializer(
+            filtered_enterprise_enrollments,
+            many=True,
+            context={
+                'request': request,
+                'enterprise_customer_user': enterprise_customer_user,
+                'course_overviews': course_overviews,
+            }
+        ).data
+        page = self.paginate_queryset(serialized_data)
+        if page is not None:
+            grouped_data = self._group_course_enrollments_by_status(page)
+            return self.get_paginated_response(grouped_data)
+
+        grouped_data = self._group_course_enrollments_by_status(serialized_data)
+        return Response(grouped_data)
+
+    def _group_course_enrollments_by_status(self, course_enrollments):
+        """
+        Groups course enrollments by their status.
+
+        Args:
+            enrollments (list): List of course enrollment dictionaries.
+
+        Returns:
+            dict: A dictionary where keys are status names and values are lists of enrollments with that status.
+        """
+        statuses = {
+            CourseRunProgressStatuses.IN_PROGRESS: [],
+            CourseRunProgressStatuses.UPCOMING: [],
+            CourseRunProgressStatuses.COMPLETED: [],
+            CourseRunProgressStatuses.SAVED_FOR_LATER: [],
+        }
+        for enrollment in course_enrollments:
+            status = enrollment.get('course_run_status')
+            if status in statuses:
+                statuses[status].append(enrollment)
+        return statuses