diff --git a/.importlinter b/.importlinter
index 5aad15ed9..13bbad04c 100644
--- a/.importlinter
+++ b/.importlinter
@@ -9,6 +9,7 @@ root_packages =
openedx_learning
openedx_content
openedx_tagging
+ openedx_catalog
openedx_django_lib
openedx_core
contract_types =
@@ -21,9 +22,9 @@ contract_types =
name = "top-level source folders are layered correctly"
type = layers
layers =
- # Learning-domain features (currently CBE; Learning Pathways to follow).
- # May build on content, catalog and tagging. Nothing below may import it:
- # in particular, openedx_tagging must never know that CBE exists.
+ # Learning-domain features (CBE and Pathways). May build on content, catalog and
+ # tagging. Nothing below may import it: in particular, openedx_tagging must never
+ # know that CBE exists.
openedx_learning
# Content: authoring-side models and APIs. May reference catalog models
diff --git a/src/openedx_catalog/admin.py b/src/openedx_catalog/admin.py
index 2aa4f3ca6..f21a02a6b 100644
--- a/src/openedx_catalog/admin.py
+++ b/src/openedx_catalog/admin.py
@@ -13,13 +13,16 @@
from django.utils.html import format_html
from django.utils.translation import gettext_lazy as _
-from .models import CatalogCourse, CourseRun
+from .models import CatalogCourse, CatalogPathway, CourseRun, PathwayCategory, PathwayEnrollment
if TYPE_CHECKING:
class CatalogCourseWithRunCount(CatalogCourse):
run_count: int
+ class PathwayCategoryWithPathwayCount(PathwayCategory):
+ pathway_count: int
+
class CatalogCourseAdmin(admin.ModelAdmin):
"""
@@ -110,3 +113,88 @@ def warnings(self, obj: CourseRun) -> str | None:
admin.site.register(CourseRun, CourseRunAdmin)
+
+
+@admin.register(PathwayCategory)
+class PathwayCategoryAdmin(admin.ModelAdmin):
+ """
+ The PathwayCategory model admin.
+
+ Renaming a category changes what learners see. It does not change the authoring-side terminology, which is always
+ "Pathway".
+ """
+
+ list_display = ["title", "title_plural_display", "category_code", "pathways_summary"]
+ search_fields = ["title", "title_plural", "category_code"]
+
+ def get_readonly_fields(self, request, obj: PathwayCategory | None = None) -> tuple[str, ...]:
+ if obj: # editing an existing object; the code is what other systems key off
+ return ("category_code",)
+ return tuple()
+
+ def get_queryset(self, request) -> QuerySet[PathwayCategoryWithPathwayCount]:
+ """Add the 'pathway_count' to the list_display queryset"""
+ qs = super().get_queryset(request)
+ qs = qs.annotate(pathway_count=Count("pathways"))
+ return qs
+
+ @admin.display(description=_("Plural"))
+ def title_plural_display(self, obj: PathwayCategory) -> str:
+ """Show the plural learners see, whether set explicitly or derived from the title."""
+ return obj.get_title_plural()
+
+ @admin.display(description=_("Pathways"), ordering="pathway_count")
+ def pathways_summary(self, obj: PathwayCategoryWithPathwayCount) -> str:
+ """Link to the catalog pathways using this category"""
+ if obj.pathway_count == 0:
+ return "-"
+ url = reverse("admin:openedx_catalog_catalogpathway_changelist") + f"?category={obj.pk}"
+ return format_html('{} ', url, obj.pathway_count)
+
+
+@admin.register(CatalogPathway)
+class CatalogPathwayAdmin(admin.ModelAdmin):
+ """
+ The CatalogPathway model admin.
+
+ This edits only the catalog half of a Pathway. The Items a learner must complete live on the content side, in the
+ openedx_learning app, and are versioned there.
+ """
+
+ list_filter = ["org__short_name", "category"]
+ list_display = ["title", "category", "org_display", "pathway_code", "key_str", "created_date", "modified"]
+ search_fields = ["title", "pathway_code"]
+
+ def get_readonly_fields(self, request, obj: CatalogPathway | None = None) -> tuple[str, ...]:
+ if obj: # editing an existing object
+ return ("org", "pathway_code")
+ return tuple()
+
+ @admin.display(description="Organization", ordering="org__short_name")
+ def org_display(self, obj: CatalogPathway) -> str:
+ """Display the organization, only showing the short_name if different from full name"""
+ if obj.org.name == obj.org.short_name:
+ return obj.org.short_name
+ return str(obj.org)
+
+ @admin.display(description=_("Created"), ordering="created")
+ def created_date(self, obj: CatalogPathway) -> datetime.date:
+ """Display the created date without the timestamp"""
+ return obj.created.date()
+
+
+@admin.register(PathwayEnrollment)
+class PathwayEnrollmentAdmin(admin.ModelAdmin):
+ """
+ The PathwayEnrollment model admin.
+ """
+
+ list_display = ["user", "catalog_pathway", "is_active", "created_date", "modified"]
+ list_filter = ["is_active", "catalog_pathway__category"]
+ # There may be very many users and a fair number of pathways, so don't use
+ raw_id_fields = ["user", "catalog_pathway"]
+
+ @admin.display(description=_("Enrolled"), ordering="created")
+ def created_date(self, obj: PathwayEnrollment) -> datetime.date:
+ """Display the enrollment date without the timestamp"""
+ return obj.created.date()
diff --git a/src/openedx_catalog/api.py b/src/openedx_catalog/api.py
index 457f33e9a..320536753 100644
--- a/src/openedx_catalog/api.py
+++ b/src/openedx_catalog/api.py
@@ -25,6 +25,7 @@
# Import only the public API methods denoted with __all__
# pylint: disable=wildcard-import
-from .api_impl import *
+from .api_impl.courses import *
+from .api_impl.pathways import *
# You'll also want the models from .models_api
diff --git a/src/openedx_catalog/api_impl/__init__.py b/src/openedx_catalog/api_impl/__init__.py
new file mode 100644
index 000000000..b205d91e6
--- /dev/null
+++ b/src/openedx_catalog/api_impl/__init__.py
@@ -0,0 +1,3 @@
+"""
+Implementation of the `openedx_catalog` API. Import it from `openedx_catalog.api` instead.
+"""
diff --git a/src/openedx_catalog/api_impl.py b/src/openedx_catalog/api_impl/courses.py
similarity index 98%
rename from src/openedx_catalog/api_impl.py
rename to src/openedx_catalog/api_impl/courses.py
index 03c9a3ee7..0620b38a0 100644
--- a/src/openedx_catalog/api_impl.py
+++ b/src/openedx_catalog/api_impl/courses.py
@@ -1,5 +1,5 @@
"""
-Implementation of the `openedx_catalog` API.
+Implementation of the course part of the `openedx_catalog` API: catalog courses and course runs.
"""
import logging
@@ -9,7 +9,7 @@
from organizations.api import ensure_organization # type: ignore[import]
from organizations.api import exceptions as org_exceptions
-from .models import CatalogCourse, CourseRun
+from ..models import CatalogCourse, CourseRun
log = logging.getLogger(__name__)
diff --git a/src/openedx_catalog/api_impl/pathways.py b/src/openedx_catalog/api_impl/pathways.py
new file mode 100644
index 000000000..e72d30b98
--- /dev/null
+++ b/src/openedx_catalog/api_impl/pathways.py
@@ -0,0 +1,323 @@
+"""
+Implementation of the Pathway part of the `openedx_catalog` API: categories, catalog pathways and enrollment.
+
+A Pathway is split into a catalog half (these models) and a versioned content half in
+`openedx_learning.applets.pathways`. See the openedx_learning ADR 0007. The functions below only touch the catalog half;
+creating and versioning the *definition* of a Pathway is done through `openedx_learning.api`.
+"""
+
+import logging
+from typing import overload
+
+from django.db.models.query import QuerySet
+from django.utils import timezone
+
+from ..models import CatalogPathway, PathwayCategory, PathwayEnrollment
+
+log = logging.getLogger(__name__)
+
+# These are the public API methods that anyone can use
+__all__ = [
+ "get_pathway_category",
+ "get_pathway_categories",
+ "create_pathway_category",
+ "update_pathway_category",
+ "delete_pathway_category",
+ "get_catalog_pathway",
+ "get_catalog_pathways",
+ "create_catalog_pathway",
+ "update_catalog_pathway",
+ "delete_catalog_pathway",
+ "enroll_in_pathway",
+ "unenroll_from_pathway",
+ "is_enrolled_in_pathway",
+ "get_pathway_enrollments",
+]
+
+
+def get_pathway_category(category_code: str) -> PathwayCategory:
+ """
+ Get a `PathwayCategory` by its stable code.
+
+ ⚠️ Does not check permissions.
+ """
+ return PathwayCategory.objects.get(category_code=category_code)
+
+
+def _pathway_category(category: PathwayCategory | str) -> PathwayCategory:
+ """
+ Resolve a `PathwayCategory` given either as an instance or by its code.
+ """
+ if isinstance(category, PathwayCategory):
+ return category
+ return get_pathway_category(category)
+
+
+def get_pathway_categories() -> QuerySet[PathwayCategory]:
+ """
+ List all pathway categories, ordered by title, e.g. for a picker shown when creating a catalog pathway.
+
+ ⚠️ Does not check permissions.
+ """
+ return PathwayCategory.objects.all()
+
+
+def create_pathway_category(*, category_code: str, title: str, title_plural: str = "") -> PathwayCategory:
+ """
+ Create a `PathwayCategory`.
+
+ The ``category_code`` can't be changed later, so pick one that stays accurate if the ``title`` is reworded. Leave
+ ``title_plural`` blank unless the title doesn't pluralize by appending "s" (see `PathwayCategory.title_plural`).
+
+ ⚠️ Does not check permissions.
+ """
+ return PathwayCategory.objects.create(category_code=category_code, title=title, title_plural=title_plural)
+
+
+def update_pathway_category(
+ category: PathwayCategory | str,
+ *,
+ title: str | None = None,
+ title_plural: str | None = None,
+) -> None:
+ """
+ Update a `PathwayCategory`, given as an instance or by its code. Pass `None` for a field to leave it unchanged.
+
+ Pass ``title_plural=""`` to go back to deriving the plural from the title. The ``category_code`` can't be changed,
+ because it's what code and imports key off.
+
+ ⚠️ Does not check permissions.
+ """
+ pc = _pathway_category(category)
+ update_fields = []
+ for field_name, value in (
+ ("title", title),
+ ("title_plural", title_plural),
+ ):
+ if value is not None:
+ setattr(pc, field_name, value)
+ update_fields.append(field_name)
+ if update_fields:
+ pc.save(update_fields=update_fields)
+
+
+def delete_pathway_category(category: PathwayCategory | str) -> None:
+ """
+ Delete a `PathwayCategory`, given as an instance or by its code.
+
+ This fails with a `ProtectedError` while any catalog pathway uses it; move those pathways to another category first
+ with `update_catalog_pathway`.
+
+ ⚠️ Does not check permissions.
+ """
+ _pathway_category(category).delete()
+
+
+def _catalog_pathways() -> QuerySet[CatalogPathway]:
+ """
+ Catalog pathways along with what's needed to show them: their org and category.
+ """
+ return CatalogPathway.objects.select_related("org", "category")
+
+
+@overload
+def get_catalog_pathway(*, org_code: str, pathway_code: str) -> CatalogPathway: ...
+@overload
+def get_catalog_pathway(*, key_str: str) -> CatalogPathway: ...
+@overload
+def get_catalog_pathway(*, pk: CatalogPathway.ID) -> CatalogPathway: ...
+
+
+def get_catalog_pathway(
+ pk: CatalogPathway.ID | None = None,
+ key_str: str = "",
+ org_code: str = "",
+ pathway_code: str = "",
+) -> CatalogPathway:
+ """
+ Get a catalog pathway.
+
+ Its org and category come with it, so showing them costs no further queries.
+
+ ⚠️ Does not check permissions or visibility rules.
+
+ The `CatalogPathway` may not have any content implementing it yet. Because `openedx_catalog` never knows about the
+ content side, this function cannot tell you whether it does; ask `openedx_learning.api` instead.
+ """
+ assert pk or key_str or (org_code and pathway_code)
+ if pk:
+ assert not org_code
+ assert not key_str
+ return _catalog_pathways().get(pk=pk)
+ if key_str:
+ assert key_str.startswith("catalog-pathway:")
+ assert not org_code
+ assert not pathway_code
+ _, org_code, pathway_code = key_str.split(":", 2)
+ return _catalog_pathways().get(org__short_name=org_code, pathway_code=pathway_code)
+
+
+def get_catalog_pathways(
+ *,
+ org_code: str | None = None,
+ category_code: str | None = None,
+) -> QuerySet[CatalogPathway]:
+ """
+ List catalog pathways, most recently created first, optionally narrowed down to an org and/or a category.
+
+ Both filters match exactly; an unknown org or category simply matches nothing. The result is a `QuerySet`, so that
+ callers can narrow, search and paginate it further. Each pathway's org and category come with it, so showing them
+ costs no further queries.
+
+ ⚠️ Does not check permissions or visibility rules. That suits authoring and administration, but a listing shown to
+ learners will need the visibility logic described in the `openedx_catalog.api` docstring, which doesn't exist yet.
+
+ As with `get_catalog_pathway`, this can't tell you which pathways have content implementing them yet; ask
+ `openedx_learning.api` instead.
+ """
+ pathways = _catalog_pathways()
+ if org_code is not None:
+ pathways = pathways.filter(org__short_name=org_code)
+ if category_code is not None:
+ pathways = pathways.filter(category__category_code=category_code)
+ return pathways
+
+
+def create_catalog_pathway(
+ *,
+ org_code: str,
+ pathway_code: str,
+ category: PathwayCategory,
+ title: str = "",
+ description: str = "",
+) -> CatalogPathway:
+ """
+ Create a `CatalogPathway`.
+
+ The `Organization` identified by `org_code` must already exist. The category is always required: there is no
+ default, so a UI that wants to preselect one decides which itself.
+
+ This creates only the catalog half of a Pathway. Use `openedx_learning.api` to create the versioned content that
+ implements it.
+
+ ⚠️ Does not check permissions.
+ """
+ pathway = CatalogPathway(pathway_code=pathway_code, title=title, category=category, description=description)
+ pathway.org_code = org_code # Resolves the Organization by short_name; raises Organization.DoesNotExist.
+ pathway.save()
+ return pathway
+
+
+def update_catalog_pathway(
+ catalog_pathway: CatalogPathway | CatalogPathway.ID,
+ *,
+ title: str | None = None,
+ category: PathwayCategory | None = None,
+ description: str | None = None,
+) -> None:
+ """
+ Update a `CatalogPathway`. Pass `None` for a field to leave it unchanged.
+
+ None of these edits create a new content version: catalog copy and the Pathway definition change at different rates
+ and are edited by different people, which is the whole point of the split.
+
+ ⚠️ Does not check permissions.
+ """
+ if isinstance(catalog_pathway, CatalogPathway):
+ cp = catalog_pathway
+ else:
+ cp = CatalogPathway.objects.get(pk=catalog_pathway)
+
+ update_fields = []
+ for field_name, value in (
+ ("title", title),
+ ("category", category),
+ ("description", description),
+ ):
+ if value is not None:
+ setattr(cp, field_name, value)
+ update_fields.append(field_name)
+ if update_fields:
+ cp.save(update_fields=update_fields + ["modified"])
+
+
+def delete_catalog_pathway(catalog_pathway: CatalogPathway | CatalogPathway.ID) -> None:
+ """
+ Delete a `CatalogPathway`.
+
+ This fails with a `ProtectedError` while anyone is or ever was enrolled in it, including learners who have since
+ unenrolled, so that deleting the wrong pathway can't wipe out enrollment history; delete those enrollments first if
+ that's really intended. It also fails while any Pathway content still implements it, because that link is a
+ `PROTECT` foreign key on the content side.
+
+ ⚠️ Does not check permissions.
+ """
+ if isinstance(catalog_pathway, CatalogPathway):
+ cp = catalog_pathway
+ else:
+ cp = CatalogPathway.objects.get(pk=catalog_pathway)
+ cp.delete()
+
+
+def enroll_in_pathway(user_id: int, catalog_pathway: CatalogPathway | CatalogPathway.ID) -> PathwayEnrollment:
+ """
+ Enroll a learner in a `CatalogPathway`, or return their existing active enrollment.
+
+ If the learner had previously unenrolled, their existing row is reactivated rather than replaced, so the original
+ enrollment date is kept.
+
+ Enrollment does not pin a content version: progress is always evaluated against whatever is published at the time,
+ so that authoring changes reach learners who are already enrolled.
+
+ Raises `CatalogPathway.DoesNotExist` if given the ID of a pathway that doesn't exist.
+
+ ⚠️ Does not check permissions.
+ """
+ if not isinstance(catalog_pathway, CatalogPathway):
+ catalog_pathway = CatalogPathway.objects.get(pk=catalog_pathway)
+ enrollment, created = PathwayEnrollment.objects.get_or_create(user_id=user_id, catalog_pathway=catalog_pathway)
+ if not created and not enrollment.is_active:
+ enrollment.is_active = True
+ enrollment.save(update_fields=["is_active", "modified"])
+ return enrollment
+
+
+def unenroll_from_pathway(user_id: int, catalog_pathway: CatalogPathway | CatalogPathway.ID) -> None:
+ """
+ Unenroll a learner from a `CatalogPathway`. A no-op if they aren't enrolled.
+
+ The enrollment row is deactivated, not deleted.
+
+ ⚠️ Does not check permissions.
+ """
+ pathway_id = catalog_pathway.id if isinstance(catalog_pathway, CatalogPathway) else catalog_pathway
+ PathwayEnrollment.objects.filter(user_id=user_id, catalog_pathway_id=pathway_id, is_active=True).update(
+ is_active=False, modified=timezone.now()
+ )
+
+
+def is_enrolled_in_pathway(user_id: int, catalog_pathway: CatalogPathway | CatalogPathway.ID) -> bool:
+ """
+ Check whether this learner is actively enrolled in this `CatalogPathway`.
+
+ ⚠️ Does not check permissions.
+ """
+ pathway_id = catalog_pathway.id if isinstance(catalog_pathway, CatalogPathway) else catalog_pathway
+ return PathwayEnrollment.objects.filter(user_id=user_id, catalog_pathway_id=pathway_id, is_active=True).exists()
+
+
+def get_pathway_enrollments(user_id: int, *, include_inactive: bool = False) -> QuerySet[PathwayEnrollment]:
+ """
+ Get a learner's pathway enrollments, most recent first.
+
+ Only active enrollments are returned unless ``include_inactive`` is set. Each enrollment's catalog pathway comes
+ with it, along with the pathway's org and category, so a dashboard can show them without further queries.
+
+ ⚠️ Does not check permissions or visibility rules.
+ """
+ enrollments = PathwayEnrollment.objects.filter(user_id=user_id).select_related(
+ "catalog_pathway__org", "catalog_pathway__category"
+ )
+ if not include_inactive:
+ enrollments = enrollments.filter(is_active=True)
+ return enrollments
diff --git a/src/openedx_catalog/migrations/0002_pathways.py b/src/openedx_catalog/migrations/0002_pathways.py
new file mode 100644
index 000000000..ba500f3c6
--- /dev/null
+++ b/src/openedx_catalog/migrations/0002_pathways.py
@@ -0,0 +1,289 @@
+"""
+Create the catalog half of a Pathway: PathwayCategory, CatalogPathway, and PathwayEnrollment.
+
+Every CatalogPathway must have a category. Rather than falling back to the word "Pathway" in code, we ship a database
+row with that name, so that the behavior is uniform and operators can rename it or add categories of their own without a
+code change (ADR 0007, decision 2). The default row is created right after its table and before CatalogPathway exists.
+Because Django unapplies operations in reverse order, the reverse step deletes that row only after the CatalogPathway
+table is already gone, so nothing can still reference it.
+"""
+
+import re
+
+import django.core.validators
+import django.db.models.deletion
+import django.db.models.functions.text
+import django.db.models.lookups
+from django.conf import settings
+from django.db import migrations, models
+
+import openedx_django_lib.fields
+import openedx_django_lib.validators
+
+
+def create_initial_pathway_category(apps, schema_editor):
+ """Create the initial category."""
+ PathwayCategory = apps.get_model("openedx_catalog", "PathwayCategory")
+ PathwayCategory.objects.get_or_create(
+ category_code="pathway",
+ defaults={"title": "Pathway"},
+ )
+
+
+def delete_initial_pathway_category(apps, schema_editor):
+ """Remove the initial category on reverse."""
+ PathwayCategory = apps.get_model("openedx_catalog", "PathwayCategory")
+ PathwayCategory.objects.filter(category_code="pathway").delete()
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ("openedx_catalog", "0001_initial"),
+ ("organizations", "0004_auto_20230727_2054"),
+ migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+ ]
+
+ operations = [
+ migrations.CreateModel(
+ name="PathwayCategory",
+ fields=[
+ (
+ "id",
+ models.AutoField(
+ editable=False,
+ help_text="The internal database ID for this pathway category. Should not be exposed to users nor in APIs.",
+ primary_key=True,
+ serialize=False,
+ verbose_name="Primary Key",
+ ),
+ ),
+ (
+ "category_code",
+ openedx_django_lib.fields.MultiCollationCharField(
+ db_collations={"mysql": "utf8mb4_bin", "sqlite": "BINARY"},
+ help_text='A stable slug identifying this category, e.g. "masters-degree". Not shown to learners.',
+ max_length=255,
+ validators=[
+ django.core.validators.RegexValidator(
+ re.compile("^[a-zA-Z0-9_.-]+\\Z"),
+ 'Enter a valid "code name" consisting of latin letters (A-Z, a-z), numbers, underscores, hyphens, or periods.',
+ "invalid",
+ )
+ ],
+ ),
+ ),
+ (
+ "title",
+ openedx_django_lib.fields.MultiCollationCharField(
+ db_collations={"mysql": "utf8mb4_unicode_ci", "sqlite": "NOCASE"},
+ help_text='The learner-facing title of this category, e.g. "Master\'s Degree". Operators may change this.',
+ max_length=255,
+ ),
+ ),
+ (
+ "title_plural",
+ openedx_django_lib.fields.MultiCollationCharField(
+ blank=True,
+ db_collations={"mysql": "utf8mb4_unicode_ci", "sqlite": "NOCASE"},
+ default="",
+ help_text='The plural of the title. Leave blank to use the title with an "s" appended, which suits most titles; set it for titles that don\'t pluralize that way, e.g. "Classes" for "Class".',
+ max_length=255,
+ ),
+ ),
+ ],
+ options={
+ "verbose_name": "Pathway Category",
+ "verbose_name_plural": "Pathway Categories",
+ "ordering": ("title",),
+ "constraints": [
+ models.UniqueConstraint(
+ django.db.models.functions.text.Lower("category_code"),
+ name="oex_catalog_pathwaycategory_code_uniq_ci",
+ ),
+ models.CheckConstraint(
+ condition=django.db.models.lookups.Regex(models.F("category_code"), "^[a-zA-Z0-9_.-]+\\Z"),
+ name="oex_catalog_pathwaycategory_code_regex",
+ violation_error_message='Enter a valid "code name" consisting of latin letters (A-Z, a-z), numbers, underscores, hyphens, or periods.',
+ ),
+ models.CheckConstraint(
+ condition=models.Q(("title__length__gt", 0)), name="oex_catalog_pathwaycategory_title_not_blank"
+ ),
+ ],
+ },
+ ),
+ migrations.RunPython(
+ create_initial_pathway_category,
+ reverse_code=delete_initial_pathway_category,
+ ),
+ migrations.CreateModel(
+ name="CatalogPathway",
+ fields=[
+ (
+ "id",
+ models.BigAutoField(
+ editable=False,
+ help_text="The internal database ID for this catalog pathway. Should not be exposed to users nor in APIs.",
+ primary_key=True,
+ serialize=False,
+ verbose_name="Primary Key",
+ ),
+ ),
+ (
+ "pathway_code",
+ openedx_django_lib.fields.MultiCollationCharField(
+ db_collations={"mysql": "utf8mb4_bin", "sqlite": "BINARY"},
+ help_text='The pathway code/number, e.g. "DataScience2026".',
+ max_length=255,
+ validators=[
+ django.core.validators.RegexValidator(
+ re.compile("^[a-zA-Z0-9_.-]+\\Z"),
+ 'Enter a valid "code name" consisting of latin letters (A-Z, a-z), numbers, underscores, hyphens, or periods.',
+ "invalid",
+ )
+ ],
+ ),
+ ),
+ (
+ "created",
+ models.DateTimeField(
+ auto_now_add=True, validators=[openedx_django_lib.validators.validate_utc_datetime]
+ ),
+ ),
+ (
+ "modified",
+ models.DateTimeField(
+ auto_now=True,
+ help_text="When the catalog fields of this pathway were last edited. Unrelated to its content.",
+ validators=[openedx_django_lib.validators.validate_utc_datetime],
+ ),
+ ),
+ (
+ "title",
+ openedx_django_lib.fields.MultiCollationCharField(
+ blank=True,
+ db_collations={"mysql": "utf8mb4_unicode_ci", "sqlite": "NOCASE"},
+ help_text='The full title (display name) of this pathway, e.g. "Data Science Professional Certificate". Leave blank to use the pathway code as the title.',
+ max_length=255,
+ ),
+ ),
+ (
+ "description",
+ openedx_django_lib.fields.MultiCollationTextField(
+ blank=True,
+ db_collations={"mysql": "utf8mb4_unicode_ci", "sqlite": "NOCASE"},
+ default="",
+ help_text="The description shown to learners browsing the catalog.",
+ max_length=10000,
+ ),
+ ),
+ (
+ "org",
+ models.ForeignKey(
+ on_delete=django.db.models.deletion.PROTECT,
+ related_name="catalog_pathways",
+ to="organizations.organization",
+ ),
+ ),
+ (
+ "category",
+ models.ForeignKey(
+ help_text="The learner-facing kind of pathway this is.",
+ on_delete=django.db.models.deletion.PROTECT,
+ related_name="pathways",
+ to="openedx_catalog.pathwaycategory",
+ ),
+ ),
+ ],
+ options={
+ "verbose_name": "Catalog Pathway",
+ "verbose_name_plural": "Catalog Pathways",
+ "ordering": ("-created",),
+ },
+ ),
+ migrations.CreateModel(
+ name="PathwayEnrollment",
+ fields=[
+ (
+ "id",
+ models.BigAutoField(
+ editable=False,
+ help_text="The internal database ID for this enrollment. Should not be exposed to users nor in APIs.",
+ primary_key=True,
+ serialize=False,
+ verbose_name="Primary Key",
+ ),
+ ),
+ (
+ "created",
+ models.DateTimeField(
+ auto_now_add=True, validators=[openedx_django_lib.validators.validate_utc_datetime]
+ ),
+ ),
+ (
+ "modified",
+ models.DateTimeField(auto_now=True, validators=[openedx_django_lib.validators.validate_utc_datetime]),
+ ),
+ (
+ "is_active",
+ models.BooleanField(
+ default=True,
+ help_text="False once the learner has unenrolled. The row is kept so re-enrolling reuses it.",
+ ),
+ ),
+ (
+ "catalog_pathway",
+ models.ForeignKey(
+ on_delete=django.db.models.deletion.PROTECT,
+ related_name="enrollments",
+ to="openedx_catalog.catalogpathway",
+ ),
+ ),
+ (
+ "user",
+ models.ForeignKey(
+ on_delete=django.db.models.deletion.CASCADE,
+ related_name="pathway_enrollments",
+ to=settings.AUTH_USER_MODEL,
+ ),
+ ),
+ ],
+ options={
+ "verbose_name": "Pathway Enrollment",
+ "verbose_name_plural": "Pathway Enrollments",
+ "ordering": ("-created",),
+ },
+ ),
+ migrations.AddIndex(
+ model_name="catalogpathway",
+ index=models.Index(fields=["org", "pathway_code"], name="openedx_cat_org_id_037ff8_idx"),
+ ),
+ migrations.AddConstraint(
+ model_name="catalogpathway",
+ constraint=models.UniqueConstraint(
+ models.F("org"),
+ django.db.models.functions.text.Lower("pathway_code"),
+ name="oex_catalog_catalogpathway_org_code_uniq_ci",
+ ),
+ ),
+ migrations.AddConstraint(
+ model_name="catalogpathway",
+ constraint=models.CheckConstraint(
+ condition=django.db.models.lookups.Regex(models.F("pathway_code"), "^[a-zA-Z0-9_.-]+\\Z"),
+ name="oex_catalog_catalogpathway_pathway_code_regex",
+ violation_error_message='Enter a valid "code name" consisting of latin letters (A-Z, a-z), numbers, underscores, hyphens, or periods.',
+ ),
+ ),
+ migrations.AddConstraint(
+ model_name="catalogpathway",
+ constraint=models.CheckConstraint(
+ condition=models.Q(("title__length__gt", 0)), name="oex_catalog_catalogpathway_title_not_blank"
+ ),
+ ),
+ migrations.AddConstraint(
+ model_name="pathwayenrollment",
+ constraint=models.UniqueConstraint(
+ fields=("user", "catalog_pathway"), name="oex_catalog_pathwayenrollment_uniq_user_pathway"
+ ),
+ ),
+ ]
diff --git a/src/openedx_catalog/models/__init__.py b/src/openedx_catalog/models/__init__.py
index 31da59962..513141541 100644
--- a/src/openedx_catalog/models/__init__.py
+++ b/src/openedx_catalog/models/__init__.py
@@ -3,4 +3,7 @@
"""
from .catalog_course import CatalogCourse
+from .catalog_pathway import CatalogPathway
from .course_run import CourseRun
+from .pathway_category import PathwayCategory
+from .pathway_enrollment import PathwayEnrollment
diff --git a/src/openedx_catalog/models/catalog_pathway.py b/src/openedx_catalog/models/catalog_pathway.py
new file mode 100644
index 000000000..410ef14a1
--- /dev/null
+++ b/src/openedx_catalog/models/catalog_pathway.py
@@ -0,0 +1,181 @@
+"""
+CatalogPathway model
+"""
+
+import logging
+from typing import NewType
+
+from django.contrib import admin
+from django.db import models
+from django.db.models.functions import Length, Lower
+from django.utils.translation import gettext_lazy as _
+from organizations.models import Organization # type: ignore[import]
+
+from openedx_django_lib.fields import (
+ MultiCollationTextField,
+ TypedBigAutoField,
+ case_insensitive_char_field,
+ code_field,
+ code_field_check,
+)
+from openedx_django_lib.validators import validate_utc_datetime
+
+from .pathway_category import PathwayCategory
+
+log = logging.getLogger(__name__)
+
+# Make 'length' available for CHECK constraints. OK if this is called multiple times.
+models.CharField.register_lookup(Length)
+
+
+class CatalogPathway(models.Model):
+ """
+ The learner-browsable, enrollable half of a Pathway.
+
+ A Pathway is split in two (see the openedx_learning ADR 0007). This model is the catalog half: the display name, the
+ description shown in the catalog, and the `PathwayCategory`. It is **not versioned**, because marketing copy is
+ revised frequently and casually and versioning it would be pure overhead.
+
+ The other half - the *definition* of the Pathway, meaning its Items and completion criteria - lives in
+ `openedx_learning.applets.pathways` and *is* versioned, so that progress and credentials can be judged against the
+ definition that was in effect at the time.
+
+ The link between the two halves lives on the content side (`Pathway.catalog_pathway`, one-to-one), never here: this
+ app stays unaware of content, so that changes in how content is represented never reach it (see the openedx_catalog
+ ADR 0001, decision 2). A `CatalogPathway` therefore cannot, on its own, tell you which content implements it;
+ queries in that direction start from the content side.
+
+ A `CatalogPathway` may exist before any content implements it, in the same way that a `CatalogCourse` may exist as a
+ marketing placeholder for a course that has no content yet.
+
+ Like `CatalogCourse`, this model is intentionally minimal. Additional catalog-side fields should generally go in a
+ related model in your own app, with a `ForeignKey` or `OneToOneField` to this one.
+
+ .. no_pii:
+ """
+
+ CatalogPathwayID = NewType("CatalogPathwayID", int)
+ type ID = CatalogPathwayID
+
+ class IDField(TypedBigAutoField[ID]): # Boilerplate for fully-typed ID field.
+ pass
+
+ id = IDField(
+ primary_key=True,
+ verbose_name=_("Primary Key"),
+ help_text=_("The internal database ID for this catalog pathway. Should not be exposed to users nor in APIs."),
+ editable=False,
+ )
+ org = models.ForeignKey(
+ Organization,
+ on_delete=models.PROTECT,
+ null=False,
+ related_name="catalog_pathways",
+ )
+ pathway_code = code_field(
+ unicode=False,
+ help_text=_('The pathway code/number, e.g. "DataScience2026".'),
+ )
+ created = models.DateTimeField(
+ auto_now_add=True,
+ validators=[validate_utc_datetime],
+ editable=False,
+ )
+ # This reflects edits to the *catalog* fields on this row only (title, category, description). It says nothing about
+ # when the pathway's *definition* - its Items and completion criteria - last changed. That happens on the content
+ # side and is versioned there; ask `openedx_learning.api` for it.
+ modified = models.DateTimeField(
+ auto_now=True,
+ validators=[validate_utc_datetime],
+ editable=False,
+ help_text=_("When the catalog fields of this pathway were last edited. Unrelated to its content."),
+ )
+ title = case_insensitive_char_field(
+ max_length=255,
+ blank=True, # Only allowed to be blank temporarily when creating a new instance in the Django admin form.
+ help_text=_(
+ 'The full title (display name) of this pathway, e.g. "Data Science Professional Certificate". '
+ "Leave blank to use the pathway code as the title."
+ ),
+ )
+ category = models.ForeignKey(
+ PathwayCategory,
+ on_delete=models.PROTECT,
+ null=False,
+ related_name="pathways",
+ help_text=_("The learner-facing kind of pathway this is."),
+ )
+ description = MultiCollationTextField(
+ blank=True,
+ null=False,
+ default="",
+ max_length=10_000,
+ # We don't expect to sort by this column, but we may want case-insensitive searches over it.
+ db_collations={
+ "sqlite": "NOCASE",
+ "mysql": "utf8mb4_unicode_ci",
+ },
+ help_text=_("The description shown to learners browsing the catalog."),
+ )
+
+ # 🛑 Avoid adding additional fields here. Anything that describes what a learner must *do* belongs on the content
+ # side, where it is versioned. Anything else catalog-related should go in a related model in your own app.
+
+ @property
+ @admin.display(ordering="org__short_name")
+ def org_code(self) -> str:
+ """
+ Get the org code (Organization short_name) of this pathway, e.g. "MITx".
+ """
+ return self.org.short_name
+
+ @org_code.setter
+ def org_code(self, org_code: str) -> None:
+ """
+ Convenience method to set the related organization using its short_name.
+ """
+ # As with CatalogCourse, we don't use `get_organization_by_short_name` because it filters for active orgs only,
+ # and we need to allow inactive orgs to support historical data and backfills.
+ self.org = Organization.objects.get(short_name__iexact=org_code)
+
+ @property
+ def key_str(self) -> str:
+ """
+ A string key that can be used to identify this catalog pathway in URLs or APIs.
+
+ As with `CatalogCourse.key_str`, this may become based on an editable `SlugField` or an opaque key in the
+ future, so don't assume it never changes.
+ """
+ return f"catalog-pathway:{self.org_code}:{self.pathway_code}"
+
+ def clean(self) -> None:
+ """Validate/normalize fields when edited via Django admin."""
+ # Set a default value for title:
+ if not self.title:
+ self.title = self.pathway_code
+
+ def save(self, *args, **kwargs):
+ """Save the model, with some defaults and validation."""
+ self.clean()
+ super().save(*args, **kwargs)
+
+ def __str__(self) -> str:
+ return f"{self.title} ({self.org_code} {self.pathway_code})"
+
+ class Meta:
+ verbose_name = _("Catalog Pathway")
+ verbose_name_plural = _("Catalog Pathways")
+ ordering = ("-created",)
+ indexes = [
+ # We need fast lookups by (org, pathway_code) pairs. We generally want this lookup to be case sensitive.
+ models.Index(fields=["org", "pathway_code"]),
+ ]
+ constraints = [
+ # The pathway_code must be case-insensitively unique per org:
+ models.UniqueConstraint("org", Lower("pathway_code"), name="oex_catalog_catalogpathway_org_code_uniq_ci"),
+ code_field_check("pathway_code", name="oex_catalog_catalogpathway_pathway_code_regex", unicode=False),
+ # Enforce at the DB level that these required fields are not blank:
+ models.CheckConstraint(
+ condition=models.Q(title__length__gt=0), name="oex_catalog_catalogpathway_title_not_blank"
+ ),
+ ]
diff --git a/src/openedx_catalog/models/pathway_category.py b/src/openedx_catalog/models/pathway_category.py
new file mode 100644
index 000000000..7c7b98a6c
--- /dev/null
+++ b/src/openedx_catalog/models/pathway_category.py
@@ -0,0 +1,89 @@
+"""
+PathwayCategory model
+"""
+
+import logging
+from typing import NewType
+
+from django.db import models
+from django.db.models.functions import Length, Lower
+from django.utils.translation import gettext_lazy as _
+
+from openedx_django_lib.fields import TypedAutoField, case_insensitive_char_field, code_field, code_field_check
+
+log = logging.getLogger(__name__)
+
+# Make 'length' available for CHECK constraints. OK if this is called multiple times.
+models.CharField.register_lookup(Length)
+
+
+class PathwayCategory(models.Model):
+ """
+ A student-facing label for a kind of Pathway.
+
+ Learners are shown the category ("Master's Degree", "Annual Training") rather than the word "Pathway". In authoring
+ contexts - Studio, Django admin, code, docs - the terminology stays "Pathway", with the category shown explicitly;
+ relabelling is a learner-facing concern of the catalog side only.
+
+ The ``category_code`` is the stable identifier that code and imports may key off. The ``title`` is what learners
+ see, and operators are free to change it.
+
+ Where learners see several pathways of a category together ("3 Master's Degrees"), use `get_title_plural`.
+
+ .. no_pii:
+ """
+
+ PathwayCategoryID = NewType("PathwayCategoryID", int)
+ type ID = PathwayCategoryID
+
+ # A 32-bit key is plenty: an instance has a handful of categories, not millions.
+ class IDField(TypedAutoField[ID]): # Boilerplate for fully-typed ID field.
+ pass
+
+ id = IDField(
+ primary_key=True,
+ verbose_name=_("Primary Key"),
+ help_text=_("The internal database ID for this pathway category. Should not be exposed to users nor in APIs."),
+ editable=False,
+ )
+ category_code = code_field(
+ unicode=False,
+ help_text=_('A stable slug identifying this category, e.g. "masters-degree". Not shown to learners.'),
+ )
+ title = case_insensitive_char_field(
+ max_length=255,
+ blank=False,
+ help_text=_('The learner-facing title of this category, e.g. "Master\'s Degree". Operators may change this.'),
+ )
+ title_plural = case_insensitive_char_field(
+ max_length=255,
+ blank=True,
+ default="",
+ help_text=_(
+ 'The plural of the title. Leave blank to use the title with an "s" appended, which suits most titles; '
+ 'set it for titles that don\'t pluralize that way, e.g. "Classes" for "Class".'
+ ),
+ )
+
+ def get_title_plural(self) -> str:
+ """
+ Get the plural of the title: ``title_plural`` if set, and otherwise ``title`` with an "s" appended.
+ """
+ return self.title_plural or f"{self.title}s"
+
+ def __str__(self) -> str:
+ return self.title
+
+ class Meta:
+ verbose_name = _("Pathway Category")
+ verbose_name_plural = _("Pathway Categories")
+ ordering = ("title",)
+ constraints = [
+ # The category_code must be case-insensitively unique:
+ models.UniqueConstraint(Lower("category_code"), name="oex_catalog_pathwaycategory_code_uniq_ci"),
+ code_field_check("category_code", name="oex_catalog_pathwaycategory_code_regex", unicode=False),
+ # Enforce at the DB level that this required field is not blank:
+ models.CheckConstraint(
+ condition=models.Q(title__length__gt=0), name="oex_catalog_pathwaycategory_title_not_blank"
+ ),
+ ]
diff --git a/src/openedx_catalog/models/pathway_enrollment.py b/src/openedx_catalog/models/pathway_enrollment.py
new file mode 100644
index 000000000..dd95c6d03
--- /dev/null
+++ b/src/openedx_catalog/models/pathway_enrollment.py
@@ -0,0 +1,90 @@
+"""
+PathwayEnrollment model
+"""
+
+import logging
+from typing import NewType
+
+from django.conf import settings
+from django.db import models
+from django.utils.translation import gettext_lazy as _
+
+from openedx_django_lib.fields import TypedBigAutoField
+from openedx_django_lib.validators import validate_utc_datetime
+
+from .catalog_pathway import CatalogPathway
+
+log = logging.getLogger(__name__)
+
+
+class PathwayEnrollment(models.Model):
+ """
+ Ties a learner to a `CatalogPathway`.
+
+ Enrollment is against the *catalog* half of a Pathway, never against a version of its content. Progress is evaluated
+ against whichever content version is published at the time of evaluation, not against a version frozen at enrollment
+ time, so that authoring changes reach learners who are already enrolled. That is why this model pins no version. See
+ the openedx_learning ADR 0007, decision 5.
+
+ Unenrolling sets ``is_active`` to False rather than deleting the row, so that "unenrolled" can be told apart from
+ "never enrolled" and the original enrollment date survives. Re-enrolling reactivates the same row.
+
+ .. no_pii:
+ """
+
+ PathwayEnrollmentID = NewType("PathwayEnrollmentID", int)
+ type ID = PathwayEnrollmentID
+
+ class IDField(TypedBigAutoField[ID]): # Boilerplate for fully-typed ID field.
+ pass
+
+ id = IDField(
+ primary_key=True,
+ verbose_name=_("Primary Key"),
+ help_text=_("The internal database ID for this enrollment. Should not be exposed to users nor in APIs."),
+ editable=False,
+ )
+ user = models.ForeignKey(
+ settings.AUTH_USER_MODEL,
+ on_delete=models.CASCADE,
+ null=False,
+ related_name="pathway_enrollments",
+ )
+ # PROTECT, so that deleting the wrong catalog pathway can't silently wipe out everyone's enrollment history; the
+ # enrollments have to be deleted deliberately first.
+ catalog_pathway = models.ForeignKey(
+ CatalogPathway,
+ on_delete=models.PROTECT,
+ null=False,
+ related_name="enrollments",
+ )
+ created = models.DateTimeField(
+ auto_now_add=True,
+ validators=[validate_utc_datetime],
+ editable=False,
+ )
+ modified = models.DateTimeField(
+ auto_now=True,
+ validators=[validate_utc_datetime],
+ editable=False,
+ )
+ is_active = models.BooleanField(
+ default=True,
+ help_text=_("False once the learner has unenrolled. The row is kept so re-enrolling reuses it."),
+ )
+
+ def __str__(self) -> str:
+ return f"{self.user} in {self.catalog_pathway}"
+
+ class Meta:
+ verbose_name = _("Pathway Enrollment")
+ verbose_name_plural = _("Pathway Enrollments")
+ ordering = ("-created",)
+ constraints = [
+ # There is only ever one row per (learner, pathway) pair; unenrolling flips `is_active` rather than
+ # deleting or adding a row.
+ models.UniqueConstraint(
+ fields=["user", "catalog_pathway"],
+ name="oex_catalog_pathwayenrollment_uniq_user_pathway",
+ ),
+ ]
diff --git a/src/openedx_catalog/models_api.py b/src/openedx_catalog/models_api.py
index 2836a2aa1..90a67b1a3 100644
--- a/src/openedx_catalog/models_api.py
+++ b/src/openedx_catalog/models_api.py
@@ -7,4 +7,4 @@
"""
# pylint: disable=unused-import
-from .models import CatalogCourse, CourseRun
+from .models import CatalogCourse, CatalogPathway, CourseRun, PathwayCategory, PathwayEnrollment
diff --git a/tests/openedx_catalog/test_api.py b/tests/openedx_catalog/test_api.py
index 05f26d161..52d48e395 100644
--- a/tests/openedx_catalog/test_api.py
+++ b/tests/openedx_catalog/test_api.py
@@ -266,7 +266,7 @@ def test_create_course_run_for_modulestore_course_with_existing_org_different_ca
# Verify that a warning was logged. We actually get two warnings - one from the API and one from model.clean():
assert caplog.record_tuples == [
(
- "openedx_catalog.api_impl",
+ "openedx_catalog.api_impl.courses",
logging.WARN,
'The course with ID "course-v1:NewOrg+Test+2026" does not match its Organization.short_name "nEWoRG"',
),
@@ -348,7 +348,7 @@ def test_create_course_run_for_modulestore_course_run_that_exists(caplog: pytest
# Verify that a warning was logged:
assert caplog.record_tuples == [
(
- "openedx_catalog.api_impl",
+ "openedx_catalog.api_impl.courses",
logging.WARN,
'Expected to create CourseRun for "course-v1:NewOrg+Test+2026" but it already existed.',
),
diff --git a/tests/openedx_catalog/test_pathway_admin.py b/tests/openedx_catalog/test_pathway_admin.py
new file mode 100644
index 000000000..e0ad13680
--- /dev/null
+++ b/tests/openedx_catalog/test_pathway_admin.py
@@ -0,0 +1,21 @@
+"""
+Tests of the Django admin for the catalog half of Pathways.
+"""
+
+import pytest
+from django.contrib import admin
+
+from openedx_catalog.admin import PathwayCategoryAdmin
+from openedx_catalog.models import PathwayCategory
+
+pytestmark = pytest.mark.django_db
+
+
+def test_category_list_shows_the_plural_learners_see() -> None:
+ """The derived plural when none is set, so operators can spot titles that need an explicit one."""
+ category = PathwayCategory.objects.get(category_code="pathway")
+ model_admin = PathwayCategoryAdmin(PathwayCategory, admin.site)
+ assert model_admin.title_plural_display(category) == "Pathways"
+
+ category.title_plural = "Pathways of Study"
+ assert model_admin.title_plural_display(category) == "Pathways of Study"
diff --git a/tests/openedx_catalog/test_pathway_api.py b/tests/openedx_catalog/test_pathway_api.py
new file mode 100644
index 000000000..be64867e8
--- /dev/null
+++ b/tests/openedx_catalog/test_pathway_api.py
@@ -0,0 +1,331 @@
+"""
+Tests of the catalog-side Pathway API.
+"""
+# pylint: disable=unused-argument
+
+from datetime import datetime, timezone
+
+import pytest
+from django.contrib.auth import get_user_model
+from django.db import transaction
+from django.db.models import ProtectedError
+from django.db.utils import IntegrityError
+from freezegun import freeze_time
+from organizations.api import ensure_organization # type: ignore[import]
+
+from openedx_catalog import api as catalog_api
+from openedx_catalog.models import CatalogPathway, PathwayCategory
+
+User = get_user_model()
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture(name="org1")
+def _org1() -> None:
+ """Create an "Org1" organization for use in these tests"""
+ ensure_organization("Org1")
+
+
+@pytest.fixture(name="category")
+def _category() -> PathwayCategory:
+ """The category created by the initial migration"""
+ return catalog_api.get_pathway_category("pathway")
+
+
+@pytest.fixture(name="data_science")
+def _data_science(org1, category) -> CatalogPathway:
+ """Create a CatalogPathway for use in these tests"""
+ return catalog_api.create_catalog_pathway(
+ org_code="Org1",
+ pathway_code="DataScience",
+ category=category,
+ title="Data Science Professional Certificate",
+ description="Learn data science.",
+ )
+
+
+@pytest.fixture(name="learner")
+def _learner():
+ """Create a learner for use in these tests"""
+ return User.objects.create(username="learner", email="learner@example.com")
+
+
+def test_get_pathway_category() -> None:
+ """Categories are looked up by their stable code, not by the learner-facing name."""
+ masters = catalog_api.create_pathway_category(category_code="masters-degree", title="Master's Degree")
+ assert catalog_api.get_pathway_category("masters-degree") == masters
+ with pytest.raises(PathwayCategory.DoesNotExist):
+ catalog_api.get_pathway_category("Master's Degree")
+
+
+def test_get_pathway_categories(category) -> None:
+ """All categories, ordered by title, including the one created by the initial migration."""
+ masters = catalog_api.create_pathway_category(category_code="masters-degree", title="Master's Degree")
+ assert list(catalog_api.get_pathway_categories()) == [masters, category]
+
+
+def test_create_pathway_category() -> None:
+ created = catalog_api.create_pathway_category(category_code="class", title="Class", title_plural="Classes")
+ category = catalog_api.get_pathway_category("class")
+ assert category == created
+ assert (category.title, category.get_title_plural()) == ("Class", "Classes")
+
+
+def test_create_pathway_category_with_a_taken_code() -> None:
+ """Codes are case-insensitively unique, so the initial "pathway" code can't be reused in any case."""
+ with pytest.raises(IntegrityError), transaction.atomic():
+ catalog_api.create_pathway_category(category_code="Pathway", title="Another Pathway")
+
+
+def test_update_pathway_category() -> None:
+ """Only the fields passed are changed, and an empty plural goes back to deriving it from the title."""
+ catalog_api.create_pathway_category(category_code="class", title="Class")
+
+ catalog_api.update_pathway_category("class", title_plural="Classes")
+ category = catalog_api.get_pathway_category("class")
+ assert (category.title, category.get_title_plural()) == ("Class", "Classes")
+
+ catalog_api.update_pathway_category(category, title="Program", title_plural="")
+ category = catalog_api.get_pathway_category("class")
+ assert (category.title, category.get_title_plural()) == ("Program", "Programs")
+
+
+def test_update_pathway_category_with_nothing_to_change(django_assert_num_queries) -> None:
+ """Passing no fields is a no-op that doesn't touch the database."""
+ category = catalog_api.create_pathway_category(category_code="class", title="Class")
+ with django_assert_num_queries(0):
+ catalog_api.update_pathway_category(category)
+
+
+def test_delete_pathway_category() -> None:
+ """The category may be given as an instance or by its code."""
+ catalog_api.create_pathway_category(category_code="class", title="Class")
+ catalog_api.delete_pathway_category("class")
+ catalog_api.delete_pathway_category(catalog_api.get_pathway_category("pathway"))
+ assert not catalog_api.get_pathway_categories().exists()
+ with pytest.raises(PathwayCategory.DoesNotExist):
+ catalog_api.delete_pathway_category("class")
+
+
+def test_delete_pathway_category_in_use(data_science, category) -> None:
+ """A category in use can't be deleted until its pathways are moved to another one."""
+ with pytest.raises(ProtectedError), transaction.atomic():
+ catalog_api.delete_pathway_category(category)
+
+ masters = catalog_api.create_pathway_category(category_code="masters-degree", title="Master's Degree")
+ catalog_api.update_catalog_pathway(data_science, category=masters)
+ catalog_api.delete_pathway_category(category)
+ assert list(catalog_api.get_pathway_categories()) == [masters]
+
+
+def test_create_catalog_pathway(org1) -> None:
+ """Operators can add categories of their own, and a blank title falls back to the code."""
+ masters = catalog_api.create_pathway_category(category_code="masters-degree", title="Master's Degree")
+ pathway = catalog_api.create_catalog_pathway(org_code="Org1", pathway_code="CompSci", category=masters)
+ assert pathway.category == masters
+ assert pathway.title == "CompSci"
+
+
+def test_get_catalog_pathway(data_science) -> None:
+ """A catalog pathway can be looked up by pk, key string, or org + code."""
+ assert catalog_api.get_catalog_pathway(pk=data_science.id) == data_science
+ assert catalog_api.get_catalog_pathway(key_str=data_science.key_str) == data_science
+ assert catalog_api.get_catalog_pathway(org_code="Org1", pathway_code="DataScience") == data_science
+ with pytest.raises(CatalogPathway.DoesNotExist):
+ catalog_api.get_catalog_pathway(org_code="Org1", pathway_code="Nope")
+
+
+@pytest.fixture(name="three_pathways")
+def _three_pathways(org1, category) -> tuple[CatalogPathway, CatalogPathway, CatalogPathway]:
+ """Create three catalog pathways across two orgs and two categories, a day apart, oldest first."""
+ ensure_organization("Org2")
+ masters = catalog_api.create_pathway_category(category_code="masters-degree", title="Master's Degree")
+ with freeze_time(datetime(2026, 1, 1, tzinfo=timezone.utc)):
+ data_science = catalog_api.create_catalog_pathway(
+ org_code="Org1", pathway_code="DataScience", category=category
+ )
+ with freeze_time(datetime(2026, 1, 2, tzinfo=timezone.utc)):
+ comp_sci = catalog_api.create_catalog_pathway(org_code="Org1", pathway_code="CompSci", category=masters)
+ with freeze_time(datetime(2026, 1, 3, tzinfo=timezone.utc)):
+ history = catalog_api.create_catalog_pathway(org_code="Org2", pathway_code="History", category=masters)
+ return data_science, comp_sci, history
+
+
+def test_get_catalog_pathways(three_pathways) -> None:
+ """All catalog pathways, most recently created first."""
+ data_science, comp_sci, history = three_pathways
+ assert list(catalog_api.get_catalog_pathways()) == [history, comp_sci, data_science]
+
+
+def test_get_catalog_pathways_by_org_and_category(three_pathways) -> None:
+ """The filters combine, and an unknown org or category matches nothing rather than failing."""
+ data_science, comp_sci, history = three_pathways
+ assert list(catalog_api.get_catalog_pathways(org_code="Org1")) == [comp_sci, data_science]
+ assert list(catalog_api.get_catalog_pathways(category_code="masters-degree")) == [history, comp_sci]
+ assert list(catalog_api.get_catalog_pathways(org_code="Org1", category_code="masters-degree")) == [comp_sci]
+ assert list(catalog_api.get_catalog_pathways(category_code="pathway")) == [data_science]
+ assert not catalog_api.get_catalog_pathways(org_code="NoSuchOrg").exists()
+ assert not catalog_api.get_catalog_pathways(category_code="no-such-category").exists()
+
+
+def test_get_catalog_pathways_includes_org_and_category(three_pathways, django_assert_num_queries) -> None:
+ """A listing can show each pathway's org and category in a single query, rather than a query per pathway."""
+ with django_assert_num_queries(1):
+ labels = [(p.org_code, p.category.get_title_plural()) for p in catalog_api.get_catalog_pathways()]
+ assert labels == [("Org2", "Master's Degrees"), ("Org1", "Master's Degrees"), ("Org1", "Pathways")]
+
+
+def test_get_catalog_pathway_includes_org_and_category(three_pathways, django_assert_num_queries) -> None:
+ """Whichever way the pathway is looked up."""
+ _data_science, comp_sci, _history = three_pathways
+ lookups = [
+ {"pk": comp_sci.id},
+ {"key_str": comp_sci.key_str},
+ {"org_code": "Org1", "pathway_code": "CompSci"},
+ ]
+ for lookup in lookups:
+ with django_assert_num_queries(1):
+ pathway = catalog_api.get_catalog_pathway(**lookup)
+ assert (pathway.org_code, pathway.category.title) == ("Org1", "Master's Degree")
+
+
+def test_update_catalog_pathway_by_id_and_category(data_science) -> None:
+ """The pathway may be given by ID, and the category can be changed like any other catalog field."""
+ masters = catalog_api.create_pathway_category(category_code="masters-degree", title="Master's Degree")
+ catalog_api.update_catalog_pathway(data_science.id, category=masters)
+ assert catalog_api.get_catalog_pathway(pk=data_science.id).category == masters
+
+
+def test_update_catalog_pathway(data_science) -> None:
+ """Only the fields passed are changed; the rest are left alone. `modified` records the edit."""
+ edited_at = datetime(2026, 3, 1, tzinfo=timezone.utc)
+ with freeze_time(edited_at):
+ catalog_api.update_catalog_pathway(data_science, description="Learn even more data science.")
+
+ reloaded = catalog_api.get_catalog_pathway(pk=data_science.id)
+ assert reloaded.description == "Learn even more data science."
+ assert reloaded.title == "Data Science Professional Certificate"
+ assert reloaded.modified == edited_at
+
+
+def test_update_catalog_pathway_with_nothing_to_change(data_science) -> None:
+ """Passing no fields is a no-op, so `modified` is not bumped."""
+ before = catalog_api.get_catalog_pathway(pk=data_science.id).modified
+ catalog_api.update_catalog_pathway(data_science)
+ assert catalog_api.get_catalog_pathway(pk=data_science.id).modified == before
+
+
+def test_delete_catalog_pathway(data_science) -> None:
+ catalog_api.delete_catalog_pathway(data_science.id)
+ with pytest.raises(CatalogPathway.DoesNotExist):
+ catalog_api.get_catalog_pathway(pk=data_science.id)
+
+
+def test_enrollment_round_trip(data_science, learner) -> None:
+ assert not catalog_api.is_enrolled_in_pathway(learner.id, data_science)
+
+ enrollment = catalog_api.enroll_in_pathway(learner.id, data_science)
+ assert catalog_api.is_enrolled_in_pathway(learner.id, data_science)
+ assert list(catalog_api.get_pathway_enrollments(learner.id)) == [enrollment]
+
+ catalog_api.unenroll_from_pathway(learner.id, data_science)
+ assert not catalog_api.is_enrolled_in_pathway(learner.id, data_science)
+
+
+def test_get_pathway_enrollments_includes_org_and_category(three_pathways, learner, django_assert_num_queries) -> None:
+ """A learner's dashboard can show each pathway's org and category in a single query, rather than a query per row."""
+ for pathway in three_pathways:
+ catalog_api.enroll_in_pathway(learner.id, pathway)
+
+ with django_assert_num_queries(1):
+ labels = [
+ (e.catalog_pathway.org_code, e.catalog_pathway.category.title)
+ for e in catalog_api.get_pathway_enrollments(learner.id)
+ ]
+ assert sorted(labels) == [("Org1", "Master's Degree"), ("Org1", "Pathway"), ("Org2", "Master's Degree")]
+
+
+def test_enrolling_twice_is_idempotent(data_science, learner) -> None:
+ """Enrolling again returns the existing enrollment rather than failing, and doesn't touch `modified`."""
+ with freeze_time(datetime(2026, 1, 1, tzinfo=timezone.utc)):
+ first = catalog_api.enroll_in_pathway(learner.id, data_science)
+ with freeze_time(datetime(2026, 2, 1, tzinfo=timezone.utc)):
+ second = catalog_api.enroll_in_pathway(learner.id, data_science.id)
+ assert first == second
+ assert second.modified == datetime(2026, 1, 1, tzinfo=timezone.utc)
+ assert catalog_api.get_pathway_enrollments(learner.id).count() == 1
+
+
+def test_unenrolling_keeps_the_row(data_science, learner) -> None:
+ """Unenrolling deactivates rather than deletes, so history survives."""
+ enrollment = catalog_api.enroll_in_pathway(learner.id, data_science)
+ catalog_api.unenroll_from_pathway(learner.id, data_science)
+
+ assert not catalog_api.is_enrolled_in_pathway(learner.id, data_science)
+ assert catalog_api.get_pathway_enrollments(learner.id).count() == 0
+ inactive = catalog_api.get_pathway_enrollments(learner.id, include_inactive=True)
+ assert list(inactive) == [enrollment]
+ assert not inactive[0].is_active
+
+
+def test_re_enrolling_reactivates_the_same_row(data_science, learner) -> None:
+ """
+ Re-enrolling reuses the original row, keeping the original enrollment date. `modified` tracks each flip of
+ `is_active`, so it ends up as "when this learner last enrolled or unenrolled".
+ """
+ enrolled_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
+ unenrolled_at = datetime(2026, 2, 1, tzinfo=timezone.utc)
+ re_enrolled_at = datetime(2026, 3, 1, tzinfo=timezone.utc)
+
+ with freeze_time(enrolled_at):
+ original = catalog_api.enroll_in_pathway(learner.id, data_science)
+ assert original.modified == enrolled_at
+
+ with freeze_time(unenrolled_at):
+ catalog_api.unenroll_from_pathway(learner.id, data_science)
+ inactive = catalog_api.get_pathway_enrollments(learner.id, include_inactive=True).get()
+ assert not inactive.is_active
+ assert inactive.modified == unenrolled_at
+
+ with freeze_time(re_enrolled_at):
+ reactivated = catalog_api.enroll_in_pathway(learner.id, data_science)
+ assert reactivated.id == original.id
+ assert reactivated.is_active
+ assert reactivated.created == enrolled_at
+ assert reactivated.modified == re_enrolled_at
+ assert catalog_api.is_enrolled_in_pathway(learner.id, data_science)
+
+
+def test_unenrolling_when_not_enrolled_is_a_no_op(data_science, learner) -> None:
+ catalog_api.unenroll_from_pathway(learner.id, data_science) # Should not raise.
+ assert not catalog_api.is_enrolled_in_pathway(learner.id, data_science)
+ assert catalog_api.get_pathway_enrollments(learner.id, include_inactive=True).count() == 0
+
+
+def test_unenrolling_twice_does_not_bump_modified(data_science, learner) -> None:
+ """Unenrolling only touches rows that are actually active, so a repeat call leaves `modified` alone."""
+ catalog_api.enroll_in_pathway(learner.id, data_science)
+ with freeze_time(datetime(2026, 2, 1, tzinfo=timezone.utc)):
+ catalog_api.unenroll_from_pathway(learner.id, data_science)
+ with freeze_time(datetime(2026, 3, 1, tzinfo=timezone.utc)):
+ catalog_api.unenroll_from_pathway(learner.id, data_science)
+
+ row = catalog_api.get_pathway_enrollments(learner.id, include_inactive=True).get()
+ assert row.modified == datetime(2026, 2, 1, tzinfo=timezone.utc)
+
+
+def test_deleting_a_pathway_keeps_enrollment_history(data_science, learner) -> None:
+ """A pathway anyone was ever enrolled in can't be deleted, even after they've all unenrolled."""
+ catalog_api.enroll_in_pathway(learner.id, data_science)
+ catalog_api.unenroll_from_pathway(learner.id, data_science)
+
+ with pytest.raises(ProtectedError), transaction.atomic():
+ catalog_api.delete_catalog_pathway(data_science)
+ assert catalog_api.get_pathway_enrollments(learner.id, include_inactive=True).count() == 1
+
+
+def test_enrolling_in_a_missing_pathway(org1, learner) -> None:
+ """A bad ID raises DoesNotExist, like the rest of the API, rather than an IntegrityError on some backends."""
+ with pytest.raises(CatalogPathway.DoesNotExist):
+ catalog_api.enroll_in_pathway(learner.id, CatalogPathway.CatalogPathwayID(999_999))
diff --git a/tests/openedx_catalog/test_pathway_models.py b/tests/openedx_catalog/test_pathway_models.py
new file mode 100644
index 000000000..00aad1419
--- /dev/null
+++ b/tests/openedx_catalog/test_pathway_models.py
@@ -0,0 +1,229 @@
+"""
+Tests related to the catalog half of Pathways.
+"""
+# pylint: disable=unused-argument
+# mypy: disable-error-code="misc"
+# (Ignore 'Unexpected attribute "org_code" for model "CatalogPathway"' until
+# https://github.com/typeddjango/django-stubs/issues/1034 is fixed.)
+
+from datetime import datetime, timezone
+
+import pytest
+from django.contrib.auth import get_user_model
+from django.db import transaction
+from django.db.models import ProtectedError
+from django.db.utils import IntegrityError
+from freezegun import freeze_time
+from organizations.api import ensure_organization # type: ignore[import]
+from organizations.models import Organization # type: ignore[import]
+
+from openedx_catalog.models import CatalogPathway, PathwayCategory, PathwayEnrollment
+
+User = get_user_model()
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture(name="org1")
+def _org1() -> None:
+ """Create an "Org1" organization for use in these tests"""
+ ensure_organization("Org1")
+
+
+@pytest.fixture(name="org2")
+def _org2() -> None:
+ """Create an "Org2" organization for use in these tests"""
+ ensure_organization("Org2")
+
+
+@pytest.fixture(name="category")
+def _category() -> PathwayCategory:
+ """The category created by the initial migration, which operators may rename like any other"""
+ return PathwayCategory.objects.get(category_code="pathway")
+
+
+@pytest.fixture(name="data_science")
+def _data_science(org1, category) -> CatalogPathway:
+ """Create a CatalogPathway for use in these tests"""
+ return CatalogPathway.objects.create(org_code="Org1", pathway_code="DataScience", category=category)
+
+
+@pytest.fixture(name="learner")
+def _learner():
+ """Create a learner for use in these tests"""
+ return User.objects.create(username="learner", email="learner@example.com")
+
+
+# PathwayCategory
+
+
+def test_initial_category_is_created() -> None:
+ """
+ The initial migration creates a "Pathway" category, so that a new instance has one to pick. It's a database row
+ rather than a fallback in code, so operators can rename it without a code change (ADR 0007, decision 2).
+ """
+ assert PathwayCategory.objects.get(category_code="pathway").title == "Pathway"
+
+
+def test_category_is_required(org1) -> None:
+ """There is no default category, so a pathway can't be saved without one."""
+ with pytest.raises(IntegrityError), transaction.atomic():
+ CatalogPathway.objects.create(org_code="Org1", pathway_code="NoCategory")
+
+
+def test_category_code_unique_ci() -> None:
+ """Category codes are case-insensitively unique."""
+ PathwayCategory.objects.create(category_code="masters-degree", title="Master's Degree")
+ with pytest.raises(IntegrityError), transaction.atomic():
+ PathwayCategory.objects.create(category_code="Masters-Degree", title="Duplicate")
+
+
+def test_category_title_cannot_be_blank() -> None:
+ """The learner-facing title is required at the database level."""
+ with pytest.raises(IntegrityError), transaction.atomic():
+ PathwayCategory.objects.create(category_code="blank-title", title="")
+
+
+def test_category_in_use_cannot_be_deleted(data_science) -> None:
+ """Deleting a category out from under a pathway would leave it without one."""
+ with pytest.raises(ProtectedError):
+ data_science.category.delete()
+
+
+def test_category_string_representation(category) -> None:
+ """The string representation of a category is its title."""
+ assert str(category) == "Pathway"
+
+
+def test_title_plural_defaults_to_title_with_an_s(category) -> None:
+ """Most titles pluralize by appending "s", so title_plural is optional."""
+ assert category.title_plural == ""
+ assert category.get_title_plural() == "Pathways"
+
+
+def test_title_plural_when_appending_an_s_is_wrong() -> None:
+ category = PathwayCategory.objects.create(
+ category_code="class",
+ title="Class",
+ title_plural="Classes",
+ )
+ assert category.get_title_plural() == "Classes"
+
+
+def test_title_plural_follows_a_renamed_title(category) -> None:
+ """Without an explicit plural, renaming the title renames the plural too."""
+ category.title = "Program"
+ category.save()
+ assert category.get_title_plural() == "Programs"
+
+
+# CatalogPathway
+
+
+def test_invalid_org() -> None:
+ """The Organization must exist in the DB before a CatalogPathway can be created"""
+ with pytest.raises(Organization.DoesNotExist):
+ CatalogPathway.objects.create(org_code="NewOrg", pathway_code="Whatever")
+
+
+def test_pathway_code_unique_per_org_ci(org1, org2, category) -> None:
+ """The pathway_code is case-insensitively unique per org, but not across orgs."""
+ CatalogPathway.objects.create(org_code="Org1", pathway_code="DataScience", category=category)
+ with pytest.raises(IntegrityError), transaction.atomic():
+ CatalogPathway.objects.create(org_code="Org1", pathway_code="datascience", category=category)
+ # A different org may use the same code:
+ CatalogPathway.objects.create(org_code="Org2", pathway_code="DataScience", category=category)
+
+
+def test_title_defaults_to_pathway_code(data_science) -> None:
+ """A blank title falls back to the code, rather than failing the not-blank constraint."""
+ assert data_science.title == "DataScience"
+
+
+def test_key_str(data_science) -> None:
+ """The key is derived from the org and pathway codes."""
+ assert data_science.key_str == "catalog-pathway:Org1:DataScience"
+
+
+def test_catalog_edits_are_free(data_science) -> None:
+ """
+ Catalog copy is not versioned. Editing it is an ordinary save, with no version to create and no trace left behind.
+ """
+ data_science.title = "Data Science Professional Program"
+ data_science.description = "Learn data science."
+ data_science.save()
+
+ reloaded = CatalogPathway.objects.get(pk=data_science.pk)
+ assert reloaded.title == "Data Science Professional Program"
+ assert reloaded.description == "Learn data science."
+
+
+def test_modified_tracks_catalog_edits(org1, category) -> None:
+ """
+ `modified` moves when the catalog fields change and `created` does not.
+ """
+ created_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
+ edited_at = datetime(2026, 2, 1, tzinfo=timezone.utc)
+ with freeze_time(created_at):
+ pathway = CatalogPathway.objects.create(org_code="Org1", pathway_code="Timestamps", category=category)
+ assert pathway.created == created_at
+ assert pathway.modified == created_at
+
+ with freeze_time(edited_at):
+ pathway.title = "Renamed"
+ pathway.save()
+
+ reloaded = CatalogPathway.objects.get(pk=pathway.pk)
+ assert reloaded.created == created_at
+ assert reloaded.modified == edited_at
+
+
+def test_pathway_string_representation(data_science) -> None:
+ """Test the string representation of a pathway."""
+ data_science.title = "Data Science Professional Program"
+ data_science.save()
+ data_science.refresh_from_db()
+ assert str(data_science) == "Data Science Professional Program (Org1 DataScience)"
+
+
+# PathwayEnrollment
+
+
+def test_enrollment_is_unique_per_learner(data_science, learner) -> None:
+ """A learner is either enrolled in a pathway or not; there is never a second row."""
+ PathwayEnrollment.objects.create(user=learner, catalog_pathway=data_science)
+ with pytest.raises(IntegrityError), transaction.atomic():
+ PathwayEnrollment.objects.create(user=learner, catalog_pathway=data_science)
+
+
+def test_enrollment_pins_no_version(data_science, learner) -> None:
+ """
+ Enrollment ties a learner to the catalog half only. There is deliberately no field pinning a content version,
+ because progress is evaluated against whatever is published at the time (ADR 0007, decision 5).
+ """
+ enrollment = PathwayEnrollment.objects.create(user=learner, catalog_pathway=data_science)
+ field_names = {field.name for field in enrollment._meta.get_fields()}
+ assert not any("version" in name for name in field_names)
+
+
+def test_enrollment_is_active_by_default(data_science, learner) -> None:
+ """A fresh enrollment is active; deactivating it is how unenrolling is recorded."""
+ enrollment = PathwayEnrollment.objects.create(user=learner, catalog_pathway=data_science)
+ assert enrollment.is_active
+
+
+def test_a_pathway_with_enrollments_cannot_be_deleted(data_science, learner) -> None:
+ """PROTECT, so that deleting the wrong pathway can't silently wipe out its enrollment history."""
+ enrollment = PathwayEnrollment.objects.create(user=learner, catalog_pathway=data_science, is_active=False)
+ with pytest.raises(ProtectedError), transaction.atomic():
+ data_science.delete()
+
+ enrollment.delete() # Deliberately removing the history first is fine.
+ data_science.delete()
+ assert not CatalogPathway.objects.filter(pk=data_science.pk).exists()
+
+
+def test_enrollment_string_representation(data_science, learner) -> None:
+ """Test the string representation of a pathway enrollment."""
+ enrollment = PathwayEnrollment.objects.create(user=learner, catalog_pathway=data_science)
+ assert str(enrollment) == f"{learner} in {data_science}"