feat: route Insights video APIs through Snowflake

analytics_data_api/insights_snowflake/mappers/videos.py

@@ -0,0 +1,62 @@
+"""Map Snowflake video rows into the existing API response shape."""
+
+COURSE_VIDEO_FIELDS = (
+    'pipeline_video_id',
+    'encoded_module_id',
+    'duration',
+    'segment_length',
+    'users_at_start',
+    'users_at_end',
+    'created',
+)
+VIDEO_TIMELINE_FIELDS = (
+    'segment',
+    'num_users',
+    'num_views',
+    'created',
+)
+INTEGER_FIELDS = (
+    'duration',
+    'segment_length',
+    'users_at_start',
+    'users_at_end',
+    'segment',
+    'num_users',
+    'num_views',
+)
+
+
+def _row_value(row, name):
+    """Return a row value from dictionary rows produced by the Snowflake client."""
+    if name in row:
+        return row[name]
+    return row[name.upper()]
+
+
+def _api_value(row, name):
+    """Return the API-compatible value for a Snowflake row field."""
+    value = _row_value(row, name)
+    if name in INTEGER_FIELDS:
+        return int(value)
+    return value
+
+
+def _map_rows(rows, fields):
+    """Map Snowflake rows to dictionaries containing only API response fields."""
+    return [
+        {
+            field: _api_value(row, field)
+            for field in fields
+        }
+        for row in rows or []
+    ]
+
+
+def map_course_video_rows(rows):
+    """Map Snowflake course video rows into existing API response dictionaries."""
+    return _map_rows(rows, COURSE_VIDEO_FIELDS)
+
+
+def map_video_timeline_rows(rows):
+    """Map Snowflake video timeline rows into existing API response dictionaries."""
+    return _map_rows(rows, VIDEO_TIMELINE_FIELDS)

analytics_data_api/insights_snowflake/queries/videos.py

@@ -0,0 +1,44 @@
+"""Snowflake queries for video engagement metrics."""
+
+from analytics_data_api.insights_snowflake.client import fetch_all, get_qualified_table_name
+
+VIDEO_TABLE = 'VIDEO'
+VIDEO_TIMELINE_TABLE = 'VIDEO_TIMELINE'
+
+
+def get_course_video_rows(course_id):
+    """Return Snowflake rows for videos in a course."""
+    table_name = get_qualified_table_name(VIDEO_TABLE)
+    sql = """
+SELECT
+    courserun_key AS course_id,
+    pipeline_video_id,
+    encoded_module_id,
+    duration,
+    segment_length,
+    users_at_start,
+    users_at_end,
+    created
+FROM {table_name}
+WHERE courserun_key = %(course_id)s
+ORDER BY pipeline_video_id
+""".format(table_name=table_name)
+
+    return fetch_all(sql, {'course_id': course_id})
+
+
+def get_video_timeline_rows(video_id):
+    """Return Snowflake rows for a video's timeline."""
+    table_name = get_qualified_table_name(VIDEO_TIMELINE_TABLE)
+    sql = """
+SELECT
+    segment,
+    num_users,
+    num_views,
+    created
+FROM {table_name}
+WHERE pipeline_video_id = %(video_id)s
+ORDER BY segment
+""".format(table_name=table_name)
+
+    return fetch_all(sql, {'video_id': video_id})

analytics_data_api/insights_snowflake/service.py

@@ -10,6 +10,7 @@
     map_course_enrollment_mode_rows,
 )
 from analytics_data_api.insights_snowflake.mappers.programs import map_program_metadata_rows
+from analytics_data_api.insights_snowflake.mappers.videos import map_course_video_rows, map_video_timeline_rows
 from analytics_data_api.insights_snowflake.queries.activity import get_course_activity_weekly_rows
 from analytics_data_api.insights_snowflake.queries.course_summaries import (
     get_course_recent_enrollment_rows,
@@ -24,6 +25,7 @@
     get_course_enrollment_mode_rows,
 )
 from analytics_data_api.insights_snowflake.queries.programs import get_program_metadata_rows
+from analytics_data_api.insights_snowflake.queries.videos import get_course_video_rows, get_video_timeline_rows
 
 
 def get_course_activity_weekly(course_id, start_date=None, end_date=None):
@@ -83,3 +85,15 @@ def get_course_summaries(course_ids=None, include_programs=False, recent_date=No
         recent_rows=recent_rows,
         exclude=exclude,
     )
+
+
+def get_course_videos(course_id):
+    """Return course videos in the existing API response shape."""
+    rows = get_course_video_rows(course_id)
+    return map_course_video_rows(rows)
+
+
+def get_video_timeline(video_id):
+    """Return video timeline metrics in the existing API response shape."""
+    rows = get_video_timeline_rows(video_id)
+    return map_video_timeline_rows(rows)

analytics_data_api/tests/test_insights_snowflake.py

@@ -18,6 +18,7 @@
     map_course_enrollment_mode_rows,
 )
 from analytics_data_api.insights_snowflake.mappers.programs import map_program_metadata_rows
+from analytics_data_api.insights_snowflake.mappers.videos import map_course_video_rows, map_video_timeline_rows
 from analytics_data_api.insights_snowflake.queries.activity import get_course_activity_weekly_rows
 from analytics_data_api.insights_snowflake.queries.course_summaries import (
     COURSE_ENROLLMENT_DAILY_TABLE,
@@ -44,6 +45,12 @@
     COURSE_PROGRAM_METADATA_TABLE,
     get_program_metadata_rows,
 )
+from analytics_data_api.insights_snowflake.queries.videos import (
+    VIDEO_TABLE,
+    VIDEO_TIMELINE_TABLE,
+    get_course_video_rows,
+    get_video_timeline_rows,
+)
 from analytics_data_api.insights_snowflake.response_headers import (
     DATA_SOURCE_HEADER,
     DATA_SOURCE_SNOWFLAKE,
@@ -57,7 +64,9 @@
     get_course_enrollment_location,
     get_course_enrollment_mode,
     get_course_summaries,
+    get_course_videos,
     get_program_metadata,
+    get_video_timeline,
 )
 from analytics_data_api.insights_snowflake.toggles import (
     COURSE_ACTIVITY_SNOWFLAKE_FLAG,
@@ -357,6 +366,51 @@ def test_get_program_metadata_rows_filters_program_ids(self, mock_get_table_name
         })
 
 
+class InsightsSnowflakeVideoQueryTests(SimpleTestCase):
+    """Cover video query construction with mocked Snowflake execution."""
+
+    @patch('analytics_data_api.insights_snowflake.queries.videos.fetch_all')
+    @patch(
+        'analytics_data_api.insights_snowflake.queries.videos.get_qualified_table_name',
+        Mock(return_value='PROD.INSIGHTS.VIDEO')
+    )
+    def test_get_course_video_rows_uses_expected_table(self, mock_fetch_all):
+        mock_fetch_all.return_value = [{'pipeline_video_id': 'video-1'}]
+        course_id = 'course-v1:edX+DemoX+Demo_Course'
+
+        rows = get_course_video_rows(course_id)
+
+        self.assertEqual(rows, [{'pipeline_video_id': 'video-1'}])
+        sql, params = mock_fetch_all.call_args[0]
+        self.assertIn('FROM PROD.INSIGHTS.VIDEO', sql)
+        self.assertIn('courserun_key AS course_id', sql)
+        self.assertIn('WHERE courserun_key = %(course_id)s', sql)
+        self.assertEqual(params, {'course_id': course_id})
+
+    @patch('analytics_data_api.insights_snowflake.queries.videos.fetch_all')
+    @patch('analytics_data_api.insights_snowflake.queries.videos.get_qualified_table_name')
+    def test_video_query_functions_use_expected_tables(self, mock_get_table_name, _mock_fetch_all):
+        mock_get_table_name.return_value = 'PROD.INSIGHTS.VIDEO_TABLE'
+
+        get_course_video_rows('course-v1:edX+DemoX+Demo_Course')
+
+        mock_get_table_name.assert_called_once_with(VIDEO_TABLE)
+        sql, params = _mock_fetch_all.call_args[0]
+        self.assertIn('ORDER BY pipeline_video_id', sql)
+        self.assertEqual(params, {'course_id': 'course-v1:edX+DemoX+Demo_Course'})
+
+        mock_get_table_name.reset_mock()
+        _mock_fetch_all.reset_mock()
+
+        get_video_timeline_rows('video-1')
+
+        mock_get_table_name.assert_called_once_with(VIDEO_TIMELINE_TABLE)
+        sql, params = _mock_fetch_all.call_args[0]
+        self.assertIn('WHERE pipeline_video_id = %(video_id)s', sql)
+        self.assertIn('ORDER BY segment', sql)
+        self.assertEqual(params, {'video_id': 'video-1'})
+
+
 class InsightsSnowflakeActivityMapperTests(SimpleTestCase):
     """Cover Snowflake activity row mapping into the existing API shape."""
 
@@ -772,6 +826,48 @@ def test_map_course_summary_rows_handles_null_sort_values(self):
         self.assertEqual(mapped_rows[0]['enrollment_modes'][enrollment_modes.PROFESSIONAL]['count'], 4)
 
 
+class InsightsSnowflakeVideoMapperTests(SimpleTestCase):
+    """Cover Snowflake video rows mapping into the existing API shapes."""
+
+    def test_map_course_video_rows_accepts_uppercase_snowflake_keys(self):
+        created = datetime.datetime(2014, 1, 2, tzinfo=datetime.timezone.utc)
+        rows = [{
+            'PIPELINE_VIDEO_ID': 'video-1',
+            'ENCODED_MODULE_ID': 'i4x-test-video-1',
+            'DURATION': 100,
+            'SEGMENT_LENGTH': 5,
+            'USERS_AT_START': 50,
+            'USERS_AT_END': 10,
+            'CREATED': created,
+        }]
+
+        self.assertEqual(map_course_video_rows(rows), [{
+            'pipeline_video_id': 'video-1',
+            'encoded_module_id': 'i4x-test-video-1',
+            'duration': 100,
+            'segment_length': 5,
+            'users_at_start': 50,
+            'users_at_end': 10,
+            'created': created,
+        }])
+
+    def test_map_video_timeline_rows(self):
+        created = datetime.datetime(2014, 1, 2, tzinfo=datetime.timezone.utc)
+        rows = [{
+            'segment': 0,
+            'num_users': 50,
+            'num_views': 100,
+            'created': created,
+        }]
+
+        self.assertEqual(map_video_timeline_rows(rows), [{
+            'segment': 0,
+            'num_users': 50,
+            'num_views': 100,
+            'created': created,
+        }])
+
+
 class InsightsSnowflakeServiceTests(SimpleTestCase):
     """Cover service orchestration without real Snowflake calls."""
 
@@ -862,6 +958,32 @@ def test_get_course_enrollment_location_calls_query_and_mapper(self):
             'analytics_data_api.insights_snowflake.service.map_course_enrollment_location_rows',
         )
 
+    @patch('analytics_data_api.insights_snowflake.service.map_course_video_rows')
+    @patch('analytics_data_api.insights_snowflake.service.get_course_video_rows')
+    def test_get_course_videos_calls_query_and_mapper(self, mock_get_rows, mock_map_rows):
+        raw_rows = [{'pipeline_video_id': 'video-1'}]
+        mapped_rows = [{'pipeline_video_id': 'video-1', 'duration': 100}]
+        mock_get_rows.return_value = raw_rows
+        mock_map_rows.return_value = mapped_rows
+
+        self.assertEqual(get_course_videos('course-v1:edX+DemoX+Demo_Course'), mapped_rows)
+
+        mock_get_rows.assert_called_once_with('course-v1:edX+DemoX+Demo_Course')
+        mock_map_rows.assert_called_once_with(raw_rows)
+
+    @patch('analytics_data_api.insights_snowflake.service.map_video_timeline_rows')
+    @patch('analytics_data_api.insights_snowflake.service.get_video_timeline_rows')
+    def test_get_video_timeline_calls_query_and_mapper(self, mock_get_rows, mock_map_rows):
+        raw_rows = [{'pipeline_video_id': 'video-1'}]
+        mapped_rows = [{'segment': 0, 'num_users': 50, 'num_views': 100}]
+        mock_get_rows.return_value = raw_rows
+        mock_map_rows.return_value = mapped_rows
+
+        self.assertEqual(get_video_timeline('video-1'), mapped_rows)
+
+        mock_get_rows.assert_called_once_with('video-1')
+        mock_map_rows.assert_called_once_with(raw_rows)
+
     @patch('analytics_data_api.insights_snowflake.service.map_program_metadata_rows')
     @patch('analytics_data_api.insights_snowflake.service.get_program_metadata_rows')
     def test_get_program_metadata_calls_query_and_mapper(self, mock_get_rows, mock_map_rows):

analytics_data_api/v0/tests/views/test_courses.py

@@ -1015,6 +1015,10 @@ def test_get_404(self):
 @ddt.ddt
 @set_databases
 class CourseVideosListViewTests(TestCaseWithAuthentication):
+    def tearDown(self):
+        thread_data.analyticsapi_database = getattr(settings, 'ANALYTICS_DATABASE', 'analytics')
+        super().tearDown()
+
     def _get_data(self, course_id):
         """
         Retrieve videos for a specified course.
@@ -1066,6 +1070,71 @@ def test_get(self, course_id):
         self.assertEqual(response.status_code, 200)
         self.assertListEqual(response.data, expected)
 
+    def test_get_uses_aurora_when_global_snowflake_flag_disabled(self):
+        course_id = CourseSamples.course_ids[0]
+        module_id = 'i4x-test-video-1'
+        video_id = 'v1d30'
+        created = timezone.now()
+        G(models.Video, course_id=course_id, encoded_module_id=module_id,
+          pipeline_video_id=video_id, duration=100, segment_length=1, users_at_start=50, users_at_end=10,
+          created=created)
+
+        with patch('analytics_data_api.v0.views.courses.is_insights_snowflake_enabled', return_value=False), \
+                patch('analytics_data_api.v0.views.courses.get_course_videos') as mock_get_videos:
+            response = self._get_data(course_id)
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response['X-Insights-Data-Source'], 'aurora')
+        mock_get_videos.assert_not_called()
+
+    def test_get_uses_snowflake_service_when_global_flag_enabled(self):
+        course_id = CourseSamples.course_ids[0]
+        created = timezone.now()
+        snowflake_data = [{
+            'pipeline_video_id': 'v1d30',
+            'encoded_module_id': 'i4x-test-video-1',
+            'duration': 100,
+            'segment_length': 1,
+            'users_at_start': 50,
+            'users_at_end': 10,
+            'created': created,
+        }]
+        expected = [{
+            'pipeline_video_id': 'v1d30',
+            'encoded_module_id': 'i4x-test-video-1',
+            'duration': 100,
+            'segment_length': 1,
+            'users_at_start': 50,
+            'users_at_end': 10,
+            'created': created.strftime(settings.DATETIME_FORMAT),
+        }]
+
+        with patch('analytics_data_api.v0.views.courses.is_insights_snowflake_enabled', return_value=True):
+            with patch(
+                    'analytics_data_api.v0.views.courses.get_course_videos',
+                    return_value=snowflake_data,
+            ) as mock_get_videos:
+                response = self.authenticated_get(f'/api/v1/courses/{course_id}/videos/')
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.data, expected)
+        self.assertEqual(response['X-Insights-Data-Source'], 'snowflake')
+        mock_get_videos.assert_called_once_with(course_id)
+
+    def test_get_returns_404_when_snowflake_service_returns_no_data(self):
+        course_id = CourseSamples.course_ids[0]
+
+        with patch('analytics_data_api.v0.views.courses.is_insights_snowflake_enabled', return_value=True):
+            with patch(
+                    'analytics_data_api.v0.views.courses.get_course_videos',
+                    return_value=[],
+            ) as mock_get_videos:
+                response = self.authenticated_get(f'/api/v1/courses/{course_id}/videos/')
+
+        self.assertEqual(response.status_code, 404)
+        self.assertEqual(response['X-Insights-Data-Source'], 'snowflake')
+        mock_get_videos.assert_called_once_with(course_id)
+
     def test_get_404(self):
         response = self._get_data('foo/bar/course')
         self.assertEqual(response.status_code, 404)