feat: API to get a historic published version of a unit

Author: bradenmacdonald

openedx_learning/apps/authoring/containers/api.py

@@ -108,16 +108,10 @@ def create_next_defined_list(
         # 2. Only associate existent rows to the new entity list iff: the order is the same, the PublishableEntity is in entity_pks and versions are not pinned
         # 3. If the order is different for a row with the PublishableEntity, create new row with the same PublishableEntity for the new order
         # and associate the new row to the new entity list
-        current_rows = previous_entity_list.entitylistrow_set.all()
-        publishable_entities_in_rows = {row.entity.pk: row for row in current_rows}
         new_rows = []
         for order_num, entity_pk, entity_version_pk in zip(
             order_nums, entity_pks, entity_version_pks
         ):
-            row = publishable_entities_in_rows.get(entity_pk)
-            if row and row.order_num == order_num:
-                new_entity_list.entitylistrow_set.add(row)
-                continue
             new_rows.append(
                 EntityListRow(
                     entity_list=new_entity_list,

openedx_learning/apps/authoring/publishing/api.py

@@ -513,3 +513,18 @@ def filter_publishable_entities(
         entities = entities.filter(published__version__isnull=not has_published)
 
     return entities
+
+
+def get_published_version_as_of(entity_id: int, publish_log_id: int) -> PublishableEntityVersion | None:
+    """
+    Get the published version of the given entity, at a specific snapshot in the
+    history of this Learning Package, given by the PublishLog ID.
+
+    This is a semi-private function, only available to other apps in the
+    authoring package.
+    """
+    record = PublishLogRecord.objects.filter(
+        entity_id=entity_id,
+        publish_log_id__lte=publish_log_id,
+    ).order_by('-publish_log_id').first()
+    return record.new_version if record else None

openedx_learning/apps/authoring/units/api.py

@@ -9,6 +9,7 @@
 from openedx_learning.apps.authoring.components.models import Component, ComponentVersion
 from openedx_learning.apps.authoring.containers.models import EntityListRow
 from ..publishing import api as publishing_api
+from ..publishing.api import get_published_version_as_of
 from ..containers import api as container_api
 from .models import Unit, UnitVersion
 from django.db.models import QuerySet
@@ -29,6 +30,7 @@
     "UnitListEntry",
     "get_components_in_draft_unit",
     "get_components_in_published_unit",
+    "get_components_in_published_unit_as_of",
 ]
 
 
@@ -240,7 +242,7 @@ def get_components_in_published_unit(
 ) -> list[UnitListEntry]:
     """
     [ 🛑 UNSTABLE ]
-    Get the list of entities and their versions in the draft version of the
+    Get the list of entities and their versions in the published version of the
     given container.
     """
     assert isinstance(unit, Unit)
@@ -254,3 +256,35 @@ def get_components_in_published_unit(
         assert isinstance(component_version, ComponentVersion)
         entity_list.append(UnitListEntry(component_version=component_version, pinned=entry.pinned))
     return entity_list
+
+
+def get_components_in_published_unit_as_of(
+    unit: Unit,
+    publish_log_id: int,  # TODO: or UUID
+) -> list[UnitListEntry] | None:
+    """
+    [ 🛑 UNSTABLE ]
+    Get the list of entities and their versions in the published version of the
+    given container as of the given PublishLog version (which is essentially a
+    version for the entire learning package).
+    """
+    assert isinstance(unit, Unit)
+    unit_pub_entity_version = get_published_version_as_of(unit.publishable_entity, publish_log_id)
+    if unit_pub_entity_version is None:
+        return None  # This unit was not published as of the given PublishLog ID.
+    unit_version = unit_pub_entity_version.unitversion
+
+    entity_list = []
+    rows = unit_version.container_entity_version.defined_list.entitylistrow_set.order_by("order_num")
+    for row in rows:
+        if row.entity_version is not None:
+            component_version = row.entity_version.componentversion
+            assert isinstance(component_version, ComponentVersion)
+            entity_list.append(UnitListEntry(component_version=component_version, pinned=True))
+        else:
+            # Unpinned component - figure out what its latest published version was.
+            # This is not optimized. It could be done in one query per unit rather than one query per component.
+            pub_entity_version = get_published_version_as_of(row.entity, publish_log_id)
+            if pub_entity_version:
+                entity_list.append(UnitListEntry(component_version=pub_entity_version.componentversion, pinned=False))
+    return entity_list

tests/openedx_learning/apps/authoring/units/test_api.py

@@ -133,7 +133,7 @@ def test_create_next_unit_version_with_two_unpinned_components(self):
 
     def test_create_next_unit_version_with_unpinned_and_pinned_components(self):
         """
-        Test creating a unit version with one unpinned and one pinned component.
+        Test creating a unit version with one unpinned and one pinned 📌 component.
         """
         unit, unit_version = authoring_api.create_unit_and_version(
             learning_package_id=self.learning_package.id,
@@ -145,15 +145,15 @@ def test_create_next_unit_version_with_unpinned_and_pinned_components(self):
         unit_version_v2 = authoring_api.create_next_unit_version(
             unit=unit,
             title="Unit",
-            components=[self.component_1, self.component_2_v1],  # Note the "v1" pinning it to version 2
+            components=[self.component_1, self.component_2_v1],  # Note the "v1" pinning 📌 the second one to version 1
             created=self.now,
             created_by=None,
         )
         assert unit_version_v2.version_num == 2
         assert unit_version_v2 in unit.versioning.versions.all()
         assert authoring_api.get_components_in_draft_unit(unit) == [
             authoring_api.UnitListEntry(component_version=self.component_1_v1, pinned=False),
-            authoring_api.UnitListEntry(component_version=self.component_2_v1, pinned=True),  # Pinned to v1
+            authoring_api.UnitListEntry(component_version=self.component_2_v1, pinned=True),  # Pinned 📌 to v1
         ]
         assert authoring_api.get_components_in_published_unit(unit) is None
 
@@ -318,6 +318,58 @@ def test_query_count_of_contains_unpublished_changes(self):
     # Test viewing old snapshots of units and components by passing in a timestamp (or PublishLog PK) to a
     #     get_historic_unit() API?
 
+    def test_snapshots_of_published_unit(self):
+        """
+        Test that we can access snapshots of the historic published version of
+        units and their contents.
+        """
+        # At first the unit has one component (unpinned):
+        unit = self.create_unit_with_components([self.component_1])
+        self.modify_component(self.component_1, title="Component 1 as of checkpoint 1")
+
+        # Publish everything, creating Checkpoint 1
+        checkpoint_1 = authoring_api.publish_all_drafts(self.learning_package.id, message="checkpoint 1")
+
+        ########################################################################
+
+        # Now we update the title of the component.
+        self.modify_component(self.component_1, title="Component 1 as of checkpoint 2")
+        # Publish everything, creating Checkpoint 2
+        checkpoint_2 = authoring_api.publish_all_drafts(self.learning_package.id, message="checkpoint 2")
+        ########################################################################
+
+        # Now add a second component to the unit:
+        self.modify_component(self.component_1, title="Component 1 as of checkpoint 3")
+        self.modify_component(self.component_2, title="Component 2 as of checkpoint 3")
+        authoring_api.create_next_unit_version(
+            unit=unit,
+            title="Unit title in checkpoint 3",
+            components=[self.component_1, self.component_2],
+            created=self.now,
+        )
+        # Publish everything, creating Checkpoint 3
+        checkpoint_3 = authoring_api.publish_all_drafts(self.learning_package.id, message="checkpoint 3")
+        ########################################################################
+
+        # Modify the drafts, but don't publish:
+        self.modify_component(self.component_1, title="Component 1 draft")
+        self.modify_component(self.component_2, title="Component 2 draft")
+
+        # Now fetch the snapshots:
+        as_of_checkpoint_1 = authoring_api.get_components_in_published_unit_as_of(unit, checkpoint_1.pk)
+        assert [cv.component_version.title for cv in as_of_checkpoint_1] == [
+            "Component 1 as of checkpoint 1",
+        ]
+        as_of_checkpoint_2 = authoring_api.get_components_in_published_unit_as_of(unit, checkpoint_2.pk)
+        assert [cv.component_version.title for cv in as_of_checkpoint_2] == [
+            "Component 1 as of checkpoint 2",
+        ]
+        as_of_checkpoint_3 = authoring_api.get_components_in_published_unit_as_of(unit, checkpoint_3.pk)
+        assert [cv.component_version.title for cv in as_of_checkpoint_3] == [
+            "Component 1 as of checkpoint 3",
+            "Component 2 as of checkpoint 3",
+        ]
+
     def test_next_version_with_different_different_title(self):
         """Test creating a unit version with a different title.