feat: api to bulk update legacy library references (#37789)

Adds API to fetch all legacy library content blocks that are ready to be updated to use library v2 and convert to item banks.
Also adds API to update all the references via a user celery task and to fetch its status.
This commit is contained in:
Navin Karkera
2025-12-22 23:19:43 +05:30
committed by GitHub
parent f6633dafb1
commit 32b7f27c46
10 changed files with 331 additions and 18 deletions

View File

@@ -19,6 +19,7 @@ from .tasks import send_task_complete_email
LOGGER = logging.getLogger(__name__)
LIBRARY_CONTENT_TASK_NAME_TEMPLATE = 'updating .*type@library_content.* from library'
LIBRARY_IMPORT_TASK_NAME_TEMPLATE = '(.*)?migrate_from_modulestore'
LEGACY_LIB_CONTENT_REF_UPDATE_TASK_TEMPLATE = 'updating legacy library content blocks references of course-v1:.*'
@receiver(user_task_stopped, dispatch_uid="cms_user_task_stopped")
@@ -63,6 +64,14 @@ def user_task_stopped_handler(sender, **kwargs): # pylint: disable=unused-argum
p = re.compile(LIBRARY_IMPORT_TASK_NAME_TEMPLATE)
return p.match(task_name) is not None
def is_legacy_library_content_reference_update(task_name: str) -> bool:
"""
Decides whether to suppress an end-of-task email on the basis that the just-ended task
was a legacy library content reference update operation.
"""
p = re.compile(LEGACY_LIB_CONTENT_REF_UPDATE_TASK_TEMPLATE)
return p.match(task_name) is not None
def get_olx_validation_from_artifact():
"""
Get olx validation error if available for current task.
@@ -96,7 +105,8 @@ def user_task_stopped_handler(sender, **kwargs): # pylint: disable=unused-argum
is_library_content_update(task_name) or
is_library_backup_task(task_name) or
is_library_restore_task(task_name) or
is_library_import_task(task_name)
is_library_import_task(task_name) or
is_legacy_library_content_reference_update(task_name)
)
status = kwargs['status']

View File

@@ -231,6 +231,16 @@ class TestUserTaskStopped(APITestCase):
self.assertEqual(len(mail.outbox), 0)
def test_email_not_sent_with_legacy_libary_content_ref_update(self):
"""
Check the signal receiver and email sending.
"""
end_of_task_status = self.status
end_of_task_status.name = "Updating legacy library content blocks references of course-v1:UNIX+UN1+2025_T4"
user_task_stopped.send(sender=UserTaskStatus, status=end_of_task_status)
self.assertEqual(len(mail.outbox), 0)
def test_email_sent_with_olx_validations_with_config_enabled(self):
"""
Tests that email is sent with olx validation errors.