feat: entity links api

openedx_learning/apps/authoring/linking/api.py

@@ -18,6 +18,7 @@
 __all__ = [
     'delete_entity_link',
     'get_entity_links',
+    'get_entity_links_by_downstream',
     'get_or_create_learning_context_link_status',
     'update_or_create_entity_link',
     'update_learning_context_link_status',
@@ -127,3 +128,16 @@ def update_or_create_entity_link(
 def delete_entity_link(downstream_usage_key: str):
     """Detele upstream->downstream entity link from database"""
     PublishableEntityLink.objects.filter(downstream_usage_key=downstream_usage_key).delete()
+
+
+def get_entity_links_by_downstream(downstream_context_key: str) -> QuerySet[PublishableEntityLink]:
+    """
+    Filter publishable entity links by given downstream_context_key.
+    Returns latest published version number of upstream_block as well.
+    """
+    return PublishableEntityLink.objects.filter(
+        downstream_context_key=downstream_context_key
+    ).select_related(
+        "upstream_block__published__version",
+        "upstream_block__learning_package"
+    )

openedx_learning/apps/authoring/linking/models.py

@@ -62,6 +62,19 @@ class PublishableEntityLink(models.Model):
     def __str__(self):
         return f"{self.upstream_usage_key}->{self.downstream_usage_key}"
 
+    @property
+    def upstream_version(self) -> int | None:
+        version_num = None
+        if hasattr(self.upstream_block, 'published'):
+            if hasattr(self.upstream_block.published, 'version'):
+                if hasattr(self.upstream_block.published.version, 'version_num'):
+                    version_num = self.upstream_block.published.version.version_num
+        return version_num
+
+    @property
+    def upstream_context_title(self) -> str:
+        return self.upstream_block.learning_package.title
+
     class Meta:
         constraints = [
             # A downstream entity can only link to single upstream entity

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

@@ -39,6 +39,7 @@ def setUpTestData(cls) -> None:
             created=cls.now,
             created_by=None,
         )
+        publishing_api.publish_all_drafts(cls.learning_package.id)
 
     def test_get_or_create_learning_context_link_status(self) -> None:
         """
@@ -100,3 +101,37 @@ def test_delete_entity_link(self) -> None:
         assert PublishableEntityLink.objects.filter(downstream_usage_key=downstream_usage_key).exists()
         linking_api.delete_entity_link(downstream_usage_key)
         assert not PublishableEntityLink.objects.filter(downstream_usage_key=downstream_usage_key).exists()
+
+    def test_get_entity_links_by_downstream(self) -> None:
+        """
+        Test get_entity_links_by_downstream api.
+        """
+        downstream_context_key = "course-v1:test-course-1"
+        downstream_context_key_2 = "course-v1:test-course-1"
+        entity_args_1 = {
+            "upstream_usage_key": "u-usage-1",
+            "upstream_context_key": "u-context-1",
+            "downstream_usage_key": "d-usage-1",
+            "downstream_context_key": downstream_context_key,
+            "downstream_context_title": "Course title 1",
+            "version_synced": 1,
+        }
+        entity_args_2 = {
+            "upstream_usage_key": "u-usage-1",
+            "upstream_context_key": "u-context-1",
+            "downstream_usage_key": "d-usage-2",
+            "downstream_context_key": downstream_context_key,
+            "downstream_context_title": "Course title 2",
+            "version_synced": 1,
+        }
+        # Create new links
+        linking_api.update_or_create_entity_link(self.html_component, **entity_args_1)  # type: ignore[arg-type]
+        linking_api.update_or_create_entity_link(self.html_component, **entity_args_2)  # type: ignore[arg-type]
+        entity_args_1["downstream_context_key"] = downstream_context_key_2
+        entity_args_2["downstream_context_key"] = downstream_context_key_2
+        linking_api.update_or_create_entity_link(self.html_component, **entity_args_1)  # type: ignore[arg-type]
+        linking_api.update_or_create_entity_link(self.html_component, **entity_args_2)  # type: ignore[arg-type]
+        with self.assertNumQueries(1):
+            links = linking_api.get_entity_links_by_downstream(downstream_context_key)
+            for link in links:
+                assert link.upstream_version == 1