feat: implement Pathways

Author: Agrendalath

openedx_learning/apps/openedx_content/admin.py

@@ -10,4 +10,5 @@
 from .applets.publishing.admin import *
 from .applets.sections.admin import *
 from .applets.subsections.admin import *
+from .applets.pathways.admin import *
 from .applets.units.admin import *

openedx_learning/apps/openedx_content/applets/pathways/__init__.py

@@ -0,0 +1,150 @@
+"""Django admin for Pathways."""
+
+from django.contrib import admin
+
+from openedx_learning.lib.admin_utils import ReadOnlyModelAdmin
+
+from .models import (
+    Pathway,
+    PathwayEnrollment,
+    PathwayEnrollmentAllowed,
+    PathwayEnrollmentAudit,
+    PathwayStep,
+)
+
+
+class PathwayStepInline(admin.TabularInline):
+    """Inline table for pathway steps within a pathway."""
+
+    model = PathwayStep
+    fields = ["order", "step_type", "context_key"]
+    ordering = ["order"]
+    extra = 0
+
+
+@admin.register(Pathway)
+class PathwayAdmin(admin.ModelAdmin):
+    """Admin for Pathway model."""
+
+    list_display = ["key", "display_name", "org", "is_active", "sequential", "created"]
+    list_filter = ["is_active", "sequential", "invite_only", "org"]
+    search_fields = ["key", "display_name"]
+    inlines = [PathwayStepInline]
+
+
+class PathwayEnrollmentAuditInline(admin.TabularInline):
+    """Inline admin for PathwayEnrollmentAudit records."""
+
+    model = PathwayEnrollmentAudit
+    fk_name = "enrollment"
+    extra = 0
+    exclude = ["enrollment_allowed"]
+    readonly_fields = [
+        "state_transition",
+        "enrolled_by",
+        "reason",
+        "org",
+        "role",
+        "created",
+    ]
+
+    def has_add_permission(self, request, obj=None):
+        """Disable manual creation of audit records."""
+        return False
+
+    def has_delete_permission(self, request, obj=None):
+        """Disable deletion of audit records."""
+        return False
+
+
+@admin.register(PathwayEnrollment)
+class PathwayEnrollmentAdmin(admin.ModelAdmin):
+    """Admin for PathwayEnrollment model."""
+
+    raw_id_fields = ("user",)
+    autocomplete_fields = ["pathway"]
+    list_display = ["id", "user", "pathway", "is_active", "created"]
+    list_filter = ["pathway__key", "created", "is_active"]
+    search_fields = ["id", "user__username", "pathway__key", "pathway__display_name"]
+    inlines = [PathwayEnrollmentAuditInline]
+
+
+class PathwayEnrollmentAllowedAuditInline(admin.TabularInline):
+    """Inline admin for PathwayEnrollmentAudit records related to enrollment allowed."""
+
+    model = PathwayEnrollmentAudit
+    fk_name = "enrollment_allowed"
+    extra = 0
+    exclude = ["enrollment"]
+    readonly_fields = [
+        "state_transition",
+        "enrolled_by",
+        "reason",
+        "org",
+        "role",
+        "created",
+    ]
+
+    def has_add_permission(self, request, obj=None):
+        """Disable manual creation of audit records."""
+        return False
+
+    def has_delete_permission(self, request, obj=None):
+        """Disable deletion of audit records."""
+        return False
+
+
+@admin.register(PathwayEnrollmentAllowed)
+class PathwayEnrollmentAllowedAdmin(admin.ModelAdmin):
+    """Admin for PathwayEnrollmentAllowed model."""
+
+    autocomplete_fields = ["pathway"]
+    list_display = ["id", "email", "get_user", "pathway", "created"]
+    list_filter = ["pathway", "created"]
+    search_fields = ["email", "user__username", "user__email", "pathway__key"]
+    readonly_fields = ["user", "created"]
+    inlines = [PathwayEnrollmentAllowedAuditInline]
+
+    def get_user(self, obj):
+        """Get the associated user, if any."""
+        return obj.user.username if obj.user else "-"
+
+    get_user.short_description = "User"
+
+
+@admin.register(PathwayEnrollmentAudit)
+class PathwayEnrollmentAuditAdmin(ReadOnlyModelAdmin):
+    """Admin configuration for PathwayEnrollmentAudit model."""
+
+    list_display = ["id", "state_transition", "enrolled_by", "get_enrollee", "get_pathway", "created", "org", "role"]
+    list_filter = ["state_transition", "created", "org", "role"]
+    search_fields = [
+        "enrolled_by__username",
+        "enrolled_by__email",
+        "enrollment__user__username",
+        "enrollment__user__email",
+        "enrollment_allowed__email",
+        "enrollment__pathway__key",
+        "enrollment_allowed__pathway__key",
+        "reason",
+    ]
+
+    def get_enrollee(self, obj):
+        """Get the enrollee (user or email)."""
+        if obj.enrollment:
+            return obj.enrollment.user.username
+        elif obj.enrollment_allowed:
+            return obj.enrollment_allowed.user.username if obj.enrollment_allowed.user else obj.enrollment_allowed.email
+        return "-"
+
+    get_enrollee.short_description = "Enrollee"
+
+    def get_pathway(self, obj):
+        """Get the pathway title."""
+        if obj.enrollment:
+            return obj.enrollment.pathway_id
+        elif obj.enrollment_allowed:
+            return obj.enrollment_allowed.pathway_id
+        return "-"
+
+    get_pathway.short_description = "Pathway"

openedx_learning/apps/openedx_content/applets/pathways/admin.py

@@ -0,0 +1,70 @@
+"""
+Opaque key for Pathways.
+
+Format: path-v1:{org}+{path_id}
+
+Can be moved to opaque-keys later if needed.
+"""
+
+import re
+from typing import Self
+from django.core.exceptions import ValidationError
+from opaque_keys import InvalidKeyError, OpaqueKey
+from opaque_keys.edx.keys import LearningContextKey
+from opaque_keys.edx.django.models import LearningContextKeyField
+
+PATHWAY_NAMESPACE = "path-v1"
+PATHWAY_PATTERN = r"([^+]+)\+([^+]+)"
+PATHWAY_URL_PATTERN = rf"(?P<pathway_key_str>{PATHWAY_NAMESPACE}:{PATHWAY_PATTERN})"
+
+
+class PathwayKey(LearningContextKey):
+    """
+    Key for identifying a Pathway.
+
+    Format: path-v1:{org}+{path_id}
+    Example: path-v1:OpenedX+DemoPathway
+    """
+
+    CANONICAL_NAMESPACE = PATHWAY_NAMESPACE
+    KEY_FIELDS = ("org", "path_id")
+    CHECKED_INIT = False
+
+    __slots__ = KEY_FIELDS
+    _pathway_key_regex = re.compile(PATHWAY_PATTERN)
+
+    def __init__(self, org: str, path_id: str):
+        super().__init__(org=org, path_id=path_id)
+
+    @classmethod
+    def _from_string(cls, serialized: str) -> Self:
+        """Return an instance of this class constructed from the given string."""
+        match = cls._pathway_key_regex.fullmatch(serialized)
+        if not match:
+            raise InvalidKeyError(cls, serialized)
+        return cls(*match.groups())
+
+    def _to_string(self) -> str:
+        """Return a string representing this key."""
+        return f"{self.org}+{self.path_id}"
+
+
+class PathwayKeyField(LearningContextKeyField):
+    """Django model field for PathwayKey."""
+
+    description = "A PathwayKey object"
+    KEY_CLASS = PathwayKey
+    # Declare the field types for the django-stubs mypy type hint plugin:
+    _pyi_private_set_type: PathwayKey | str | None
+    _pyi_private_get_type: PathwayKey | None
+
+    def __init__(self, *args, **kwargs):
+        kwargs.setdefault("max_length", 255)
+        super().__init__(*args, **kwargs)
+
+    def to_python(self, value) -> None | OpaqueKey:
+        """Convert the input value to a PathwayKey object."""
+        try:
+            return super().to_python(value)
+        except InvalidKeyError:
+            raise ValidationError("Invalid format. Use: 'path-v1:{org}+{path_id}'")

openedx_learning/apps/openedx_content/applets/pathways/keys.py

@@ -0,0 +1,17 @@
+"""Models that comprise the pathways applet."""
+
+from .enrollment import (
+    PathwayEnrollment,
+    PathwayEnrollmentAllowed,
+    PathwayEnrollmentAudit,
+)
+from .pathway import Pathway
+from .pathway_step import PathwayStep
+
+__all__ = [
+    "Pathway",
+    "PathwayEnrollment",
+    "PathwayEnrollmentAllowed",
+    "PathwayEnrollmentAudit",
+    "PathwayStep",
+]

openedx_learning/apps/openedx_content/applets/pathways/models/__init__.py

@@ -0,0 +1,139 @@
+"""Enrollment models for Pathways."""
+
+from django.conf import settings
+from django.db import models
+from django.utils.translation import gettext_lazy as _
+
+from openedx_learning.lib.validators import validate_utc_datetime
+
+from .pathway import Pathway
+
+
+class PathwayEnrollment(models.Model):
+    """
+    Tracks a user's enrollment in a pathway.
+
+    .. no_pii:
+    """
+
+    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="pathway_enrollments")
+    pathway = models.ForeignKey(Pathway, on_delete=models.CASCADE, related_name="enrollments")
+    is_active = models.BooleanField(default=True, help_text=_("Indicates whether the learner is enrolled."))
+    created = models.DateTimeField(auto_now_add=True, validators=[validate_utc_datetime])
+    modified = models.DateTimeField(auto_now=True, validators=[validate_utc_datetime])
+
+    def __str__(self) -> str:
+        """User-friendly string representation of this model."""
+        return f"PathwayEnrollment of user={self.user_id} in {self.pathway_id}"
+
+    class Meta:
+        """Model options."""
+
+        verbose_name = _("Pathway Enrollment")
+        verbose_name_plural = _("Pathway Enrollments")
+        constraints = [
+            models.UniqueConstraint(
+                fields=["user", "pathway"],
+                name="oel_pathway_enroll_uniq",
+            ),
+        ]
+
+
+class PathwayEnrollmentAllowed(models.Model):
+    """
+    Pre-registration allowlist for invite-only pathways.
+
+    These entities are created when learners are invited/enrolled before they register an account.
+
+    .. pii: The email field is not retired to allow future learners to enroll.
+    .. pii_types: email_address
+    .. pii_retirement: retained
+    """
+
+    pathway = models.ForeignKey(Pathway, on_delete=models.CASCADE, related_name="enrollment_allowed")
+    email = models.EmailField(db_index=True)
+    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, blank=True, null=True)
+    is_active = models.BooleanField(
+        default=True, db_index=True, help_text=_("Indicates if the enrollment allowance is active")
+    )
+    created = models.DateTimeField(auto_now_add=True, validators=[validate_utc_datetime])
+
+    def __str__(self) -> str:
+        """User-friendly string representation of this model."""
+        return f"PathwayEnrollmentAllowed for {self.email} in {self.pathway_id}"
+
+    class Meta:
+        """Model options."""
+
+        verbose_name = _("Pathway Enrollment Allowed")
+        verbose_name_plural = _("Pathway Enrollments Allowed")
+        constraints = [
+            models.UniqueConstraint(
+                fields=["pathway", "email"],
+                name="oel_pathway_enrollallow_uniq",
+            ),
+        ]
+
+
+# TODO: Create receivers to automatically create audit records.
+class PathwayEnrollmentAudit(models.Model):
+    """
+    Audit log for pathway enrollment changes.
+
+    .. no_pii:
+    """
+
+    # State transition constants (copied from openedx-platform to maintain consistency)
+    UNENROLLED_TO_ALLOWEDTOENROLL = "from unenrolled to allowed to enroll"
+    ALLOWEDTOENROLL_TO_ENROLLED = "from allowed to enroll to enrolled"
+    ENROLLED_TO_ENROLLED = "from enrolled to enrolled"
+    ENROLLED_TO_UNENROLLED = "from enrolled to unenrolled"
+    UNENROLLED_TO_ENROLLED = "from unenrolled to enrolled"
+    ALLOWEDTOENROLL_TO_UNENROLLED = "from allowed to enroll to unenrolled"
+    UNENROLLED_TO_UNENROLLED = "from unenrolled to unenrolled"
+    DEFAULT_TRANSITION_STATE = "N/A"
+
+    TRANSITION_STATES = (
+        (UNENROLLED_TO_ALLOWEDTOENROLL, UNENROLLED_TO_ALLOWEDTOENROLL),
+        (ALLOWEDTOENROLL_TO_ENROLLED, ALLOWEDTOENROLL_TO_ENROLLED),
+        (ENROLLED_TO_ENROLLED, ENROLLED_TO_ENROLLED),
+        (ENROLLED_TO_UNENROLLED, ENROLLED_TO_UNENROLLED),
+        (UNENROLLED_TO_ENROLLED, UNENROLLED_TO_ENROLLED),
+        (ALLOWEDTOENROLL_TO_UNENROLLED, ALLOWEDTOENROLL_TO_UNENROLLED),
+        (UNENROLLED_TO_UNENROLLED, UNENROLLED_TO_UNENROLLED),
+        (DEFAULT_TRANSITION_STATE, DEFAULT_TRANSITION_STATE),
+    )
+
+    enrolled_by = models.ForeignKey(
+        settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, related_name="pathway_enrollment_audits"
+    )
+    enrollment = models.ForeignKey(PathwayEnrollment, on_delete=models.CASCADE, null=True, related_name="audit_log")
+    enrollment_allowed = models.ForeignKey(
+        PathwayEnrollmentAllowed, on_delete=models.CASCADE, null=True, related_name="audit_log"
+    )
+    state_transition = models.CharField(max_length=255, choices=TRANSITION_STATES, default=DEFAULT_TRANSITION_STATE)
+    reason = models.TextField(blank=True)
+    org = models.CharField(max_length=255, blank=True, db_index=True)
+    role = models.CharField(max_length=255, blank=True)
+    created = models.DateTimeField(auto_now_add=True, validators=[validate_utc_datetime])
+
+    def __str__(self):
+        """User-friendly string representation of this model."""
+        enrollee = "unknown"
+        pathway = "unknown"
+
+        if self.enrollment:
+            enrollee = self.enrollment.user
+            pathway = self.enrollment.pathway_id
+        elif self.enrollment_allowed:
+            enrollee = self.enrollment_allowed.user or self.enrollment_allowed.email
+            pathway = self.enrollment_allowed.pathway_id
+
+        return f"{self.state_transition} for {enrollee} in {pathway}"
+
+    class Meta:
+        """Model options."""
+
+        verbose_name = _("Pathway Enrollment Audit")
+        verbose_name_plural = _("Pathway Enrollment Audits")
+        ordering = ["-created"]