feat: Legacy Libraries Migration (Prototype)

cms/djangoapps/contentstore/views/course.py

@@ -673,15 +673,40 @@ def library_listing(request):
 def _format_library_for_view(library, request):
     """
     Return a dict of the data which the view requires for each library
-    """
 
+    @@TODO This is a hacky prototype implementation. In a real implementation, we'd probably want to change the schema
+           of this API to include both the old an (if migrate) new library metadata, and then leave the messaging and
+           URL-building logic to frontend-app-authoring.
+    """
+    # @@TODO: Either put ContentLibraryMigration behind a Python API, or move it to the contentstore app.
+    #         That way, this app doesn't need to import from another app's models.
+    from openedx.core.djangoapps.content_libraries.models import ContentLibraryMigration
+    try:
+        migration = ContentLibraryMigration.objects.select_related(
+            "target", "target__learning_package", "target_collection"
+        ).get(source_key=library.id)
+    except ContentLibraryMigration.DoesNotExist:
+        # Library is not yet migrated. Point to legacy legacyy.
+        display_name = library.display_name
+        url = reverse_library_url('library_handler', str(library.location.library_key))
+        key_for_access_check = library.context_key
+    else:
+        url = f"{settings.COURSE_AUTHORING_MICROFRONTEND_URL}/library/{migration.target.library_key}"
+        new_library_title = migration.target.learning_package.title
+        if migration.target_collection:
+            url = f"{url}/collection/{migration.target_collection.key}"
+            collection_title = migration.target_collection.title
+            display_name = f"{library.display_name} (migrated to '{collection_title}' in '{new_library_title}')"
+        else:
+            display_name = f"{library.display_name} (migrated to '{new_library_title}')"
+        key_for_access_check = migration.target.library_key
     return {
-        'display_name': library.display_name,
+        'display_name': display_name,
+        'url': url,
         'library_key': str(library.location.library_key),
-        'url': reverse_library_url('library_handler', str(library.location.library_key)),
         'org': library.display_org_with_default,
         'number': library.display_number_with_default,
-        'can_edit': has_studio_write_access(request.user, library.location.library_key),
+        'can_edit': has_studio_write_access(request.user, key_for_access_check),
     }
 
 

cms/lib/xblock/upstream_sync.py

@@ -146,16 +146,47 @@ def get_for_block(cls, downstream: XBlock) -> t.Self:
         If link exists, is supported, and is followable, returns UpstreamLink.
         Otherwise, raises an UpstreamLinkException.
         """
-        if not downstream.upstream:
+        if downstream.upstream:
+            if not isinstance(downstream.usage_key.context_key, CourseKey):
+                raise BadDownstream(_("Cannot update content because it does not belong to a course."))
+            if downstream.has_children:
+                raise BadDownstream(_("Updating content with children is not yet supported."))
+
+        # We need to determine the usage key of this block's upstream.
+        upstream_key: LibraryUsageLocatorV2
+        version_synced: int | None
+        version_available: int | None
+        # A few different scenarios...
+
+        # Do we have an upstream explicitly defined on the block? If so, use that.
+        if downstream.upstream:
+            try:
+                upstream_key = LibraryUsageLocatorV2.from_string(downstream.upstream)
+            except InvalidKeyError as exc:
+                raise BadUpstream(_("Reference to linked library item is malformed")) from exc
+            version_synced = downstream.upstream_version
+            version_declined = downstream.upstream_version_declined
+
+        # Otherwise, is this the child of a LegacyLibraryContentBlock?
+        # If so, then we know that this block was derived from block in a legacy (v1) content library.
+        # Try to get that block's migrated (v2) content library equivalent and use it as our upstream.
+        elif downstream.parent and downstream.parent.block_type == "library_content":
+            from xmodule.library_content_block import LegacyLibraryContentBlock
+            parent: LegacyLibraryContentBlock = downstream.get_parent()
+            # Next line will raise UpstreamLinkException if no matching V2 library block.
+            upstream_key = parent.get_migrated_upstream_for_child(downstream.usage_key.block_id)
+            # If we are here, then there is indeed a migrated V2 library block, but we have not yet synced from it
+            # (otherwise `.upstream` would have been explicitly set). So, it is fair to set the version information
+            # to "None". That way, as soon as an updated version of the migrated upstream is published, it will be
+            # available to the course author.
+            version_synced = None
+            version_declined = None
+
+        # Otherwise, we don't have an upstream. Raise.
+        else:
             raise NoUpstream()
-        if not isinstance(downstream.usage_key.context_key, CourseKey):
-            raise BadDownstream(_("Cannot update content because it does not belong to a course."))
-        if downstream.has_children:
-            raise BadDownstream(_("Updating content with children is not yet supported."))
-        try:
-            upstream_key = LibraryUsageLocatorV2.from_string(downstream.upstream)
-        except InvalidKeyError as exc:
-            raise BadUpstream(_("Reference to linked library item is malformed")) from exc
+
+        # Ensure that the upstream block is of a compatible type.
         downstream_type = downstream.usage_key.block_type
         if upstream_key.block_type != downstream_type:
             # Note: Currently, we strictly enforce that the downstream and upstream block_types must exactly match.
@@ -178,8 +209,8 @@ def get_for_block(cls, downstream: XBlock) -> t.Self:
         except XBlockNotFoundError as exc:
             raise BadUpstream(_("Linked library item was not found in the system")) from exc
         return cls(
-            upstream_ref=downstream.upstream,
-            version_synced=downstream.upstream_version,
+            upstream_ref=str(upstream_key),
+            version_synced=downstream.upstream_version if downstream.upstream else 0,
             version_available=(lib_meta.published_version_num if lib_meta else None),
             version_declined=downstream.upstream_version_declined,
             error_message=None,
@@ -201,6 +232,13 @@ def sync_from_upstream(downstream: XBlock, user: User) -> None:
     _update_tags(upstream=upstream, downstream=downstream)
     downstream.upstream_version = link.version_available
 
+    # Explicitly set the `upstream` setting of the downstream block from the upstream's usage key.
+    # In most cases, this is a no-op, since that is normally how we'd spefically an upstream.
+    # However, it is also possible for a block to have implicitly-defined upstream-- particularly, if it is the child of
+    # a LegacyLibraryContentBlock, whose source library was recently migrated from a V1 library to a V2 library.
+    # In that case, we want to "migrate" the downstream to the new schema by explicitly setting its `upstream` setting.
+    downstream.upstream = str(upstream.usage_key)
+
 
 def fetch_customizable_fields(*, downstream: XBlock, user: User, upstream: XBlock | None = None) -> None:
     """
@@ -213,6 +251,9 @@ def fetch_customizable_fields(*, downstream: XBlock, user: User, upstream: XBloc
         _link, upstream = _load_upstream_link_and_block(downstream, user)
     _update_customizable_fields(upstream=upstream, downstream=downstream, only_fetch=True)
 
+    # (see comment in sync_from_upstream)
+    downstream.upstream = str(upstream.usage_key)
+
 
 def _load_upstream_link_and_block(downstream: XBlock, user: User) -> tuple[UpstreamLink, XBlock]:
     """
@@ -227,14 +268,16 @@ def _load_upstream_link_and_block(downstream: XBlock, user: User) -> tuple[Upstr
     # We import load_block here b/c UpstreamSyncMixin is used by cms/envs, which loads before the djangoapps are ready.
     from openedx.core.djangoapps.xblock.api import load_block, CheckPerm, LatestVersion  # pylint: disable=wrong-import-order
     try:
+        # We know that upstream_ref cannot be None, since get_for_block returned successfully.
+        upstream_ref: str = link.upstream_ref  # type: ignore[assignment]
         lib_block: XBlock = load_block(
-            LibraryUsageLocatorV2.from_string(downstream.upstream),
+            LibraryUsageLocatorV2.from_string(upstream_ref),
             user,
             check_permission=CheckPerm.CAN_READ_AS_AUTHOR,
             version=LatestVersion.PUBLISHED,
         )
     except (NotFound, PermissionDenied) as exc:
-        raise BadUpstream(_("Linked library item could not be loaded: {}").format(downstream.upstream)) from exc
+        raise BadUpstream(_("Linked library item could not be loaded: {}").format(link.upstream_ref)) from exc
     return link, lib_block
 
 

common/djangoapps/split_modulestore_django/admin.py

@@ -1,12 +1,90 @@
 """
 Admin registration for Split Modulestore Django Backend
 """
-from django.contrib import admin
+import logging
+
+from django import forms
+from django.contrib import admin, messages
+from django.contrib.admin.helpers import ActionForm
+from opaque_keys import InvalidKeyError
+from opaque_keys.edx.locator import LibraryLocator as LegacyLibraryLocator, LibraryLocatorV2, LibraryCollectionLocator
 from simple_history.admin import SimpleHistoryAdmin
 
+from openedx.core.djangoapps.content_libraries.migration_api import migrate_legacy_library
+
 from .models import SplitModulestoreCourseIndex
 
 
+logger = logging.getLogger(__name__)
+
+
+@admin.action(description="Migrate Legacy Library to new Library or Collection")
+def migrate(modeladmin, request, queryset):
+    """
+    Migrate legacy modulestore index entries to Learning Core, based on `migration_target_key`.
+
+    Currently, this only works for LEGACY LIBRARY (library-v1:...) index entries.
+    Will fail if used on any other course entry.
+
+    The only valid targets are currently V2 Libraries and their Collections.
+    Will fail on any other type of target key.
+
+    WARNING: This does not delete the remaining legacy index item! It's up to Studio to recognize that an item has been
+    migrated, and that the legacy entry should be ignored.
+    """
+    target_key_string = request.POST['migration_target_key']
+    target_library_key: LibraryLocatorV2
+    target_collection_slug: str | None
+    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.library_key
+            target_collection_slug = target_collection_key.collection_id
+        except InvalidKeyError:
+            modeladmin.message_user(
+                request,
+                f"Migration target key is not a valid V2 Library or Collection key: {target_key_string}",
+                level=messages.ERROR,
+            )
+            return
+    for obj in queryset:
+        if not isinstance(obj.course_id, LegacyLibraryLocator):
+            modeladmin.message_user(
+                request,
+                f"Selected entry is not a Legacy Library: {obj.course_id}. Skipping.",
+                level=messages.WARNING,
+            )
+            continue
+        try:
+            migrate_legacy_library(
+                source_key=obj.course_id,
+                target_key=target_library_key,
+                collection_slug=target_collection_slug,
+                user=request.user,
+            )
+        except Exception as exc:  # pylint: disable=broad-except
+            modeladmin.message_user(
+                request,
+                f"Failed to migrate {obj.course_id} to {target_key_string}: {exc}. See logs for details.",
+                level=messages.ERROR,
+            )
+            logger.exception(exc)
+            continue
+        else:
+            modeladmin.message_user(
+                request,
+                f"Migrated {obj.course_id} to {target_key_string}",
+                level=messages.SUCCESS,
+            )
+
+
+class MigrationTargetForm(ActionForm):
+    migration_target_key = forms.CharField()
+
+
 @admin.register(SplitModulestoreCourseIndex)
 class SplitModulestoreCourseIndexAdmin(SimpleHistoryAdmin):
     """
@@ -15,4 +93,5 @@ class SplitModulestoreCourseIndexAdmin(SimpleHistoryAdmin):
     list_display = ('course_id', 'draft_version', 'published_version', 'library_version', 'wiki_slug', 'last_update')
     search_fields = ('course_id', 'wiki_slug')
     ordering = ('course_id', )
-    readonly_fields = ('id', 'objectid', 'course_id', 'org', )
+    actions = [migrate]
+    action_form = MigrationTargetForm

openedx/core/djangoapps/content_libraries/admin.py

@@ -2,7 +2,9 @@
 Admin site for content libraries
 """
 from django.contrib import admin
-from .models import ContentLibrary, ContentLibraryPermission
+from .models import (
+    ContentLibrary, ContentLibraryPermission, ContentLibraryMigration, ContentLibraryBlockMigration
+)
 
 
 class ContentLibraryPermissionInline(admin.TabularInline):
@@ -39,3 +41,20 @@ def get_readonly_fields(self, request, obj=None):
             return ["library_key", "org", "slug"]
         else:
             return ["library_key", ]
+
+
+class ContentLibraryBlockMigrationInline(admin.TabularInline):
+    """
+    Django admin UI for content library block migrations
+    """
+    model = ContentLibraryBlockMigration
+    list_display = ("library_migration", "block_type", "source_block_id", "target_block_id")
+
+
+@admin.register(ContentLibraryMigration)
+class ContentLibraryMigrationAdmin(admin.ModelAdmin):
+    """
+    Django admin UI for content library migrations
+    """
+    list_display = ("source_key", "target", "target_collection")
+    inlines = (ContentLibraryBlockMigrationInline,)

openedx/core/djangoapps/content_libraries/management/commands/migrate_legacy_library.py

@@ -0,0 +1,49 @@
+"""
+Implements ./manage.py cms migrate_legacy_library
+"""
+import logging
+
+from django.contrib.auth.models import User   # pylint: disable=imported-auth-user
+from django.core.management import BaseCommand
+
+from opaque_keys.edx.locator import LibraryLocator, LibraryLocatorV2
+from openedx.core.djangoapps.content_libraries.migration_api import migrate_legacy_library
+
+
+log = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """
+    @TODO
+    """
+
+    def add_arguments(self, parser):
+        """
+        Add arguments to the argument parser.
+        """
+        parser.add_argument(
+            'legacy_library',
+            type=LibraryLocator.from_string,
+        )
+        parser.add_argument(
+            'new_library',
+            type=LibraryLocatorV2.from_string,
+        )
+        parser.add_argument(
+            'collection',
+            type=str,
+        )
+
+    def handle(  # pylint: disable=arguments-differ
+        self,
+        legacy_library: LibraryLocator,
+        new_library: LibraryLocatorV2,
+        collection: str | None,
+        **kwargs,
+    ) -> None:
+        """
+        Handle the command.
+        """
+        user = User.objects.filter(is_superuser=True)[0]
+        migrate_legacy_library(legacy_library, new_library, collection_slug=collection, user=user)