feat: store information for failed block migrations

cms/djangoapps/modulestore_migrator/admin.py

@@ -178,6 +178,7 @@ class ModulestoreBlockMigrationInline(admin.TabularInline):
         "source",
         "target",
         "change_log_record",
+        "unsupported_reason",
     )
     list_display = ("id", *readonly_fields)
 

cms/djangoapps/modulestore_migrator/api.py

@@ -1,6 +1,7 @@
 """
 API for migration from modulestore to learning core
 """
+from uuid import UUID
 from collections import defaultdict
 from celery.result import AsyncResult
 from opaque_keys import InvalidKeyError
@@ -181,5 +182,30 @@ def construct_usage_key(lib_key_str: str, component: Component) -> LibraryUsageL
     return {
         obj.source.key: construct_usage_key(obj.target.learning_package.key, obj.target.component)
         for obj in query_set
-        if obj.source.key is not None
+        if obj.source.key is not None and obj.target is not None
     }
+
+
+def get_migration_blocks_info(
+    target_key: str,
+    source_key: str | None,
+    target_collection_key: str | None,
+    task_uuid: str | None,
+    is_failed: bool | None,
+):
+    """
+    Given the target key, and optional source key, target collection key, task_uuid and is_failed get a dictionary
+    containing information about migration blocks.
+    """
+    filters: dict[str, str | UUID | bool] = {
+        'overall_migration__target__key': target_key
+    }
+    if source_key:
+        filters['overall_migration__source__key'] = source_key
+    if target_collection_key:
+        filters['overall_migration__target_collection__key'] = target_collection_key
+    if task_uuid:
+        filters['overall_migration__task_status__uuid'] = UUID(task_uuid)
+    if is_failed is not None:
+        filters['target__isnull'] = is_failed
+    return ModulestoreBlockMigration.objects.filter(**filters)

cms/djangoapps/modulestore_migrator/migrations/0004_alter_modulestoreblockmigration_target_squashed_0005_modulestoreblockmigration_unsupported_reason.py

@@ -0,0 +1,32 @@
+# Generated by Django 5.2.7 on 2025-11-26 06:35
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ('modulestore_migrator', '0003_modulestoremigration_is_failed'),
+        ('oel_publishing', '0008_alter_draftchangelogrecord_options_and_more'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='modulestoreblockmigration',
+            name='target',
+            field=models.ForeignKey(
+                blank=True,
+                help_text='The target entity of this block migration, set to null if it fails to migrate',
+                null=True,
+                on_delete=django.db.models.deletion.CASCADE,
+                to='oel_publishing.publishableentity',
+            ),
+        ),
+        migrations.AddField(
+            model_name='modulestoreblockmigration',
+            name='unsupported_reason',
+            field=models.TextField(
+                blank=True, help_text='Reason if the block is unsupported and target is set to null', null=True
+            ),
+        ),
+    ]

cms/djangoapps/modulestore_migrator/models.py

@@ -6,16 +6,19 @@
 from django.contrib.auth import get_user_model
 from django.db import models
 from django.utils.translation import gettext_lazy as _
-from user_tasks.models import UserTaskStatus
-
 from model_utils.models import TimeStampedModel
 from opaque_keys.edx.django.models import (
     LearningContextKeyField,
     UsageKeyField,
 )
 from openedx_learning.api.authoring_models import (
-    LearningPackage, PublishableEntity, Collection, DraftChangeLog, DraftChangeLogRecord
+    Collection,
+    DraftChangeLog,
+    DraftChangeLogRecord,
+    LearningPackage,
+    PublishableEntity,
 )
+from user_tasks.models import UserTaskStatus
 
 from .data import CompositionLevel, RepeatHandlingStrategy
 
@@ -210,6 +213,9 @@ class ModulestoreBlockMigration(TimeStampedModel):
     target = models.ForeignKey(
         PublishableEntity,
         on_delete=models.CASCADE,
+        help_text=_('The target entity of this block migration, set to null if it fails to migrate'),
+        null=True,
+        blank=True,
     )
     change_log_record = models.OneToOneField(
         DraftChangeLogRecord,
@@ -218,10 +224,16 @@ class ModulestoreBlockMigration(TimeStampedModel):
         null=True,
         on_delete=models.SET_NULL,
     )
+    unsupported_reason = models.TextField(
+        null=True,
+        blank=True,
+        help_text=_('Reason if the block is unsupported and target is set to null'),
+    )
 
     class Meta:
         unique_together = [
             ('overall_migration', 'source'),
+            # By default defining a unique index on a nullable column will only enforce unicity of non-null values.
             ('overall_migration', 'target'),
         ]
 

cms/djangoapps/modulestore_migrator/rest_api/v1/serializers.py

@@ -11,7 +11,10 @@
 from user_tasks.serializers import StatusSerializer
 
 from cms.djangoapps.modulestore_migrator.data import CompositionLevel, RepeatHandlingStrategy
-from cms.djangoapps.modulestore_migrator.models import ModulestoreMigration, ModulestoreSource
+from cms.djangoapps.modulestore_migrator.models import (
+    ModulestoreMigration,
+    ModulestoreSource,
+)
 
 
 class ModulestoreMigrationSerializer(serializers.Serializer):
@@ -266,3 +269,12 @@ def get_progress(self, obj: ModulestoreMigration):
         Return the progress of the migration.
         """
         return obj.task_status.completed_steps / obj.task_status.total_steps
+
+
+class BlockMigrationInfoSerializer(serializers.Serializer):
+    """
+    Serializer for the block migration info.
+    """
+    source_key = serializers.CharField(source="source__key")
+    target_key = serializers.CharField(source="target__key")
+    unsupported_reason = serializers.CharField()

cms/djangoapps/modulestore_migrator/rest_api/v1/urls.py

@@ -1,9 +1,16 @@
 """
 Course to Library Import API v1 URLs.
 """
-from django.urls import path, include
+from django.urls import include, path
 from rest_framework.routers import SimpleRouter
-from .views import MigrationViewSet, BulkMigrationViewSet, MigrationInfoViewSet, LibraryCourseMigrationViewSet
+
+from .views import (
+    BlockMigrationInfo,
+    BulkMigrationViewSet,
+    LibraryCourseMigrationViewSet,
+    MigrationInfoViewSet,
+    MigrationViewSet,
+)
 
 ROUTER = SimpleRouter()
 ROUTER.register(r'migrations', MigrationViewSet, basename='migrations')
@@ -17,4 +24,5 @@
 urlpatterns = [
     path('', include(ROUTER.urls)),
     path('migration_info/', MigrationInfoViewSet.as_view(), name='migration-info'),
+    path('migration_blocks/', BlockMigrationInfo.as_view(), name='migration-blocks'),
 ]

cms/djangoapps/modulestore_migrator/rest_api/v1/views.py

@@ -7,33 +7,37 @@
 from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
 from edx_rest_framework_extensions.auth.session.authentication import SessionAuthenticationAllowInactiveUser
 from opaque_keys import InvalidKeyError
+from opaque_keys.edx.keys import CourseKey
 from opaque_keys.edx.locator import LibraryLocatorV2
 from rest_framework import status
 from rest_framework.exceptions import ParseError
+from rest_framework.fields import BooleanField
 from rest_framework.mixins import ListModelMixin
 from rest_framework.permissions import IsAdminUser, IsAuthenticated
+from rest_framework.request import Request
 from rest_framework.response import Response
 from rest_framework.views import APIView
 from rest_framework.viewsets import GenericViewSet
 from user_tasks.models import UserTaskStatus
 from user_tasks.views import StatusViewSet
-from opaque_keys.edx.keys import CourseKey
 
 from cms.djangoapps.modulestore_migrator.api import (
-    start_migration_to_library,
-    start_bulk_migration_to_library,
     get_all_migrations_info,
+    get_migration_blocks_info,
+    start_bulk_migration_to_library,
+    start_migration_to_library,
 )
+from common.djangoapps.student.auth import has_studio_write_access
 from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
 from openedx.core.djangoapps.content_libraries import api as lib_api
 from openedx.core.lib.api.authentication import BearerAuthenticationAllowInactiveUser
-from common.djangoapps.student.auth import has_studio_write_access
 
 from ...models import ModulestoreMigration
 from .serializers import (
+    BlockMigrationInfoSerializer,
     BulkModulestoreMigrationSerializer,
-    MigrationInfoResponseSerializer,
     LibraryMigrationCourseSerializer,
+    MigrationInfoResponseSerializer,
     ModulestoreMigrationSerializer,
     StatusWithModulestoreMigrationsSerializer,
 )
@@ -493,3 +497,105 @@ def get_queryset(self):
         queryset = queryset.filter(target__key=library_key, source__key__startswith='course-v1')
 
         return queryset
+
+
+class BlockMigrationInfo(APIView):
+    """
+    Retrieve migration blocks information given task_uuid, source_key or target_key.
+
+    It returns the migration block information for each block migrated by a specific task.
+
+    API Endpoints
+    -------------
+    GET /api/modulestore_migrator/v1/migration_blocks/
+        Retrieve migration blocks info for given task_uuid, source_key or target_key.
+
+        Query parameters:
+            task_uuid (str): task uuid
+                Example: ?task_uuid=dfe72eca-c54f-4b43-b53b-7996031f2102
+            source_key (str): Source content key
+                Example: ?source_key=course-v1:UNIX+UX1+2025_T3
+            target_key (str): target content key
+                Example: ?target_key=lib:UNIX:CIT1
+            is_failed (boolean): has the block failed to migrate/import
+                Example: ?is_failed=true
+
+        Example request:
+            GET /api/modulestore_migrator/v1/migration_blocks/?task_uuid=dfe72eca-c54f-4b43-b53b&is_failed=true
+
+        Example response:
+    """
+
+    permission_classes = (IsAuthenticated,)
+    authentication_classes = (
+        BearerAuthenticationAllowInactiveUser,
+        JwtAuthentication,
+        SessionAuthenticationAllowInactiveUser,
+    )
+
+    @apidocs.schema(
+        parameters=[
+            apidocs.string_parameter(
+                "target_key",
+                apidocs.ParameterLocation.QUERY,
+                description="Filter blocks by target key",
+            ),
+            apidocs.string_parameter(
+                "source_key",
+                apidocs.ParameterLocation.QUERY,
+                description="Filter blocks by source key",
+            ),
+            apidocs.string_parameter(
+                "target_collection_key",
+                apidocs.ParameterLocation.QUERY,
+                description="Filter blocks by target_collection_key",
+            ),
+            apidocs.string_parameter(
+                "task_uuid",
+                apidocs.ParameterLocation.QUERY,
+                description="Filter blocks by task_uuid",
+            ),
+            apidocs.string_parameter(
+                "is_failed",
+                apidocs.ParameterLocation.QUERY,
+                description="Filter blocks based on its migration status",
+            ),
+        ],
+        responses={
+            200: MigrationInfoResponseSerializer,
+            400: "Missing required parameter: target_key",
+            401: "The requester is not authenticated.",
+        },
+    )
+    def get(self, request: Request):
+        """
+        Handle the migration info `GET` request
+        """
+        source_key = request.query_params.get("source_key")
+        target_key = request.query_params.get("target_key")
+        target_collection_key = request.query_params.get("target_collection_key")
+        task_uuid = request.query_params.get("task_uuid")
+        is_failed: str | bool | None = request.query_params.get("is_failed")
+        if not target_key:
+            return Response({"error": "Target key cannot be blank."}, status=400)
+        try:
+            target_key_parsed = LibraryLocatorV2.from_string(target_key)
+        except InvalidKeyError as e:
+            return Response({"error": str(e)}, status=400)
+        lib_api.require_permission_for_library_key(
+            target_key_parsed,
+            request.user,
+            lib_api.permissions.CAN_VIEW_THIS_CONTENT_LIBRARY
+        )
+        if is_failed is not None:
+            is_failed = BooleanField().to_internal_value(is_failed)
+
+        data = get_migration_blocks_info(
+            target_key,
+            source_key,
+            target_collection_key,
+            task_uuid,
+            is_failed,
+        ).values('source__key', 'target__key', 'unsupported_reason')
+        serializer = BlockMigrationInfoSerializer(data, many=True)
+        return Response(serializer.data)