feat: REST APIs for default-enterprise-enrollment-intentions

CHANGELOG.rst

@@ -17,6 +17,10 @@ Unreleased
 ----------
 * nothing unreleased
 
+[4.30.0]
+--------
+* feat: REST APIs for default-enterprise-enrollment-intentions
+
 [4.29.0]
 --------
 * feat: Create django admin for default enrollments

enterprise/__init__.py

@@ -2,4 +2,4 @@
 Your project description goes here.
 """
 
-__version__ = "4.29.0"
+__version__ = "4.30.0"

enterprise/api/v1/serializers.py

@@ -1894,3 +1894,132 @@ def get_role_assignments(self, obj):
             return role_assignments_by_ecu_id
         else:
             return None
+
+
+class DefaultEnterpriseEnrollmentIntentionSerializer(serializers.ModelSerializer):
+    """
+    Serializer for the DefaultEnterpriseEnrollmentIntention model.
+    """
+
+    course_run_key = serializers.SerializerMethodField()
+    is_course_run_enrollable = serializers.SerializerMethodField()
+    course_run_normalized_metadata = serializers.SerializerMethodField()
+    applicable_enterprise_catalog_uuids = serializers.SerializerMethodField()
+
+    class Meta:
+        model = models.DefaultEnterpriseEnrollmentIntention
+        fields = (
+            'uuid',
+            'content_key',
+            'enterprise_customer',
+            'course_run_key',
+            'is_course_run_enrollable',
+            'applicable_enterprise_catalog_uuids',
+            'course_run_normalized_metadata',
+            'created',
+            'modified',
+        )
+
+    def get_course_run_key(self, obj):
+        """
+        Get the course run key for the enrollment intention
+        """
+        return obj.course_run_key
+
+    def get_is_course_run_enrollable(self, obj):
+        """
+        Get the course run enrollable status for the enrollment intention
+        """
+        return obj.is_course_run_enrollable
+
+    def get_course_run_normalized_metadata(self, obj):
+        """
+        Get the course run for the enrollment intention
+        """
+        return obj.course_run_normalized_metadata
+
+    def get_applicable_enterprise_catalog_uuids(self, obj):
+        return obj.applicable_enterprise_catalog_uuids
+
+
+class DefaultEnterpriseEnrollmentIntentionLearnerStatusSerializer(serializers.Serializer):
+    """
+    Serializer for the DefaultEnterpriseEnrollmentIntentionLearnerStatus model.
+    """
+
+    lms_user_id = serializers.IntegerField()
+    user_email = serializers.EmailField()
+    enterprise_customer_uuid = serializers.UUIDField()
+    enrollment_statuses = serializers.SerializerMethodField()
+    metadata = serializers.SerializerMethodField()
+
+    def needs_enrollment_counts(self):
+        """
+        Return the counts of needs_enrollment.
+        """
+        needs_enrollment = self.context.get('needs_enrollment', {})
+        needs_enrollment_enrollable = needs_enrollment.get('enrollable', [])
+        needs_enrollment_not_enrollable = needs_enrollment.get('not_enrollable', [])
+
+        return {
+            'enrollable': len(needs_enrollment_enrollable),
+            'not_enrollable': len(needs_enrollment_not_enrollable),
+        }
+
+    def already_enrolled_count(self):
+        """
+        Return the count of already enrolled.
+        """
+        already_enrolled = self.context.get('already_enrolled', {})
+        return len(already_enrolled)
+
+    def total_default_enrollment_intention_count(self):
+        """
+        Return the total count of default enrollment intentions.
+        """
+        needs_enrollment_counts = self.needs_enrollment_counts()
+        total_needs_enrollment_enrollable = needs_enrollment_counts['enrollable']
+        total_needs_enrollment_not_enrollable = needs_enrollment_counts['not_enrollable']
+        return total_needs_enrollment_enrollable + total_needs_enrollment_not_enrollable + self.already_enrolled_count()
+
+    def get_enrollment_statuses(self, obj):  # pylint: disable=unused-argument
+        """
+        Serialize the enrollment statuses by converting querysets to serialized data.
+        """
+        needs_enrollment = self.context.get('needs_enrollment', {})
+        needs_enrollment_enrollable = needs_enrollment.get('enrollable', [])
+        needs_enrollment_not_enrollable = needs_enrollment.get('not_enrollable', [])
+        already_enrolled = self.context.get('already_enrolled', {})
+
+        needs_enrollment_enrollable_data = DefaultEnterpriseEnrollmentIntentionSerializer(
+            needs_enrollment_enrollable,
+            many=True
+        ).data
+        needs_enrollment_unenrollable_data = DefaultEnterpriseEnrollmentIntentionSerializer(
+            needs_enrollment_not_enrollable,
+            many=True
+        ).data
+        already_enrolled_data = DefaultEnterpriseEnrollmentIntentionSerializer(
+            already_enrolled,
+            many=True
+        ).data
+
+        return {
+            'needs_enrollment': {
+                'enrollable': needs_enrollment_enrollable_data,
+                'not_enrollable': needs_enrollment_unenrollable_data,
+            },
+            'already_enrolled': already_enrolled_data,
+        }
+
+    def get_metadata(self, obj):  # pylint: disable=unused-argument
+        """
+        Return the metadata for the default enterprise enrollment intention, including
+        number of default enterprise enrollment intentions that need enrollment, are already
+        enrolled by the learner.
+        """
+        return {
+            'total_default_enterprise_course_enrollments': self.total_default_enrollment_intention_count(),
+            'total_needs_enrollment': self.needs_enrollment_counts(),
+            'total_already_enrolled': self.already_enrolled_count(),
+        }

enterprise/api/v1/urls.py

@@ -26,6 +26,7 @@
     pending_enterprise_customer_admin_user,
     pending_enterprise_customer_user,
     plotly_auth,
+    default_enterprise_enrollments,
 )
 
 router = DefaultRouter()
@@ -82,6 +83,11 @@
 router.register(
     "enterprise_group", enterprise_group.EnterpriseGroupViewSet, 'enterprise-group'
 )
+router.register(
+    "default-enterprise-enrollment-intentions",
+    default_enterprise_enrollments.DefaultEnterpriseEnrollmentIntentionViewSet,
+    'default-enterprise-enrollment-intentions'
+)
 
 
 urlpatterns = [

enterprise/api/v1/views/default_enterprise_enrollments.py

@@ -0,0 +1,174 @@
+"""
+Views for default enterprise enrollments.
+"""
+
+from uuid import UUID
+
+from rest_framework.decorators import action
+from rest_framework.exceptions import ValidationError
+from rest_framework.response import Response
+from rest_framework import status, viewsets
+from edx_rbac.mixins import PermissionRequiredForListingMixin
+
+from django.contrib.auth import get_user_model
+
+from enterprise import models
+from enterprise.api.v1 import serializers
+from enterprise.api.v1.views.base_views import EnterpriseViewSet
+from enterprise.constants import DEFAULT_ENTERPRISE_ENROLLMENT_INTENTIONS_PERMISSION
+
+
+class DefaultEnterpriseEnrollmentIntentionViewSet(
+    PermissionRequiredForListingMixin,
+    EnterpriseViewSet,
+    viewsets.ModelViewSet,
+):
+    """
+    API views for default enterprise enrollment intentions
+    """
+
+    permission_required = DEFAULT_ENTERPRISE_ENROLLMENT_INTENTIONS_PERMISSION
+    list_lookup_field = 'enterprise_customer__uuid'
+    serializer_class = serializers.DefaultEnterpriseEnrollmentIntentionSerializer
+    http_method_names = ['get', 'post', 'delete']
+
+    @property
+    def requested_enterprise_customer_uuid(self):
+        """
+        Get and validate the enterprise customer UUID from the query parameters.
+        """
+        if not (enterprise_customer_uuid := self.request.query_params.get('enterprise_customer_uuid')):
+            raise ValidationError({"enterprise_customer_uuid": "This query parameter is required."})
+
+        try:
+            UUID(enterprise_customer_uuid)
+        except ValueError as exc:
+            raise ValidationError({"enterprise_customer_uuid": "Invalid UUID format."}) from exc
+        return enterprise_customer_uuid
+
+    @property
+    def requested_lms_user_id(self):
+        """
+        Get the (optional) LMS user ID from the request.
+        """
+        return self.request.query_params.get('lms_user_id')
+
+    @property
+    def base_queryset(self):
+        """
+        Required by the `PermissionRequiredForListingMixin`.
+        For non-list actions, this is what's returned by `get_queryset()`.
+        For list actions, some non-strict subset of this is what's returned by `get_queryset()`.
+        """
+        kwargs = {}
+        if self.requested_enterprise_customer_uuid:
+            kwargs['enterprise_customer'] = self.requested_enterprise_customer_uuid
+        return models.DefaultEnterpriseEnrollmentIntention.objects.filter(**kwargs)
+
+    @property
+    def user_for_learner_status(self):
+        """
+        Get the user for learner status based on the request.
+        """
+        if self.request.user.is_staff and self.requested_lms_user_id is not None:
+            User = get_user_model()
+            try:
+                return User.objects.get(id=self.requested_lms_user_id)
+            except User.DoesNotExist:
+                return None
+
+        return self.request.user
+
+    def get_permission_object(self):
+        """
+        Used for "retrieve" actions. Determines the context (enterprise UUID) to check
+        against for role-based permissions.
+        """
+        return self.requested_enterprise_customer_uuid
+
+    @action(detail=False, methods=['get'], url_path='learner-status')
+    def learner_status(self, request):  # pylint: disable=unused-argument
+        """
+        Get the status of the learner's enrollment in the default enterprise course.
+        """
+        # Validate the enterprise customer uuid.
+        try:
+            enterprise_customer_uuid = self.requested_enterprise_customer_uuid
+        except ValidationError as exc:
+            return Response(exc, status=status.HTTP_400_BAD_REQUEST)
+
+        # Validate the user for learner status exists and is associated
+        # with the enterprise customer.
+        if not (user_for_learner_status := self.user_for_learner_status):
+            return Response(
+                {'detail': f'User with lms_user_id {self.requested_lms_user_id} not found.'},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        try:
+            enterprise_customer_user = models.EnterpriseCustomerUser.objects.get(
+                user_id=user_for_learner_status.id,
+                enterprise_customer=enterprise_customer_uuid,
+            )
+        except models.EnterpriseCustomerUser.DoesNotExist:
+            return Response(
+                {
+                    'detail': (
+                        f'User with lms_user_id {user_for_learner_status.id} is not associated with '
+                        f'the enterprise customer {enterprise_customer_uuid}.'
+                    ),
+                },
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        # Retrieve configured default enrollment intentions for the enterprise customer
+        default_enrollment_intentions_for_customer = models.DefaultEnterpriseEnrollmentIntention.objects.filter(
+            enterprise_customer=enterprise_customer_uuid,
+        )
+
+        # Retrieve the course enrollments for the learner
+        enterprise_course_enrollments_for_learner = models.EnterpriseCourseEnrollment.objects.filter(
+            enterprise_customer_user=enterprise_customer_user,
+        )
+        enrolled_course_ids_for_learner = enterprise_course_enrollments_for_learner.values_list('course_id', flat=True)
+
+        already_enrolled = []
+        needs_enrollment_enrollable = []
+        needs_enrollment_not_enrollable = []
+
+        # Iterate through the default enrollment intentions and categorize them based
+        # on the learner's enrollment status (already enrolled, needs enrollment, etc.)
+        # and whether the course run is enrollable.
+        for default_enrollment_intention in default_enrollment_intentions_for_customer:
+            course_run_key = default_enrollment_intention.course_run_key
+            is_course_run_enrollable = default_enrollment_intention.is_course_run_enrollable
+            applicable_enterprise_catalog_uuids = default_enrollment_intention.applicable_enterprise_catalog_uuids
+
+            if course_run_key in enrolled_course_ids_for_learner:
+                # Learner is already enrolled in this course run
+                already_enrolled.append(default_enrollment_intention)
+            elif is_course_run_enrollable and applicable_enterprise_catalog_uuids:
+                # Learner needs enrollment, the course run is enrollable, and there are applicable catalogs
+                needs_enrollment_enrollable.append(default_enrollment_intention)
+            else:
+                # Learner needs enrollment, but the course run is not enrollable and/or there are no applicable catalogs
+                needs_enrollment_not_enrollable.append(default_enrollment_intention)
+
+        serializer_data = {
+            'lms_user_id': user_for_learner_status.id,
+            'user_email': user_for_learner_status.email,
+            'enterprise_customer_uuid': enterprise_customer_uuid,
+        }
+        serializer_context = {
+            'needs_enrollment': {
+                'enrollable': needs_enrollment_enrollable,
+                'not_enrollable': needs_enrollment_not_enrollable,
+            },
+            'already_enrolled': already_enrolled,
+        }
+        serializer = serializers.DefaultEnterpriseEnrollmentIntentionLearnerStatusSerializer(
+            data=serializer_data,
+            context=serializer_context,
+        )
+        serializer.is_valid(raise_exception=True)
+        return Response(serializer.data, status=status.HTTP_200_OK)

enterprise/api_client/enterprise_catalog.py

@@ -376,11 +376,14 @@ def enterprise_contains_content_items(self, enterprise_uuid, content_ids):
         The endpoint does not differentiate between course_run_ids and program_uuids so they can be used
         interchangeably. The two query parameters are left in for backwards compatability with edx-enterprise.
         """
-        query_params = {'course_run_ids': content_ids}
+        query_params = {
+            'course_run_ids': content_ids,
+            'get_catalogs_containing_specified_content_ids': True,
+        }
         api_url = self.get_api_url(f"{self.ENTERPRISE_CUSTOMER_ENDPOINT}/{enterprise_uuid}/contains_content_items")
         response = self.client.get(api_url, params=query_params)
         response.raise_for_status()
-        return response.json()['contains_content_items']
+        return response.json()
 
     @UserAPIClient.refresh_token
     def get_content_metadata_content_identifier(self, enterprise_uuid, content_id):

enterprise/cache_utils.py

@@ -19,7 +19,7 @@ def versioned_cache_key(*args):
     """
     components = [str(arg) for arg in args]
     components.append(code_version)
-    if stamp_from_settings := getattr(settings, 'CACHE_KEY_VERSION_STAMP', None):
+    if stamp_from_settings := getattr(settings, 'ENTERPRISE_CACHE_KEY_VERSION_STAMP', None):
         components.append(stamp_from_settings)
     decoded_cache_key = CACHE_KEY_SEP.join(components)
     return hashlib.sha512(decoded_cache_key.encode()).hexdigest()

enterprise/constants.py

@@ -150,6 +150,11 @@ class CourseModes:
 ENTERPRISE_REPORTING_CONFIG_ADMIN_ROLE = 'reporting_config_admin'
 ENTERPRISE_FULFILLMENT_OPERATOR_ROLE = 'fulfillment_operator'
 ENTERPRISE_SSO_ORCHESTRATOR_OPERATOR_ROLE = 'sso_orchestrator_operator'
+
+# Default enterprise enrollment roles/permissions
+DEFAULT_ENTERPRISE_ENROLLMENT_INTENTIONS_ROLE = 'default_enterprise_enrollment_intentions_learner'
+DEFAULT_ENTERPRISE_ENROLLMENT_INTENTIONS_PERMISSION = 'enterprise.can_view_default_enterprise_enrollment_intentions'
+
 # Provisioning admins roles:
 PROVISIONING_ENTERPRISE_CUSTOMER_ADMIN_ROLE = 'provisioning_enterprise_customer_admin'
 PROVISIONING_PENDING_ENTERPRISE_CUSTOMER_ADMIN_ROLE = 'provisioning_pending_enterprise_customer_users_admin'