feat: Preview migration api [FC-0114]

cms/djangoapps/modulestore_migrator/api/read_api.py

@@ -5,18 +5,27 @@
 
 import typing as t
 from uuid import UUID
+from django.conf import settings
 
 from opaque_keys.edx.keys import UsageKey
 from opaque_keys.edx.locator import (
     LibraryLocatorV2, LibraryUsageLocatorV2, LibraryContainerLocator
 )
-from openedx_learning.api.authoring import get_draft_version
+from openedx_learning.api.authoring import get_draft_version, get_all_drafts
 from openedx_learning.api.authoring_models import (
     PublishableEntityVersion, PublishableEntity, DraftChangeLogRecord
 )
+from xblock.plugin import PluginMissingError
 
 from openedx.core.djangoapps.content_libraries.api import (
-    library_component_usage_key, library_container_locator
+    library_component_usage_key, library_container_locator,
+    validate_can_add_block_to_library, BlockLimitReachedError,
+    IncompatibleTypesError, LibraryBlockAlreadyExists,
+    ContentLibrary
+)
+from openedx.core.djangoapps.content.search.api import (
+    fetch_block_types,
+    get_all_blocks_from_context,
 )
 
 from ..data import (
@@ -32,6 +41,7 @@
     'get_forwarding_for_blocks',
     'get_migrations',
     'get_migration_blocks',
+    'preview_migration',
 )
 
 
@@ -242,3 +252,117 @@ def _block_migration_success(
         target_title=target_title,
         target_version_num=target_version_num,
     )
+
+
+def preview_migration(source_key: SourceContextKey, target_key: LibraryLocatorV2):
+    """
+    Returns a summary preview of the migration given a source key and a target key
+    on this form:
+
+    ```
+        {
+            "state": "partial",
+            "unsupported_blocks": 4,
+            "unsupported_percentage": 25,
+            "blocks_limit": 1000,
+            "total_blocks": 20,
+            "total_components": 10,
+            "sections": 2,
+            "subsections": 3,
+            "units": 5,
+        }
+    ```
+
+    List of states:
+    - 'success': The migration can be carried out in its entirety
+    - 'partial': The migration will be partial, because there are unsupported blocks.
+    - 'block_limit_reached': The migration cannot be performed because the block limit per library has been reached.
+
+    TODO: For now, the repeat_handling_strategy is not taken into account. This can be taken into
+    account for a more advanced summary.
+    """
+    # Get all containers and components from the source key
+    blocks = get_all_blocks_from_context(str(source_key), ["block_type", "block_id"])
+
+    unsupported_blocks = []
+    total_blocks = 0
+    total_components = 0
+    sections = 0
+    subsections = 0
+    units = 0
+    blocks_limit = settings.MAX_BLOCKS_PER_CONTENT_LIBRARY
+
+    # Builds the summary: counts every container and verify if each component can be added to the library
+    for block in blocks:
+        block_type = block["block_type"]
+        block_id = block["block_id"]
+        total_blocks += 1
+        if block_type not in ['chapter', 'sequential', 'vertical']:
+            total_components += 1
+            try:
+                validate_can_add_block_to_library(
+                    target_key,
+                    block_type,
+                    block_id,
+                )
+            except BlockLimitReachedError:
+                return {
+                    "state": "block_limit_reached",
+                    "unsupported_blocks": 0,
+                    "unsupported_percentage": 0,
+                    "blocks_limit": blocks_limit,
+                    "total_blocks": 0,
+                    "total_components": 0,
+                    "sections": 0,
+                    "subsections": 0,
+                    "units": 0,
+                }
+            except (IncompatibleTypesError, PluginMissingError):
+                unsupported_blocks.append(block["usage_key"])
+            except LibraryBlockAlreadyExists:
+                # Skip this validation, The block may be repeated in the library, but that's not a bad thing.
+                pass
+        elif block_type == "chapter":
+            sections += 1
+        elif block_type == "sequential":
+            subsections += 1
+        elif block_type == "vertical":
+            units += 1
+
+    # Gets the count of children of unsupported blocks
+    quoted_keys = ','.join(f'"{key}"' for key in unsupported_blocks)
+    unsupportedBlocksChildren = fetch_block_types(
+        [
+            f'context_key = "{source_key}"',
+            f'breadcrumbs.usage_key IN [{quoted_keys}]'
+        ],
+    )
+    # Final unsupported blocks count
+    # The unsupported children are subtracted from the totals since they have already been counted in the first query.
+    unsupported_blocks_count = len(unsupported_blocks)
+    total_blocks -= unsupportedBlocksChildren["estimatedTotalHits"]
+    total_components -= unsupportedBlocksChildren["estimatedTotalHits"]
+    unsupported_percentage = (unsupported_blocks_count / total_blocks) * 100
+
+    state = "success"
+    if unsupported_blocks_count:
+        state = "partial"
+
+    # Checks if this migration reaches the block limit
+    content_library = ContentLibrary.objects.get_by_key(target_key)
+    assert content_library.learning_package_id is not None
+    target_item_counts = get_all_drafts(content_library.learning_package_id).count()
+    if (target_item_counts + total_blocks - unsupported_blocks_count) > blocks_limit:
+        state = "block_limit_reached"
+
+    return {
+        "state": state,
+        "unsupported_blocks": unsupported_blocks_count,
+        "unsupported_percentage": unsupported_percentage,
+        "blocks_limit": blocks_limit,
+        "total_blocks": total_blocks,
+        "total_components": total_components,
+        "sections": sections,
+        "subsections": subsections,
+        "units": units,
+    }

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

@@ -289,3 +289,18 @@ class BlockMigrationInfoSerializer(serializers.Serializer):
     source_key = serializers.CharField()
     target_key = serializers.CharField(allow_null=True)
     unsupported_reason = serializers.CharField(allow_null=True)
+
+
+class PreviewMigrationSerializer(serializers.Serializer):
+    """
+    Serializer for the preview migration response.
+    """
+    state = serializers.CharField()
+    unsupported_blocks = serializers.IntegerField()
+    unsupported_percentage = serializers.FloatField()
+    blocks_limit = serializers.IntegerField()
+    total_blocks = serializers.IntegerField()
+    total_components = serializers.IntegerField()
+    sections = serializers.IntegerField()
+    subsections = serializers.IntegerField()
+    units = serializers.IntegerField()

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

@@ -10,6 +10,7 @@
     LibraryCourseMigrationViewSet,
     MigrationInfoViewSet,
     MigrationViewSet,
+    PreviewMigration,
 )
 
 ROUTER = SimpleRouter()
@@ -25,4 +26,5 @@
     path('', include(ROUTER.urls)),
     path('migration_info/', MigrationInfoViewSet.as_view(), name='migration-info'),
     path('migration_blocks/', BlockMigrationInfo.as_view(), name='migration-blocks'),
+    path('migration_preview/', PreviewMigration.as_view(), name='migration-preview'),
 ]

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

@@ -41,6 +41,7 @@
     MigrationInfoResponseSerializer,
     ModulestoreMigrationSerializer,
     StatusWithModulestoreMigrationsSerializer,
+    PreviewMigrationSerializer,
 )
 
 log = logging.getLogger(__name__)
@@ -668,3 +669,88 @@ def get(self, request: Request):
         ]
         serializer = BlockMigrationInfoSerializer(data, many=True)
         return Response(serializer.data)
+
+
+class PreviewMigration(APIView):
+    """
+    Retrieve the summary preview of the migration given a source key and a target key
+
+    It returns the migration block information for each block migrated by a specific task.
+
+    API Endpoints
+    -------------
+    GET /api/modulestore_migrator/v1/migration_preview/
+        Retrieve the summary preview of the migration given a source key and a target key
+
+        Query parameters:
+            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
+
+        Example request:
+            GET /api/modulestore_migrator/v1/migration_blocks/?source_key=course_key&target_key=library_key
+
+        Example response:
+    """
+
+    permission_classes = (IsAuthenticated,)
+    authentication_classes = (
+        BearerAuthenticationAllowInactiveUser,
+        JwtAuthentication,
+        SessionAuthenticationAllowInactiveUser,
+    )
+
+    @apidocs.schema(
+        parameters=[
+            apidocs.string_parameter(
+                "target_key",
+                apidocs.ParameterLocation.QUERY,
+                description="Target key of the migration",
+            ),
+            apidocs.string_parameter(
+                "source_key",
+                apidocs.ParameterLocation.QUERY,
+                description="Source key of the migration",
+            ),
+        ],
+        responses={
+            200: PreviewMigrationSerializer,
+            400: "Missing required parameter: target_key/source_key",
+            401: "The requester is not authenticated.",
+        },
+    )
+    def get(self, request: Request):
+        """
+        Handle the migration info `GET` request
+        """
+        target_key: LibraryLocatorV2 | None
+        if target_key_param := request.query_params.get("target_key"):
+            try:
+                target_key = LibraryLocatorV2.from_string(target_key_param)
+            except InvalidKeyError:
+                return Response({"error": f"Bad target_key: {target_key_param}"}, status=400)
+        else:
+            return Response({"error": "Target key cannot be blank."}, status=400)
+        source_key: SourceContextKey | None = None
+        if source_key_param := request.query_params.get("source_key"):
+            try:
+                source_key = CourseLocator.from_string(source_key_param)
+            except InvalidKeyError:
+                try:
+                    source_key = LibraryLocator.from_string(source_key_param)
+                except InvalidKeyError:
+                    return Response({"error": f"Bad source: {source_key_param}"}, status=400)
+        else:
+            return Response({"error": "Source key cannot be blank."}, status=400)
+
+        lib_api.require_permission_for_library_key(
+            target_key,
+            request.user,
+            lib_api.permissions.CAN_EDIT_THIS_CONTENT_LIBRARY,
+        )
+        result = migrator_api.preview_migration(source_key, target_key)
+
+        serializer = PreviewMigrationSerializer(result)
+
+        return Response(serializer.data)