feat: Create django admin for default enrollments

CHANGELOG.rst

@@ -17,6 +17,10 @@ Unreleased
 ----------
 * nothing unreleased
 
+[4.29.0]
+--------
+* feat: Create django admin for default enrollments
+
 [4.28.1]
 --------
 * feat: making to-be-deleted model field nullable

enterprise/__init__.py

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

enterprise/admin/__init__.py

@@ -43,6 +43,7 @@
 )
 from enterprise.api_client.lms import CourseApiClient, EnrollmentApiClient
 from enterprise.config.models import UpdateRoleAssignmentsWithCustomersConfig
+from enterprise.models import DefaultEnterpriseEnrollmentIntention
 from enterprise.utils import (
     discovery_query_url,
     get_all_field_names,
@@ -109,6 +110,31 @@ def get_formset(self, request, obj=None, **kwargs):
         return formset
 
 
+class EnterpriseCustomerDefaultEnterpriseEnrollmentIntentionInline(admin.TabularInline):
+    """
+    Django admin model for EnterpriseCustomerCatalog.
+    The admin interface has the ability to edit models on the same page as a parent model. These are called inlines.
+    https://docs.djangoproject.com/en/1.8/ref/contrib/admin/#django.contrib.admin.StackedInline
+    """
+
+    model = models.DefaultEnterpriseEnrollmentIntention
+    fields = ('content_key', 'course_run_key_for_enrollment',)
+    readonly_fields = ('course_run_key_for_enrollment',)
+    extra = 0
+    can_delete = True
+
+    @admin.display(description='Course run key for enrollment')
+    def course_run_key_for_enrollment(self, obj):
+        """
+        Returns the course run key based on the content type.
+        If the content type is a course, we retrieve the advertised_course_run key.
+        """
+        content_type = obj.content_type
+        if content_type == 'course_run':
+            return obj.content_key
+        return obj.current_course_run_key
+
+
 class PendingEnterpriseCustomerAdminUserInline(admin.TabularInline):
     """
     Django admin inline model for PendingEnterpriseCustomerAdminUser.
@@ -227,6 +253,7 @@ class EnterpriseCustomerAdmin(DjangoObjectActions, SimpleHistoryAdmin):
         EnterpriseCustomerBrandingConfigurationInline,
         EnterpriseCustomerIdentityProviderInline,
         EnterpriseCustomerCatalogInline,
+        EnterpriseCustomerDefaultEnterpriseEnrollmentIntentionInline,
         PendingEnterpriseCustomerAdminUserInline,
     ]
 
@@ -1294,7 +1321,6 @@ class LearnerCreditEnterpriseCourseEnrollmentAdmin(admin.ModelAdmin):
         'uuid',
         'fulfillment_type',
         'enterprise_course_enrollment',
-        'is_revoked',
         'modified',
     )
 
@@ -1332,8 +1358,7 @@ class DefaultEnterpriseEnrollmentIntentionAdmin(admin.ModelAdmin):
 
     readonly_fields = (
         'current_course_run_key',
-        'current_course_run_enrollable',
-        'current_course_run_enroll_by_date',
+        'is_removed',
     )
 
     search_fields = (

enterprise/cache_utils.py

@@ -0,0 +1,25 @@
+"""
+Utils for interacting with cache interfaces.
+"""
+import hashlib
+
+from django.conf import settings
+
+from enterprise import __version__ as code_version
+
+CACHE_KEY_SEP = ':'
+DEFAULT_NAMESPACE = 'edx-enterprise-default'
+
+
+def versioned_cache_key(*args):
+    """
+    Utility to produce a versioned cache key, which includes
+    an optional settings variable and the current code version,
+    so that we can perform key-based cache invalidation.
+    """
+    components = [str(arg) for arg in args]
+    components.append(code_version)
+    if stamp_from_settings := getattr(settings, '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/content_metadata/api.py

@@ -0,0 +1,52 @@
+"""
+Python API for interacting with content metadata.
+"""
+import logging
+
+from edx_django_utils.cache import TieredCache
+from requests.exceptions import HTTPError
+
+from django.conf import settings
+
+from enterprise.api_client.enterprise_catalog import EnterpriseCatalogApiClient
+from enterprise.cache_utils import versioned_cache_key
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_CACHE_TIMEOUT = getattr(settings, 'CONTENT_METADATA_CACHE_TIMEOUT', 60 * 5)
+
+
+def get_and_cache_customer_content_metadata(enterprise_customer_uuid, content_key, timeout=None):
+    """
+    Returns the metadata corresponding to the requested
+    ``content_key`` within catalogs associated to the provided ``enterprise_customer``.
+
+    The response is cached in a ``TieredCache`` (meaning in both the RequestCache,
+    _and_ the django cache for the configured expiration period).
+
+    Returns: A dict with content metadata for the given key.
+    Raises: An HTTPError if there's a problem getting the content metadata
+      via the enterprise-catalog service.
+    """
+    cache_key = versioned_cache_key('get_content_metadata_content_identifier', enterprise_customer_uuid, content_key)
+    cached_response = TieredCache.get_cached_response(cache_key)
+    if cached_response.is_found:
+        logger.info(f'cache hit for enterprise customer {enterprise_customer_uuid} and content {content_key}')
+        return cached_response.value
+
+    try:
+        result = EnterpriseCatalogApiClient().get_content_metadata_content_identifier(
+            enterprise_uuid=enterprise_customer_uuid,
+            content_id=content_key,
+        )
+    except HTTPError as exc:
+        raise exc
+
+    logger.info(
+        'Fetched catalog for customer %s and content_key %s. Result = %s',
+        enterprise_customer_uuid,
+        content_key,
+        result,
+    )
+    TieredCache.set_all_tiers(cache_key, result, timeout or DEFAULT_CACHE_TIMEOUT)
+    return result

enterprise/models.py

@@ -61,6 +61,7 @@
     FulfillmentTypes,
     json_serialized_course_modes,
 )
+from enterprise.content_metadata.api import get_and_cache_customer_content_metadata
 from enterprise.errors import LinkUserToEnterpriseError
 from enterprise.event_bus import send_learner_credit_course_enrollment_revoked_event
 from enterprise.logging import getEnterpriseLogger
@@ -71,6 +72,7 @@
     CourseEnrollmentDowngradeError,
     CourseEnrollmentPermissionError,
     NotConnectedToOpenEdX,
+    get_advertised_course_run,
     get_configuration_value,
     get_default_invite_key_expiration_date,
     get_ecommerce_worker_user,
@@ -2511,12 +2513,33 @@ class DefaultEnterpriseEnrollmentIntention(TimeStampedModel, SoftDeletableModel)
     )
     history = HistoricalRecords()
 
-    @cached_property
+    @property
+    def content_metadata_for_content_key(self):
+        """
+        Retrieves the cached content metadata for the instance content_key
+        Determines catalog from the catalogs associated to the enterprise customer
+        """
+        return get_and_cache_customer_content_metadata(
+            enterprise_customer_uuid=self.enterprise_customer.uuid,
+            content_key=self.content_key,
+        )
+
+    @property
     def current_course_run(self):  # pragma: no cover
         """
         Metadata describing the current course run for this default enrollment intention.
         """
-        return {}
+        content_metadata = self.content_metadata_for_content_key
+        if not content_metadata:
+            return {}
+        if self.content_type == 'course':
+            if course_run := get_advertised_course_run(content_metadata):
+                return course_run
+            return {}
+        course_runs = content_metadata.get('course_runs', [])
+        return next(
+            (course_run for course_run in course_runs if course_run['key'].lower() == self.content_key.lower()), {}
+        )
 
     @property
     def current_course_run_key(self):  # pragma: no cover
@@ -2525,6 +2548,22 @@ def current_course_run_key(self):  # pragma: no cover
         """
         return self.current_course_run.get('key')
 
+    @property
+    def get_content_type(self):
+        """
+        Determines the content_type for a given content_key by validating a return value
+        from `get_content_metadata_content_identifier` and first checking if the passed content_key
+        is a top level course key, if not and the API call resolved correctly, we can assume the content_type
+        is a `course_run`
+        """
+        content_metadata = self.content_metadata_for_content_key
+        if not content_metadata:
+            return None
+        content_metadata_key = content_metadata.get('key').lower()
+        if content_metadata_key == self.content_key.lower():
+            return 'course'
+        return 'course_run'
+
     @property
     def current_course_run_enrollable(self):  # pragma: no cover
         """
@@ -2539,6 +2578,15 @@ def current_course_run_enroll_by_date(self):  # pragma: no cover
         """
         return datetime.datetime.min
 
+    def clean(self):
+        if not (self.content_type or self.get_content_type):
+            raise ValidationError('The content key did not matc')
+
+    def save(self, *args, **kwargs):
+        if not self.content_type:
+            self.content_type = self.get_content_type
+        super().save(*args, **kwargs)
+
 
 class DefaultEnterpriseEnrollmentRealization(TimeStampedModel):
     """