Backup and Restore Final Adjustments

openedx_learning/apps/authoring/backup_restore/api.py

@@ -3,23 +3,28 @@
 """
 import zipfile
 
+from django.contrib.auth.models import User as UserType  # pylint: disable=imported-auth-user
+
 from openedx_learning.apps.authoring.backup_restore.zipper import LearningPackageUnzipper, LearningPackageZipper
 from openedx_learning.apps.authoring.publishing.api import get_learning_package_by_key
 
 
 def create_zip_file(lp_key: str, path: str) -> None:
     """
     Creates a dump zip file for the given learning package key at the given path.
+    The zip file contains a TOML representation of the learning package and its contents.
 
     Can throw a NotFoundError at get_learning_package_by_key
     """
     learning_package = get_learning_package_by_key(lp_key)
     LearningPackageZipper(learning_package).create_zip(path)
 
 
-def load_dump_zip_file(path: str) -> None:
+def load_learning_package(path: str, key: str | None = None, user: UserType | None = None) -> dict:
     """
-    Loads a zip file derived from create_zip_file
+    Loads a learning package from a zip file at the given path.
+    Restores the learning package and its contents to the database.
+    Returns a dictionary with the status of the operation and any errors encountered.
     """
     with zipfile.ZipFile(path, "r") as zipf:
-        LearningPackageUnzipper(zipf).load()
+        return LearningPackageUnzipper(zipf, key, user).load()

openedx_learning/apps/authoring/backup_restore/management/commands/lp_dump.py

@@ -2,6 +2,7 @@
 Django management commands to handle backup learning packages (WIP)
 """
 import logging
+import time
 
 from django.core.management import CommandError
 from django.core.management.base import BaseCommand
@@ -28,8 +29,10 @@ def handle(self, *args, **options):
         if not file_name.lower().endswith(".zip"):
             raise CommandError("Output file name must end with .zip")
         try:
+            start_time = time.time()
             create_zip_file(lp_key, file_name)
-            message = f'{lp_key} written to {file_name}'
+            elapsed = time.time() - start_time
+            message = f'{lp_key} written to {file_name} (create_zip_file: {elapsed:.2f} seconds)'
             self.stdout.write(self.style.SUCCESS(message))
         except LearningPackage.DoesNotExist as exc:
             message = f"Learning package with key {lp_key} not found"

openedx_learning/apps/authoring/backup_restore/management/commands/lp_load.py

@@ -2,11 +2,13 @@
 Django management commands to handle restore learning packages (WIP)
 """
 import logging
+import time
 
+from django.contrib.auth.models import User as UserType  # pylint: disable=imported-auth-user
 from django.core.management import CommandError
 from django.core.management.base import BaseCommand
 
-from openedx_learning.apps.authoring.backup_restore.api import load_dump_zip_file
+from openedx_learning.apps.authoring.backup_restore.api import load_learning_package
 
 logger = logging.getLogger(__name__)
 
@@ -18,15 +20,28 @@ class Command(BaseCommand):
     help = 'Load a learning package from a zip file.'
 
     def add_arguments(self, parser):
-        parser.add_argument('file_name', type=str, help='The name of the input zip file to load.')
+        parser.add_argument('file_name', type=str, help='The path of the input zip file to load.')
+        parser.add_argument('username', type=str, help='The username of the user performing the load operation.')
 
     def handle(self, *args, **options):
         file_name = options['file_name']
+        username = options['username']
         if not file_name.lower().endswith(".zip"):
             raise CommandError("Input file name must end with .zip")
         try:
-            load_dump_zip_file(file_name)
-            message = f'{file_name} loaded successfully'
+            start_time = time.time()
+            # Create a tmp user to pass to the load function
+            user = UserType.objects.get(username=username)
+
+            result = load_learning_package(file_name, user=user)
+            duration = time.time() - start_time
+            if result["status"] == "error":
+                message = "Errors encountered during restore:\n"
+                log_buffer = result.get("log_file_error")
+                if log_buffer:
+                    message += log_buffer.getvalue()
+                raise CommandError(message)
+            message = f'{file_name} loaded successfully (duration: {duration:.2f} seconds)'
             self.stdout.write(self.style.SUCCESS(message))
         except FileNotFoundError as exc:
             message = f"Learning package file {file_name} not found: {exc}"

openedx_learning/apps/authoring/backup_restore/serializers.py

@@ -8,6 +8,34 @@
 from openedx_learning.apps.authoring.components import api as components_api
 
 
+class LearningPackageSerializer(serializers.Serializer):  # pylint: disable=abstract-method
+    """
+    Serializer for learning packages.
+
+    Note:
+        The `key` field is serialized, but it is generally not trustworthy for restoration.
+        During restore, a new key may be generated or overridden.
+    """
+    title = serializers.CharField(required=True)
+    key = serializers.CharField(required=True)
+    description = serializers.CharField(required=True, allow_blank=True)
+    created = serializers.DateTimeField(required=True, default_timezone=timezone.utc)
+
+
+class LearningPackageMetadataSerializer(serializers.Serializer):  # pylint: disable=abstract-method
+    """
+    Serializer for learning package metadata.
+
+    Note:
+        This serializer handles data exported to an archive (e.g., during backup),
+        but the metadata is not restored to the database and is meant solely for inspection.
+    """
+    format_version = serializers.IntegerField(required=True)
+    created_by = serializers.CharField(required=False, allow_null=True)
+    created_at = serializers.DateTimeField(required=True, default_timezone=timezone.utc)
+    origin_server = serializers.CharField(required=False, allow_null=True)
+
+
 class EntitySerializer(serializers.Serializer):  # pylint: disable=abstract-method
     """
     Serializer for publishable entities.

openedx_learning/apps/authoring/backup_restore/toml.py

@@ -72,7 +72,6 @@ def _get_toml_publishable_entity_table(
 
     The resulting content looks like:
         [entity]
-        uuid = "f8ea9bae-b4ed-4a84-ab4f-2b9850b59cd6"
         can_stand_alone = true
         key = "xblock.v1:problem:my_published_example"
 
@@ -88,7 +87,6 @@ def _get_toml_publishable_entity_table(
     a string-like TOML fragment rather than a complete TOML document.
     """
     entity_table = tomlkit.table()
-    entity_table.add("uuid", str(entity.uuid))
     entity_table.add("can_stand_alone", entity.can_stand_alone)
     # Add key since the toml filename doesn't show the real key
     entity_table.add("key", entity.key)
@@ -133,7 +131,6 @@ def toml_publishable_entity(
 
     The resulting content looks like:
         [entity]
-        uuid = "f8ea9bae-b4ed-4a84-ab4f-2b9850b59cd6"
         can_stand_alone = true
         key = "xblock.v1:problem:my_published_example"
 
@@ -143,17 +140,16 @@ def toml_publishable_entity(
         [entity.published]
         version_num = 1
 
+        [entity.container.section] (if applicable)
+
         # ### Versions
 
         [[version]]
         title = "My published problem"
-        uuid = "2e07511f-daa7-428a-9032-17fe12a77d06"
         version_num = 1
 
-        [version.container]
+        [version.container] (if applicable)
         children = []
-
-        [version.container.unit]
     """
     # Create the TOML representation for the entity itself
     entity_table = _get_toml_publishable_entity_table(entity, draft_version, published_version)
@@ -179,32 +175,25 @@ def toml_publishable_entity_version(version: PublishableEntityVersion) -> tomlki
     The resulting content looks like:
         [[version]]
         title = "My published problem"
-        uuid = "2e07511f-daa7-428a-9032-17fe12a77d06"
         version_num = 1
 
-        [version.container]
+        [version.container] (if applicable)
         children = []
 
-        [version.container.unit]
-        graded = true
-
-     Note: This function returns a tomlkit.items.Table, which represents
+    Note: This function returns a tomlkit.items.Table, which represents
     a string-like TOML fragment rather than a complete TOML document.
     """
     version_table = tomlkit.table()
     version_table.add("title", version.title)
-    version_table.add("uuid", str(version.uuid))
     version_table.add("version_num", version.version_num)
 
-    container_table = tomlkit.table()
-
-    children = []
     if hasattr(version, 'containerversion'):
+        # If the version has a container version, add its children
+        container_table = tomlkit.table()
         children = publishing_api.get_container_children_entities_keys(version.containerversion)
-    container_table.add("children", children)
-
-    version_table.add("container", container_table)
-    return version_table  # For use in AoT
+        container_table.add("children", children)
+        version_table.add("container", container_table)
+    return version_table
 
 
 def toml_collection(collection: Collection, entity_keys: list[str]) -> str:
@@ -245,36 +234,20 @@ def parse_learning_package_toml(content: str) -> dict:
     Parse the learning package TOML content and return a dict of its fields.
     """
     lp_data: Dict[str, Any] = tomlkit.parse(content)
+    return lp_data
 
-    # Validate the minimum required fields
-    if "learning_package" not in lp_data:
-        raise ValueError("Invalid learning package TOML: missing 'learning_package' section")
-    if "title" not in lp_data["learning_package"]:
-        raise ValueError("Invalid learning package TOML: missing 'title' in 'learning_package' section")
-    if "key" not in lp_data["learning_package"]:
-        raise ValueError("Invalid learning package TOML: missing 'key' in 'learning_package' section")
-    return lp_data["learning_package"]
 
-
-def parse_publishable_entity_toml(content: str) -> tuple[Dict[str, Any], list]:
+def parse_publishable_entity_toml(content: str) -> dict:
     """
     Parse the publishable entity TOML file and return a dict of its fields.
     """
     pe_data: Dict[str, Any] = tomlkit.parse(content)
-
-    # Validate the minimum required fields
-    if "entity" not in pe_data:
-        raise ValueError("Invalid publishable entity TOML: missing 'entity' section")
-    if "version" not in pe_data:
-        raise ValueError("Invalid publishable entity TOML: missing 'version' section")
-    return pe_data["entity"], pe_data.get("version", [])
+    return pe_data
 
 
 def parse_collection_toml(content: str) -> dict:
     """
     Parse the collection TOML content and return a dict of its fields.
     """
     collection_data: Dict[str, Any] = tomlkit.parse(content)
-    if "collection" not in collection_data:
-        raise ValueError("Invalid collection TOML: missing 'collection' section")
-    return collection_data["collection"]
+    return collection_data