feat!: modulestore_migrator

.github/workflows/unit-test-shards.json

@@ -238,6 +238,7 @@
       "cms/djangoapps/cms_user_tasks/",
       "cms/djangoapps/course_creators/",
       "cms/djangoapps/export_course_metadata/",
+      "cms/djangoapps/modulestore_migrator/",
       "cms/djangoapps/maintenance/",
       "cms/djangoapps/models/",
       "cms/djangoapps/pipeline_js/",

cms/djangoapps/modulestore_migrator/admin.py

@@ -0,0 +1,192 @@
+"""
+A nice little admin interface for migrating courses and libraries from modulstore to Learning Core.
+"""
+import logging
+
+from django import forms
+from django.contrib import admin, messages
+from django.contrib.admin.helpers import ActionForm
+from django.db import models
+
+
+from opaque_keys import InvalidKeyError
+from opaque_keys.edx.locator import LibraryCollectionLocator, LibraryLocatorV2
+from user_tasks.models import UserTaskStatus
+
+from openedx.core.types.http import AuthenticatedHttpRequest
+
+from . import api
+from .data import CompositionLevel, RepeatHandlingStrategy
+from .models import ModulestoreSource, ModulestoreMigration, ModulestoreBlockSource, ModulestoreBlockMigration
+
+
+log = logging.getLogger(__name__)
+
+
+class StartMigrationTaskForm(ActionForm):
+    """
+    Params for start_migration_task admin adtion, displayed next the "Go" button.
+    """
+    target_key = forms.CharField(label="Target library or collection key →", required=False)
+    repeat_handling_strategy = forms.ChoiceField(
+        label="How to handle existing content? →",
+        choices=RepeatHandlingStrategy.supported_choices,
+        required=False,
+    )
+    preserve_url_slugs = forms.BooleanField(label="Preserve current slugs? →", required=False, initial=True)
+    forward_to_target = forms.BooleanField(label="Forward references? →", required=False)
+    composition_level = forms.ChoiceField(
+        label="Aggregate up to →", choices=CompositionLevel.supported_choices, required=False
+    )
+
+
+def task_status_details(obj: ModulestoreMigration) -> str:
+    """
+    Return the state and, if available, details of the status of the migration.
+    """
+    details: str | None = None
+    if obj.task_status.state == UserTaskStatus.FAILED:
+        # Calling fail(msg) from a task should automatically generates an "Error" artifact with that msg.
+        # https://django-user-tasks.readthedocs.io/en/latest/user_tasks.html#user_tasks.models.UserTaskStatus.fail
+        if error_artifacts := obj.task_status.artifacts.filter(name="Error"):
+            if error_text := error_artifacts.order_by("-created").first().text:
+                details = error_text
+    elif obj.task_status.state == UserTaskStatus.SUCCEEDED:
+        details = f"Migrated {obj.block_migrations.count()} blocks"
+    return f"{obj.task_status.state}: {details}" if details else obj.task_status.state
+
+
+migration_admin_fields = (
+    "target",
+    "target_collection",
+    "task_status",
+    # The next line works, but django-stubs incorrectly thinks that these should all be strings,
+    # so we will need to use type:ignore below.
+    task_status_details,
+    "composition_level",
+    "repeat_handling_strategy",
+    "preserve_url_slugs",
+    "change_log",
+    "staged_content",
+)
+
+
+class ModulestoreMigrationInline(admin.TabularInline):
+    """
+    Readonly table within the ModulestoreSource page; each row is a Migration from this Source.
+    """
+    model = ModulestoreMigration
+    fk_name = "source"
+    show_change_link = True
+    readonly_fields = migration_admin_fields  # type: ignore[assignment]
+    ordering = ("-task_status__created",)
+
+    def has_add_permission(self, _request, _obj):
+        return False
+
+
+class ModulestoreBlockSourceInline(admin.TabularInline):
+    """
+    Readonly table within the ModulestoreSource page; each row is a BlockSource.
+    """
+    model = ModulestoreBlockSource
+    fk_name = "overall_source"
+    readonly_fields = (
+        "key",
+        "forwarded"
+    )
+
+    def has_add_permission(self, _request, _obj):
+        return False
+
+
+@admin.register(ModulestoreSource)
+class ModulestoreSourceAdmin(admin.ModelAdmin):
+    """
+    Admin interface for source legacy libraries and courses.
+    """
+    readonly_fields = ("forwarded",)
+    list_display = ("id", "key", "forwarded")
+    actions = ["start_migration_task"]
+    action_form = StartMigrationTaskForm
+    inlines = [ModulestoreMigrationInline, ModulestoreBlockSourceInline]
+
+    @admin.action(description="Start migration for selected sources")
+    def start_migration_task(
+        self,
+        request: AuthenticatedHttpRequest,
+        queryset: models.QuerySet[ModulestoreSource],
+    ) -> None:
+        """
+        Start a migration for each selected source
+        """
+        form = StartMigrationTaskForm(request.POST)
+        form.is_valid()
+        target_key_string = form.cleaned_data['target_key']
+        if not target_key_string:
+            messages.add_message(request, messages.ERROR, "Target key is required")
+            return
+        try:
+            target_library_key = LibraryLocatorV2.from_string(target_key_string)
+            target_collection_slug = None
+        except InvalidKeyError:
+            try:
+                target_collection_key = LibraryCollectionLocator.from_string(target_key_string)
+                target_library_key = target_collection_key.lib_key
+                target_collection_slug = target_collection_key.collection_id
+            except InvalidKeyError:
+                messages.add_message(request, messages.ERROR, f"Invalid target key: {target_key_string}")
+                return
+        started = 0
+        total = 0
+        for source in queryset:
+            total += 1
+            try:
+                api.start_migration_to_library(
+                    user=request.user,
+                    source_key=source.key,
+                    target_library_key=target_library_key,
+                    target_collection_slug=target_collection_slug,
+                    composition_level=form.cleaned_data['composition_level'],
+                    repeat_handling_strategy=form.cleaned_data['repeat_handling_strategy'],
+                    preserve_url_slugs=form.cleaned_data['preserve_url_slugs'],
+                    forward_source_to_target=form.cleaned_data['forward_to_target'],
+                )
+            except Exception as exc:  # pylint: disable=broad-except
+                message = f"Failed to start migration {source.key} -> {target_key_string}"
+                messages.add_message(request, messages.ERROR, f"{message}: {exc}")
+                log.exception(message)
+                continue
+            started += 1
+        click_in = "Click into the source objects to see migration details."
+
+        if not started:
+            messages.add_message(request, messages.WARNING, f"Failed to start {total} migration(s).")
+        if started < total:
+            messages.add_message(request, messages.WARNING, f"Started {started} of {total} migration(s). {click_in}")
+        else:
+            messages.add_message(request, messages.INFO, f"Started {started} migration(s). {click_in}")
+
+
+class ModulestoreBlockMigrationInline(admin.TabularInline):
+    """
+    Readonly table witin the Migration admin; each row is a block
+    """
+    model = ModulestoreBlockMigration
+    fk_name = "overall_migration"
+    readonly_fields = (
+        "source",
+        "target",
+        "change_log_record",
+    )
+    list_display = ("id", *readonly_fields)
+
+
+@admin.register(ModulestoreMigration)
+class ModulestoreMigrationAdmin(admin.ModelAdmin):
+    """
+    Readonly admin page for viewing Migrations
+    """
+    readonly_fields = ("source", *migration_admin_fields)  # type: ignore[assignment]
+    list_display = ("id", "source", *migration_admin_fields)  # type: ignore[assignment]
+    inlines = [ModulestoreBlockMigrationInline]

cms/djangoapps/modulestore_migrator/api.py

@@ -0,0 +1,58 @@
+"""
+API for migration from modulestore to learning core
+"""
+from opaque_keys.edx.locator import LibraryLocatorV2
+from opaque_keys.edx.keys import LearningContextKey
+from openedx_learning.api.authoring import get_collection
+from celery.result import AsyncResult
+
+from openedx.core.djangoapps.content_libraries.api import get_library
+from openedx.core.types.user import AuthUser
+
+from . import tasks
+from .data import RepeatHandlingStrategy
+from .models import ModulestoreSource
+
+
+__all__ = (
+    "start_migration_to_library",
+)
+
+
+def start_migration_to_library(
+    *,
+    user: AuthUser,
+    source_key: LearningContextKey,
+    target_library_key: LibraryLocatorV2,
+    target_collection_slug: str | None = None,
+    composition_level: str,
+    repeat_handling_strategy: str,
+    preserve_url_slugs: bool,
+    forward_source_to_target: bool,
+) -> AsyncResult:
+    """
+    Import a course or legacy library into a V2 library (or, a collection within a V2 library).
+    """
+    # Can raise NotImplementedError for the Fork strategy
+    assert RepeatHandlingStrategy(repeat_handling_strategy).is_implemented()
+
+    source, _ = ModulestoreSource.objects.get_or_create(key=source_key)
+    target_library = get_library(target_library_key)
+    # get_library ensures that the library is connected to a learning package.
+    target_package_id: int = target_library.learning_package_id  # type: ignore[assignment]
+    target_collection_id = None
+
+    if target_collection_slug:
+        target_collection_id = get_collection(target_package_id, target_collection_slug).id
+
+    return tasks.migrate_from_modulestore.delay(
+        user_id=user.id,
+        source_pk=source.id,
+        target_package_pk=target_package_id,
+        target_library_key=str(target_library_key),
+        target_collection_pk=target_collection_id,
+        composition_level=composition_level,
+        repeat_handling_strategy=repeat_handling_strategy,
+        preserve_url_slugs=preserve_url_slugs,
+        forward_source_to_target=forward_source_to_target,
+    )

cms/djangoapps/modulestore_migrator/apps.py

@@ -0,0 +1,13 @@
+"""
+App configurations
+"""
+
+from django.apps import AppConfig
+
+
+class ModulestoreMigratorConfig(AppConfig):
+    """
+    App for importing legacy content from the modulestore.
+    """
+
+    name = 'cms.djangoapps.modulestore_migrator'

cms/djangoapps/modulestore_migrator/constants.py

@@ -0,0 +1,6 @@
+"""
+Constants
+"""
+
+CONTENT_STAGING_PURPOSE_PREFIX = "modulestore_migrator"
+CONTENT_STAGING_PURPOSE_TEMPLATE = CONTENT_STAGING_PURPOSE_PREFIX + "({source_key})"

cms/djangoapps/modulestore_migrator/data.py

@@ -0,0 +1,81 @@
+"""
+Value objects
+"""
+from __future__ import annotations
+
+from enum import Enum
+
+from openedx.core.djangoapps.content_libraries.api import ContainerType
+
+
+class CompositionLevel(Enum):
+    """
+    Enumeration of composition levels for legacy content.
+
+    Defined in increasing order of complexity so that `is_higher_than` works correctly.
+    """
+    # Components are individual XBlocks, e.g. Problem
+    Component = 'component'
+
+    # Container types currently supported by Content Libraries
+    Unit = ContainerType.Unit.value
+    Subsection = ContainerType.Subsection.value
+    Section = ContainerType.Section.value
+
+    @property
+    def is_container(self) -> bool:
+        return self is not self.Component
+
+    def is_higher_than(self, other: 'CompositionLevel') -> bool:
+        """
+        Is this composition level 'above' (more complex than) the other?
+        """
+        levels: list[CompositionLevel] = list(self.__class__)
+        return levels.index(self) > levels.index(other)
+
+    @classmethod
+    def supported_choices(cls) -> list[tuple[str, str]]:
+        """
+        Returns all supported composition levels as a list of tuples,
+        for use in a Django Models ChoiceField.
+        """
+        return [
+            (composition_level.value, composition_level.name)
+            for composition_level in cls
+        ]
+
+
+class RepeatHandlingStrategy(Enum):
+    """
+    Enumeration of repeat handling strategies for imported content.
+    """
+    Skip = 'skip'
+    Fork = 'fork'
+    Update = 'update'
+
+    @classmethod
+    def supported_choices(cls) -> list[tuple[str, str]]:
+        """
+        Returns all supported repeat handling strategies as a list of tuples,
+        for use in a Django Models ChoiceField.
+        """
+        return [
+            (strategy.value, strategy.name)
+            for strategy in cls
+        ]
+
+    @classmethod
+    def default(cls) -> RepeatHandlingStrategy:
+        """
+        Returns the default repeat handling strategy.
+        """
+        return cls.Skip
+
+    def is_implemented(self) -> bool:
+        """
+        Returns True if the repeat handling strategy is implemented.
+        """
+        if self == self.Fork:
+            raise NotImplementedError("Forking is not implemented yet.")
+
+        return True