feat: command to delete v1 libraries (#32786)

This PR adds a management command to delete v1 content libraries. CLI options are given for a singular library, as well as all libraries. The command raises some errors related to grading, as it uses the code to delete courses, but that is something I can live with for a quick and dirty version of this capability. Also, the pruner will have to be run later to remove any orphan blocks left behind by removing the inde

cms/djangoapps/contentstore/management/commands/delete_v1_libraries.py

@@ -0,0 +1,95 @@
+"""A Command to  delete V1 Content Libraries index entires."""
+
+import logging
+from textwrap import dedent
+
+from django.core.management import BaseCommand, CommandError
+
+from opaque_keys.edx.keys import CourseKey
+from opaque_keys.edx.locator import LibraryLocator
+
+from xmodule.modulestore.django import modulestore
+
+from celery import group
+
+from cms.djangoapps.contentstore.tasks import delete_v1_library
+
+from .prompt import query_yes_no
+
+log = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """
+    Delete V1 Content Libraries (default all) index entires.
+    Specfiy --all for all libraries, or space-seperated library ids for specific libraries.
+    Note this will leave orphans behind in mongo. use mongo prune to clean them up.
+
+    Example usage:
+        ./manage.py cms delete_v1_libraries 'library-v1:edx+eaa'
+        ./manage.py cms delete_v1_libraries --all
+
+    Note:
+       This Command also produces an "output file" which contains the mapping of locators and the status of the copy.
+    """
+
+    help = dedent(__doc__)
+    CONFIRMATION_PROMPT = "Deleting all libraries might be a time consuming operation. Do you want to continue?"
+
+    def add_arguments(self, parser):
+        """arguements for command"""
+
+        parser.add_argument(
+            'library_ids',
+            nargs='*',
+            help='A space-seperated list of v1 library ids to delete'
+        )
+        parser.add_argument(
+            '--all',
+            action='store_true',
+            dest='all',
+            help='Delete all libraries'
+        )
+        parser.add_argument(
+            'output_csv',
+            nargs='?',
+            default=None,
+            help='a file path to write the tasks output to. Without this the result is simply logged.'
+        )
+
+    def _parse_library_key(self, raw_value):
+        """ Parses library key from string """
+        result = CourseKey.from_string(raw_value)
+
+        if not isinstance(result, LibraryLocator):
+            raise CommandError(f"Argument {raw_value} is not a library key")
+        return result
+
+    def handle(self, *args, **options):  # lint-amnesty, pylint: disable=unused-argument
+        """Parse args and generate tasks for deleting content."""
+
+        if (not options['library_ids'] and not options['all']) or (options['library_ids'] and options['all']):
+            raise CommandError("delete_v1_libraries requires one or more <library_id>s or the --all flag.")
+
+        if options['all']:
+            store = modulestore()
+            if query_yes_no(self.CONFIRMATION_PROMPT, default="no"):
+                v1_library_keys = [
+                    library.location.library_key.replace(branch=None) for library in store.get_libraries()
+                ]
+            else:
+                return
+        else:
+            v1_library_keys = list(map(self._parse_library_key, options['library_ids']))
+
+        delete_libary_task_group = group([
+            delete_v1_library.s(str(v1_library_key)) for v1_library_key in v1_library_keys
+        ])
+
+        group_result = delete_libary_task_group.apply_async().get()
+        log.info(group_result)
+        if options['output_csv']:
+            with open(options['output_csv'][0], 'w', encoding='utf-8', newline='') as output_writer:
+                output_writer.writerow("v1_library_id", "v2_library_id", "status", "error_msg")
+                for result in group_result:
+                    output_writer.write(result.keys())

cms/djangoapps/contentstore/tasks.py

@@ -75,6 +75,10 @@ from .outlines_regenerate import CourseOutlineRegenerate
 from .toggles import bypass_olx_failure_enabled
 from .utils import course_import_olx_validation_is_enabled
 
+
+from cms.djangoapps.contentstore.utils import delete_course  # lint-amnesty, pylint: disable=wrong-import-order
+from xmodule.modulestore import ModuleStoreEnum  # lint-amnesty, pylint: disable=wrong-import-order
+
 User = get_user_model()
 
 LOGGER = get_task_logger(__name__)
@@ -969,3 +973,30 @@ def create_v2_library_from_v1_library(v1_library_key_string, collection_uuid):
         "status": "SUCCESS",
         "msg": None
     }
+
+
+@shared_task(time_limit=30)
+@set_code_owner_attribute
+def delete_v1_library(v1_library_key_string):
+    """
+    Delete a v1 library index by key string.
+    """
+    v1_library_key = CourseKey.from_string(v1_library_key_string)
+    if not modulestore().get_library(v1_library_key):
+        raise KeyError(f"Library not found: {v1_library_key}")
+    try:
+        delete_course(v1_library_key, ModuleStoreEnum.UserID.mgmt_command, True)
+        LOGGER.info(f"Deleted course {v1_library_key}")
+    except Exception as error:  # lint-amnesty, pylint: disable=broad-except
+        return {
+            "v1_library_id": v1_library_key_string,
+            "status": "FAILED",
+            "msg":
+            f"Error occurred deleting library: {str(error)}"
+        }
+
+    return {
+        "v1_library_id": v1_library_key_string,
+        "status": "SUCCESS",
+        "msg": "SUCCESS"
+    }