Merge branch 'master' into hamza/PROD-4153

course_discovery/apps/core/api_client/lms.py

@@ -209,3 +209,24 @@ def get_blocks_metadata(self, block_id: str, **kwargs):
         }
         cache_key = get_cache_key(block_id=block_id, resource=resource)
         return self._get_blocks_data(block_id, cache_key, query_parameters, resource)
+
+    def get_course_run_translations(self, course_run_id: str):
+        """
+        Get translation information for a given course run.
+
+        Args:
+            course_run_id (str): The course run ID to fetch translation information for.
+
+        Returns:
+            dict: A dictionary containing the translation information or an empty dict on error.
+        """
+        resource = settings.LMS_API_URLS['translations']
+        resource_url = urljoin(self.lms_url, resource)
+
+        try:
+            response = self.client.get(resource_url, params={'course_id': course_run_id})
+            response.raise_for_status()
+            return response.json()
+        except RequestException as e:
+            logger.exception(f'Failed to fetch translation data for course run [{course_run_id}]: {e}')
+            return {}

course_discovery/apps/core/tests/test_api_clients.py

@@ -87,8 +87,6 @@ def test_get_api_access_request_with_invalid_response(self):
         Verify that `get_api_access_request` returns None when api_access_request
         returns an invalid response.
         """
-        # API response without proper paginated structure.
-        # Following is an invalid response.
         sample_invalid_response = {
             'id': 1,
             'created': '2017-09-25T08:37:05.872566Z',
@@ -237,3 +235,45 @@ def test_get_blocks_data_cache_hit(self):
         assert self.lms.get_blocks_data(self.block_id) == data['blocks']
         assert self.lms.get_blocks_data(self.block_id) == data['blocks']
         assert len(responses.calls) == 1
+
+    @responses.activate
+    def test_get_course_run_translations(self):
+        """
+        Verify that `get_course_run_translations` returns correct translation data.
+        """
+        course_run_id = 'course-v1:edX+DemoX+Demo_Course'
+        translation_data = {
+            "en": {"title": "Course Title", "language": "English"},
+            "fr": {"title": "Titre du cours", "language": "French"}
+        }
+        resource = settings.LMS_API_URLS['translations']
+        resource_url = urljoin(self.partner.lms_url, resource)
+
+        responses.add(
+            responses.GET,
+            resource_url,
+            json=translation_data,
+            status=200
+        )
+
+        result = self.lms.get_course_run_translations(course_run_id)
+        assert result == translation_data
+
+    @responses.activate
+    def test_get_course_run_translations_with_error(self):
+        """
+        Verify that get_course_run_translations returns an empty dictionary when there's an error.
+        """
+        course_run_id = 'course-v1:edX+DemoX+Demo_Course'
+        resource = settings.LMS_API_URLS['translations']
+        resource_url = urljoin(self.partner.lms_url, resource)
+
+        responses.add(
+            responses.GET,
+            resource_url,
+            status=500
+        )
+
+        result = self.lms.get_course_run_translations(course_run_id)
+        assert result == {}
+        assert 'Failed to fetch translation data for course run [%s]' % course_run_id in self.log_messages['error'][0]

course_discovery/apps/course_metadata/gspread_client.py

@@ -80,17 +80,19 @@ def _write_rows(self, sheet_tab, headers, csv_data):
             csv_data: The data to be written in the worksheet, as a list of dictionaries, where
             each dictionary represents a row
         """
-        for row in csv_data:
-            sheet_tab.append_row(
-                [
-                    (
-                        row.get(header).replace('\"', '\"\"')  # double quote escape to preserve " in values
-                        if isinstance(row.get(header), str)
-                        else row.get(header)
-                    )
-                    for header in headers
-                ]
-            )
+        rows = [
+            [
+                row.get(header).replace('\"', '\"\"') if isinstance(row.get(header), str) else row.get(header)
+                for header in headers
+            ]
+            for row in csv_data
+        ]
+        try:
+            sheet_tab.append_rows(rows)
+        except gspread.exceptions.APIError as e:
+            logger.exception(f"[Spread Sheet Write Error]: APIError occurred while writing rows: {e}")
+        except Exception as e:  # pylint: disable=broad-except
+            logger.exception(f"[Spread Sheet Write Error]: Exception occurred while writing rows: {e}")
 
     def write_data(self, config, csv_headers, csv_data, overwrite):
         """

course_discovery/apps/course_metadata/management/commands/populate_product_catalog.py

@@ -7,7 +7,7 @@
 from django.db.models import Prefetch
 
 from course_discovery.apps.course_metadata.gspread_client import GspreadClient
-from course_discovery.apps.course_metadata.models import Course, CourseType, SubjectTranslation, Program
+from course_discovery.apps.course_metadata.models import Course, CourseType, Program, SubjectTranslation
 
 logger = logging.getLogger(__name__)