feat: initial implementation

.annotation_safe_list.yml

@@ -19,6 +19,26 @@ auth.User:
   ".. pii_retirement": consumer_api
 contenttypes.ContentType:
   ".. no_pii:": This model has no PII
+completion.BlockCompletion:
+  ".. no_pii:": This model has no PII
+completion_aggregator.Aggregator:
+  ".. no_pii:": This model has no PII
+completion_aggregator.CacheGroupInvalidation:
+  ".. no_pii:": This model has no PII
+completion_aggregator.StaleCompletion:
+  ".. pii": This model contains a username; the entries are regularly cleaned up (usually every hour)
+django_celery_beat.ClockedSchedule:
+  ".. no_pii:": This model has no PII
+django_celery_beat.CrontabSchedule:
+  ".. no_pii:": This model has no PII
+django_celery_beat.IntervalSchedule:
+  ".. no_pii:": This model has no PII
+django_celery_beat.PeriodicTask:
+  ".. no_pii:": This model has no PII
+django_celery_beat.PeriodicTasks:
+  ".. no_pii:": This model has no PII
+django_celery_beat.SolarSchedule:
+  ".. no_pii:": This model has no PII
 sessions.Session:
   ".. no_pii:": This model has no PII
 social_django.Association:

openedx_certificates/admin.py

@@ -0,0 +1,204 @@
+"""Admin page configuration for the openedx-certificates app."""
+
+from __future__ import annotations
+
+import importlib
+import inspect
+from typing import TYPE_CHECKING, Generator
+
+from django import forms
+from django.contrib import admin
+from django.utils.html import format_html
+from django_object_actions import DjangoObjectActions, action
+from django_reverse_admin import ReverseModelAdmin
+
+from .models import (
+    ExternalCertificate,
+    ExternalCertificateAsset,
+    ExternalCertificateCourseConfiguration,
+    ExternalCertificateType,
+)
+
+if TYPE_CHECKING:  # pragma: no cover
+    from django.http import HttpRequest
+    from django_celery_beat.models import IntervalSchedule
+
+
+class ExternalCertificateTypeAdminForm(forms.ModelForm):
+    """Generate a list of available functions for the function fields."""
+
+    retrieval_func = forms.ChoiceField(choices=[])
+    generation_func = forms.ChoiceField(choices=[])
+
+    @staticmethod
+    def _available_functions(module: str, prefix: str) -> Generator[tuple[str, str], None, None]:
+        """
+        Import a module and return all functions in it that start with a specific prefix.
+
+        :param module: The name of the module to import.
+        :param prefix: The prefix of the function names to return.
+
+        :return: A tuple containing the functions that start with the prefix in the module.
+        """
+        # TODO: Implement plugin support for the functions.
+        _module = importlib.import_module(module)
+        return (
+            (f'{obj.__module__}.{name}', f'{obj.__module__}.{name}')
+            for name, obj in inspect.getmembers(_module, inspect.isfunction)
+            if name.startswith(prefix)
+        )
+
+    @staticmethod
+    def _get_docstring_custom_options(func: str) -> str:
+        """
+        Get the docstring of the function and return the "Options:" section.
+
+        :param func: The function to get the docstring for.
+        :returns: The "Options:" section of the docstring.
+        """
+        try:
+            docstring = (
+                'Custom options:'
+                + inspect.getdoc(
+                    getattr(
+                        importlib.import_module(func.rsplit('.', 1)[0]),
+                        func.rsplit('.', 1)[1],
+                    ),
+                ).split("Options:")[1]
+            )
+        except IndexError:
+            docstring = (
+                'Custom options are not documented for this function. If you selected a different function, '
+                'you need to save your changes to see an updated docstring.'
+            )
+        # Use pre to preserve the newlines and indentation.
+        return f'<pre>{docstring}</pre>'
+
+    def __init__(self, *args, **kwargs):
+        """Initializes the choices for the retrieval and generation function selection fields."""
+        super().__init__(*args, **kwargs)
+        self.fields['retrieval_func'].choices = self._available_functions(
+            'openedx_certificates.processors',
+            'retrieve_',
+        )
+        if self.instance.retrieval_func:
+            self.fields['retrieval_func'].help_text = self._get_docstring_custom_options(self.instance.retrieval_func)
+        self.fields['generation_func'].choices = self._available_functions(
+            'openedx_certificates.generators',
+            'generate_',
+        )
+        if self.instance.generation_func:
+            self.fields['generation_func'].help_text = self._get_docstring_custom_options(self.instance.generation_func)
+
+    class Meta:  # noqa: D106
+        model = ExternalCertificateType
+        fields = '__all__'  # noqa: DJ007
+
+
+@admin.register(ExternalCertificateType)
+class ExternalCertificateTypeAdmin(admin.ModelAdmin):  # noqa: D101
+    form = ExternalCertificateTypeAdminForm
+    list_display = ('name', 'retrieval_func', 'generation_func')
+
+
+@admin.register(ExternalCertificateAsset)
+class ExternalCertificateAssetAdmin(admin.ModelAdmin):  # noqa: D101
+    list_display = ('description', 'asset_slug')
+    prepopulated_fields = {"asset_slug": ("description",)}  # noqa: RUF012
+
+
+@admin.register(ExternalCertificateCourseConfiguration)
+class ExternalCertificateCourseConfigurationAdmin(DjangoObjectActions, ReverseModelAdmin):
+    """
+    Admin page for the course-specific certificate configuration for each certificate type.
+
+    It manages the associations between configuration and its corresponding periodic task.
+    The reverse inline provides a way to manage the periodic task from the configuration page.
+    """
+
+    inline_type = 'stacked'
+    inline_reverse = [  # noqa: RUF012
+        (
+            'periodic_task',
+            {'fields': ['enabled', 'interval', 'crontab', 'clocked', 'start_time', 'expires', 'one_off']},
+        ),
+    ]
+    list_display = ('course_id', 'certificate_type', 'enabled', 'interval')
+    search_fields = ('course_id', 'certificate_type__name')
+    list_filter = ('course_id', 'certificate_type')
+
+    def get_inline_instances(
+        self,
+        request: HttpRequest,
+        obj: ExternalCertificateCourseConfiguration = None,
+    ) -> list[admin.ModelAdmin]:
+        """
+        Hide inlines on the "Add" view in Django admin, and show them on the "Change" view.
+
+        It differentiates "add" and change "view" based on the requested path because the `obj` parameter can be `None`
+        in the "Change" view when rendering the inlines.
+
+        :param request: HttpRequest object
+        :param obj: The object being changed, None for add view
+        :return: A list of InlineModelAdmin instances to be rendered for add/changing an object
+        """
+        return super().get_inline_instances(request, obj) if '/add/' not in request.path else []
+
+    def enabled(self, obj: ExternalCertificateCourseConfiguration) -> bool:
+        """Return the 'enabled' status of the periodic task."""
+        return obj.periodic_task.enabled
+
+    enabled.boolean = True
+
+    # noinspection PyMethodMayBeStatic
+    def interval(self, obj: ExternalCertificateCourseConfiguration) -> IntervalSchedule:
+        """Return the interval of the certificate generation task."""
+        return obj.periodic_task.interval
+
+    def get_readonly_fields(self, _request: HttpRequest, obj: ExternalCertificateCourseConfiguration = None) -> tuple:
+        """Make the course_id field read-only."""
+        if obj:  # editing an existing object
+            return *self.readonly_fields, 'course_id', 'certificate_type'
+        return self.readonly_fields
+
+    @action(label="Generate certificates")
+    def generate_certificates(self, _request: HttpRequest, obj: ExternalCertificateCourseConfiguration):
+        """
+        Custom action to generate certificates for the current ExternalCertificateCourse instance.
+
+        Args:
+            _request: The request object.
+            obj: The ExternalCertificateCourse instance.
+        """
+        # TODO: Use the celery task instead of the generate_certificates method.
+        obj.generate_certificates()
+
+    change_actions = ('generate_certificates',)
+
+
+@admin.register(ExternalCertificate)
+class ExternalCertificateAdmin(admin.ModelAdmin):  # noqa: D101
+    list_display = ('user_id', 'user_full_name', 'course_id', 'certificate_type', 'status', 'url')
+    readonly_fields = (
+        'user_id',
+        'user_full_name',
+        'course_id',
+        'certificate_type',
+        'status',
+        'url',
+        'legacy_id',
+        'generation_task_id',
+    )
+
+    def get_form(self, request: HttpRequest, obj: ExternalCertificate | None = None, **kwargs) -> forms.ModelForm:
+        """Hide the download_url field."""
+        form = super().get_form(request, obj, **kwargs)
+        form.base_fields['download_url'].widget = forms.HiddenInput()
+        return form
+
+    # noinspection PyMethodMayBeStatic
+    def url(self, obj: ExternalCertificate) -> str:
+        """Display the download URL as a clickable link."""
+        if obj.download_url:
+            return format_html("<a href='{url}'>{url}</a>", url=obj.download_url)
+        return "-"

openedx_certificates/compat.py

@@ -0,0 +1,78 @@
+"""
+Proxies and compatibility code for edx-platform features.
+
+This module moderates access to all edx-platform features allowing for cross-version compatibility code.
+It also simplifies running tests outside edx-platform's environment by stubbing these functions in unit tests.
+"""
+from __future__ import annotations
+
+from contextlib import contextmanager
+from typing import TYPE_CHECKING
+
+from celery import Celery
+from django.conf import settings
+
+if TYPE_CHECKING:  # pragma: no cover
+    from django.contrib.auth.models import User
+    from opaque_keys.edx.keys import CourseKey
+
+# TODO: Do we still need all these pylint disable comments? We switched to ruff.
+
+
+def get_celery_app() -> Celery:
+    """Get Celery app to reuse configuration and queues."""
+    if getattr(settings, "TESTING", False):
+        # We can ignore this in the testing environment.
+        return Celery(task_always_eager=True)
+
+    # noinspection PyUnresolvedReferences,PyPackageRequirements
+    from lms import CELERY_APP
+
+    return CELERY_APP  # pragma: no cover
+
+
+def get_course_grading_policy(course_id: CourseKey) -> dict:
+    """Get the course grading policy from Open edX."""
+    # noinspection PyUnresolvedReferences,PyPackageRequirements
+    from xmodule.modulestore.django import modulestore
+
+    return modulestore().get_course(course_id).grading_policy["GRADER"]
+
+
+def get_course_name(course_id: CourseKey) -> str:
+    """Get the course name from Open edX."""
+    # noinspection PyUnresolvedReferences,PyPackageRequirements
+    from openedx.core.djangoapps.content.learning_sequences.api import get_course_outline
+
+    course_outline = get_course_outline(course_id)
+    return (course_outline and course_outline.title) or str(course_id)
+
+
+def get_course_enrollments(course_id: CourseKey) -> list[User]:
+    """Get the course enrollments from Open edX."""
+    # noinspection PyUnresolvedReferences,PyPackageRequirements
+    from common.djangoapps.student.models import CourseEnrollment
+
+    enrollments = CourseEnrollment.objects.filter(course_id=course_id, is_active=True).select_related('user')
+    return [enrollment.user for enrollment in enrollments]
+
+
+@contextmanager
+def prefetch_course_grades(course_id: CourseKey, users: list[User]):
+    """Prefetch the course grades from Open edX."""
+    # noinspection PyUnresolvedReferences,PyPackageRequirements
+    from lms.djangoapps.grades.api import clear_prefetched_course_grades, prefetch_course_grades
+
+    prefetch_course_grades(course_id, users)
+    try:
+        yield
+    finally:
+        clear_prefetched_course_grades(course_id)
+
+
+def get_course_grade_factory():  # noqa: ANN201
+    """Get the course grade factory from Open edX."""
+    # noinspection PyUnresolvedReferences,PyPackageRequirements
+    from lms.djangoapps.grades.course_grade_factory import CourseGradeFactory
+
+    return CourseGradeFactory()

openedx_certificates/exceptions.py

@@ -0,0 +1,9 @@
+"""Custom exceptions for the openedx-certificates app."""
+
+
+class AssetNotFoundError(Exception):
+    """Raised when the asset_slug is not found in the ExternalCertificateAsset model."""
+
+
+class CertificateGenerationError(Exception):
+    """Raised when the certificate generation Celery task fails."""