feat: [AXM-297] Add progress to assignments in BlocksInfoInCourseView API

lms/djangoapps/mobile_api/course_info/constants.py

@@ -0,0 +1,5 @@
+"""
+Common constants for the `course_info` API.
+"""
+
+BLOCK_STRUCTURE_CACHE_TIMEOUT = 60 * 60  # 1 hour

lms/djangoapps/mobile_api/course_info/utils.py

@@ -0,0 +1,29 @@
+"""
+Common utils for the `course_info` API.
+"""
+
+from django.core.cache import cache
+
+from lms.djangoapps.courseware.access import has_access
+from lms.djangoapps.grades.api import CourseGradeFactory
+from openedx.core.djangoapps.content.block_structure.api import get_block_structure_manager
+
+
+def calculate_progress(user, course_id, cache_timeout):
+    """
+    Calculate the progress of the user in the course.
+    """
+    is_staff = bool(has_access(user, 'staff', course_id))
+
+    cache_key = f'course_block_structure_{str(course_id)}_{user.id}'
+    collected_block_structure = cache.get(cache_key)
+    if not collected_block_structure:
+        collected_block_structure = get_block_structure_manager(course_id).get_collected()
+        cache.set(cache_key, collected_block_structure, cache_timeout)
+
+    course_grade = CourseGradeFactory().read(user, collected_block_structure=collected_block_structure)
+
+    # recalculate course grade from visible grades (stored grade was calculated over all grades, visible or not)
+    course_grade.update(visible_grades_only=True, has_staff_access=is_staff)
+    subsection_grades = list(course_grade.subsection_grades.values())
+    return subsection_grades

lms/djangoapps/mobile_api/course_info/views.py

@@ -3,7 +3,7 @@
 """
 
 import logging
-from typing import Optional, Union
+from typing import Dict, Optional, Union
 
 import django
 from django.contrib.auth import get_user_model
@@ -20,11 +20,13 @@
 from lms.djangoapps.courseware.courses import get_course_info_section_block
 from lms.djangoapps.course_goals.models import UserActivity
 from lms.djangoapps.course_api.blocks.views import BlocksInCourseView
+from lms.djangoapps.mobile_api.course_info.constants import BLOCK_STRUCTURE_CACHE_TIMEOUT
 from lms.djangoapps.mobile_api.course_info.serializers import (
     CourseInfoOverviewSerializer,
     CourseAccessSerializer,
     MobileCourseEnrollmentSerializer
 )
+from lms.djangoapps.mobile_api.course_info.utils import calculate_progress
 from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
 from openedx.core.lib.api.view_utils import view_auth_classes
 from openedx.core.lib.xblock_utils import get_course_update_items
@@ -357,6 +359,12 @@ def list(self, request, **kwargs):  # pylint: disable=W0221
 
             course_info_context = {}
             if requested_user := self.get_requested_user(request.user, requested_username):
+                self._extend_sequential_info_with_assignment_progress(
+                    requested_user,
+                    course_key,
+                    response.data['blocks'],
+                )
+
                 course_info_context = {
                     'user': requested_user
                 }
@@ -380,3 +388,44 @@ def list(self, request, **kwargs):  # pylint: disable=W0221
 
             response.data.update(course_data)
         return response
+
+    @staticmethod
+    def _extend_sequential_info_with_assignment_progress(
+        requested_user: User,
+        course_id: CourseKey,
+        blocks_info_data: Dict[str, Dict],
+    ) -> None:
+        """
+        Extends sequential xblock info with assignment's name and progress.
+        """
+        subsection_grades = calculate_progress(requested_user, course_id, BLOCK_STRUCTURE_CACHE_TIMEOUT)
+        grades_with_locations = {str(grade.location): grade for grade in subsection_grades}
+
+        for block_id, block_info in blocks_info_data.items():
+            if block_info['type'] == 'sequential':
+                grade = grades_with_locations.get(block_id)
+                if grade:
+                    points_earned = (
+                        grade.graded_total.earned
+                        if grades_with_locations[block_id].graded
+                        else 0
+                    )
+                    points_possible = (
+                        grade.graded_total.possible
+                        if grades_with_locations[block_id].graded
+                        else 0
+                    )
+                    assignment_type = grade.format
+                else:
+                    points_earned, points_possible = 0, 0
+                    assignment_type = None
+
+                block_info.update(
+                    {
+                        'assignment_progress': {
+                            'assignment_type': assignment_type,
+                            'num_points_earned': points_earned,
+                            'num_points_possible': points_possible,
+                        }
+                    }
+                )

lms/djangoapps/mobile_api/tests/test_course_info_views.py

@@ -422,3 +422,31 @@ def test_course_modes(self):
 
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertListEqual(response.data['course_modes'], expected_course_modes)
+
+    def test_extend_sequential_info_with_assignment_progress_get_only_sequential(self) -> None:
+        response = self.verify_response(url=self.url, params={'block_types_filter': 'sequential'})
+
+        expected_results = (
+            {
+                'assignment_type': 'Lecture Sequence',
+                'num_points_earned': 0.0,
+                'num_points_possible': 0.0
+            },
+            {
+                'assignment_type': None,
+                'num_points_earned': 0.0,
+                'num_points_possible': 0.0
+            },
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        for sequential_info, assignment_progress in zip(response.data['blocks'].values(), expected_results):
+            self.assertDictEqual(sequential_info['assignment_progress'], assignment_progress)
+
+    @ddt.data('chapter', 'vertical', 'problem', 'video', 'html')
+    def test_extend_sequential_info_with_assignment_progress_for_other_types(self, block_type: 'str') -> None:
+        response = self.verify_response(url=self.url, params={'block_types_filter': block_type})
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        for block_info in response.data['blocks'].values():
+            self.assertNotEqual('assignment_progress', block_info)