feat: add library migration list endpoint [FC-0112] (#37567)

This PR adds the `/api/modulestore_migrator/v1/library/:libraryId/migrations/courses/` endpoint, which returns all course migrations for a target library.
This commit is contained in:
Rômulo Penido
2025-11-07 13:01:35 -03:00
committed by GitHub
parent 7f8ba45f36
commit 5ef6be4610
3 changed files with 138 additions and 7 deletions

View File

@@ -5,11 +5,13 @@ Serializers for the Course to Library Import API.
from opaque_keys import InvalidKeyError
from opaque_keys.edx.keys import LearningContextKey
from opaque_keys.edx.locator import LibraryLocatorV2
from openedx_learning.api.authoring_models import Collection
from rest_framework import serializers
from user_tasks.models import UserTaskStatus
from user_tasks.serializers import StatusSerializer
from cms.djangoapps.modulestore_migrator.data import CompositionLevel, RepeatHandlingStrategy
from cms.djangoapps.modulestore_migrator.models import ModulestoreMigration
from cms.djangoapps.modulestore_migrator.models import ModulestoreMigration, ModulestoreSource
class ModulestoreMigrationSerializer(serializers.Serializer):
@@ -173,3 +175,65 @@ class StatusWithModulestoreMigrationsSerializer(StatusSerializer):
fields = super().get_fields()
fields.pop('name', None)
return fields
class LibraryMigrationCourseSourceSerializer(serializers.ModelSerializer):
"""
Serializer for the source course of a library migration.
"""
display_name = serializers.SerializerMethodField()
class Meta:
model = ModulestoreSource
fields = ['key', 'display_name']
def get_display_name(self, obj):
"""
Return the display name of the source course
"""
return self.context["course_names"].get(str(obj.key), None)
class LibraryMigrationCollectionSerializer(serializers.ModelSerializer):
"""
Serializer for the target collection of a library migration.
"""
class Meta:
model = Collection
fields = ["key", "title"]
class LibraryMigrationCourseSerializer(serializers.ModelSerializer):
"""
Serializer for the course or legacylibrary migrations to V2 library.
"""
source = LibraryMigrationCourseSourceSerializer() # type: ignore[assignment]
target_collection = LibraryMigrationCollectionSerializer(required=False)
state = serializers.SerializerMethodField()
progress = serializers.SerializerMethodField()
class Meta:
model = ModulestoreMigration
fields = [
'source',
'target_collection',
'state',
'progress',
]
def get_state(self, obj: ModulestoreMigration):
"""
Return the state of the migration.
"""
if obj.is_failed or obj.task_status.state in [UserTaskStatus.FAILED, UserTaskStatus.CANCELED]:
return UserTaskStatus.FAILED
elif obj.task_status.state == UserTaskStatus.SUCCEEDED:
return UserTaskStatus.SUCCEEDED
return UserTaskStatus.IN_PROGRESS
def get_progress(self, obj: ModulestoreMigration):
"""
Return the progress of the migration.
"""
return obj.task_status.completed_steps / obj.task_status.total_steps

View File

@@ -3,10 +3,17 @@ Course to Library Import API v1 URLs.
"""
from rest_framework.routers import SimpleRouter
from .views import MigrationViewSet, BulkMigrationViewSet
from .views import BulkMigrationViewSet, LibraryCourseMigrationViewSet, MigrationViewSet
ROUTER = SimpleRouter()
ROUTER.register(r'migrations', MigrationViewSet, basename='migrations')
ROUTER.register(r'bulk_migration', BulkMigrationViewSet, basename='bulk-migration')
ROUTER.register(
r'library/(?P<lib_key_str>[^/.]+)/migrations/courses',
LibraryCourseMigrationViewSet,
basename='library-migrations',
)
urlpatterns = ROUTER.urls

View File

@@ -6,22 +6,30 @@ import logging
import edx_api_doc_tools as apidocs
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.locator import LibraryLocatorV2
from rest_framework import status
from rest_framework.exceptions import ParseError
from rest_framework.mixins import ListModelMixin
from rest_framework.permissions import IsAdminUser
from rest_framework.response import Response
from rest_framework import status
from rest_framework.viewsets import GenericViewSet
from user_tasks.models import UserTaskStatus
from user_tasks.views import StatusViewSet
from cms.djangoapps.modulestore_migrator.api import start_migration_to_library, start_bulk_migration_to_library
from cms.djangoapps.modulestore_migrator.api import start_bulk_migration_to_library, start_migration_to_library
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 ...models import ModulestoreMigration
from .serializers import (
StatusWithModulestoreMigrationsSerializer,
ModulestoreMigrationSerializer,
BulkModulestoreMigrationSerializer,
LibraryMigrationCourseSerializer,
ModulestoreMigrationSerializer,
StatusWithModulestoreMigrationsSerializer,
)
log = logging.getLogger(__name__)
@@ -328,3 +336,55 @@ class BulkMigrationViewSet(StatusViewSet):
We disable this endpoint to avoid confusion.
"""
raise NotImplementedError
@apidocs.schema_for(
"list",
"List all course migrations to a library.",
responses={
201: LibraryMigrationCourseSerializer,
401: "The requester is not authenticated.",
403: "The requester does not have permission to access the library.",
},
)
class LibraryCourseMigrationViewSet(GenericViewSet, ListModelMixin):
"""
Show infomation about migrations related to a destination library.
"""
serializer_class = LibraryMigrationCourseSerializer
pagination_class = None
queryset = ModulestoreMigration.objects.all().select_related('target_collection', 'target', 'task_status')
def get_serializer_context(self):
"""
Add course name list to the serializer context.
We need to display the course names in the migration view, and we get all of
them here to avoid futher queries.
"""
context = super().get_serializer_context()
queryset = self.get_queryset()
course_keys = queryset.values_list('source__key', flat=True)
courses = CourseOverview.get_all_courses(course_keys=course_keys)
context['course_names'] = dict((str(course.id), course.display_name) for course in courses)
return context
def get_queryset(self):
"""
Override the default queryset to filter by the library key and check permissions.
"""
queryset = super().get_queryset()
lib_key_str = self.kwargs['lib_key_str']
try:
library_key = LibraryLocatorV2.from_string(lib_key_str)
except InvalidKeyError as exc:
raise ParseError(detail=f"Malformed library key: {lib_key_str}") from exc
lib_api.require_permission_for_library_key(
library_key,
self.request.user,
lib_api.permissions.CAN_VIEW_THIS_CONTENT_LIBRARY
)
queryset = queryset.filter(target__key=library_key, source__key__startswith='course-v1')
return queryset