feat: migrate Insights course index APIs to Snowflake

analytics_data_api/insights_snowflake/mappers/course_summaries.py

@@ -0,0 +1,117 @@
+"""Map Snowflake course summary rows into the existing API response shape."""
+
+from itertools import groupby
+
+from analytics_data_api.constants import enrollment_modes
+
+COUNT_FIELDS = ('count', 'cumulative_count', 'count_change_7_days', 'passing_users')
+SUMMARY_META_FIELDS = (
+    'catalog_course_title',
+    'catalog_course',
+    'start_time',
+    'end_time',
+    'pacing_type',
+    'availability',
+)
+
+
+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 _count_value(row, name):
+    """Return an integer count, treating missing nullable Snowflake counts as zero."""
+    return int(_row_value(row, name) or 0)
+
+
+def _base_course_summary(course_id):
+    """Return the default course summary shape used by the existing API."""
+    summary = {
+        'course_id': course_id,
+        'created': None,
+        'enrollment_modes': {},
+    }
+    summary.update({field: 0 for field in COUNT_FIELDS})
+    summary['enrollment_modes'].update({
+        mode: {
+            count_field: 0 for count_field in COUNT_FIELDS
+        } for mode in enrollment_modes.ALL
+    })
+    return summary
+
+
+def _programs_by_course(program_rows):
+    """Return program IDs grouped by course ID."""
+    programs = {}
+    for row in program_rows or []:
+        programs.setdefault(_row_value(row, 'course_id'), []).append(_row_value(row, 'program_id'))
+    return programs
+
+
+def _recent_counts_by_course(recent_rows):
+    """Return recent enrollment counts keyed by course ID."""
+    return {
+        _row_value(row, 'course_id'): _count_value(row, 'count')
+        for row in recent_rows or []
+    }
+
+
+def _postprocess_course_summary(summary, exclude=None):
+    """Apply existing course summary response compatibility rules."""
+    modes = summary['enrollment_modes']
+    prof_no_id_mode = modes.pop(enrollment_modes.PROFESSIONAL_NO_ID, {})
+    prof_mode = modes[enrollment_modes.PROFESSIONAL]
+    for count_key in COUNT_FIELDS:
+        prof_mode[count_key] = prof_mode.get(count_key, 0) + prof_no_id_mode.pop(count_key, 0)
+
+    if summary['availability'] == 'Starting Soon':
+        summary['availability'] = 'Upcoming'
+
+    for field in exclude or []:
+        for mode in summary['enrollment_modes']:
+            summary['enrollment_modes'][mode].pop(field, None)
+
+    return summary
+
+
+def map_course_summary_rows(summary_rows, program_rows=None, recent_rows=None, exclude=None):
+    """Group course summary rows into one API item per course."""
+    rows = sorted(
+        summary_rows or [],
+        key=lambda row: (
+            _row_value(row, 'course_id') or '',
+            _row_value(row, 'enrollment_mode') or '',
+        ),
+    )
+    programs = _programs_by_course(program_rows) if program_rows is not None else None
+    recent_counts = _recent_counts_by_course(recent_rows) if recent_rows is not None else None
+    summaries = []
+
+    for course_id, group in groupby(rows, lambda row: _row_value(row, 'course_id')):
+        summary = _base_course_summary(course_id)
+
+        for row in group:
+            for field in SUMMARY_META_FIELDS:
+                summary[field] = _row_value(row, field)
+
+            mode = _row_value(row, 'enrollment_mode')
+            summary['enrollment_modes'][mode] = {field: _count_value(row, field) for field in COUNT_FIELDS}
+            created = _row_value(row, 'created')
+            summary['created'] = max(created, summary['created']) if summary['created'] else created
+            summary.update({
+                field: summary[field] + _count_value(row, field)
+                for field in COUNT_FIELDS
+            })
+
+        if recent_counts is not None:
+            summary['recent_count_change'] = summary['count'] - recent_counts.get(course_id, 0)
+
+        if programs is not None:
+            summary['programs'] = programs.get(course_id, [])
+
+        summaries.append(_postprocess_course_summary(summary, exclude=exclude))
+
+    return summaries

analytics_data_api/insights_snowflake/mappers/programs.py

@@ -0,0 +1,42 @@
+"""Map Snowflake program metadata rows into the existing API response shape."""
+
+from itertools import groupby
+
+
+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 map_program_metadata_rows(rows):
+    """Group program metadata rows into one API item per program."""
+    rows = sorted(
+        rows or [],
+        key=lambda row: (
+            _row_value(row, 'program_id') or '',
+            _row_value(row, 'course_id') or '',
+        ),
+    )
+    programs = []
+
+    for program_id, group in groupby(rows, lambda row: _row_value(row, 'program_id')):
+        item = {
+            'program_id': program_id,
+            'program_type': '',
+            'program_title': '',
+            'created': None,
+            'course_ids': [],
+        }
+
+        for row in group:
+            item['program_type'] = _row_value(row, 'program_type')
+            item['program_title'] = _row_value(row, 'program_title')
+            item['course_ids'].append(_row_value(row, 'course_id'))
+            created = _row_value(row, 'created')
+            item['created'] = max(created, item['created']) if item['created'] else created
+
+        programs.append(item)
+
+    return programs

analytics_data_api/insights_snowflake/queries/course_summaries.py

@@ -0,0 +1,100 @@
+"""Snowflake queries for course summary metadata."""
+
+import datetime
+
+from analytics_data_api.insights_snowflake.client import fetch_all, get_qualified_table_name
+
+COURSE_META_SUMMARY_ENROLLMENT_TABLE = 'COURSE_META_SUMMARY_ENROLLMENT'
+COURSE_PROGRAM_METADATA_TABLE = 'COURSE_PROGRAM_METADATA'
+COURSE_ENROLLMENT_DAILY_TABLE = 'COURSE_ENROLLMENT_DAILY'
+
+
+def _date_value(value):
+    """Return a date value for date-filtered Snowflake course summary queries."""
+    if isinstance(value, datetime.datetime):
+        return value.date()
+    return value
+
+
+def _in_filter(column_name, param_prefix, values, prefix='WHERE'):
+    """Return a parameterized Snowflake IN filter for controlled columns."""
+    if not values:
+        return '', {}
+
+    params = {}
+    placeholders = []
+    for index, value in enumerate(values):
+        param_name = '{}_{}'.format(param_prefix, index)
+        params[param_name] = value
+        placeholders.append('%({})s'.format(param_name))
+
+    return '{} {} IN ({})'.format(prefix, column_name, ', '.join(placeholders)), params
+
+
+def get_course_summary_rows(course_ids=None):
+    """Return Snowflake rows for course summary enrollment metadata."""
+    table_name = get_qualified_table_name(COURSE_META_SUMMARY_ENROLLMENT_TABLE)
+    where_clause, params = _in_filter('course_id', 'course_id', course_ids)
+    sql = """
+SELECT
+    course_id,
+    catalog_course_title,
+    catalog_course,
+    start_time,
+    end_time,
+    pacing_type,
+    availability,
+    enrollment_mode,
+    "COUNT" AS count,
+    cumulative_count,
+    count_change_7_days,
+    passing_users,
+    created
+FROM {table_name}
+{where_clause}
+ORDER BY course_id, enrollment_mode
+""".format(table_name=table_name, where_clause=where_clause)
+
+    return fetch_all(sql, params)
+
+
+def get_course_summary_program_rows(course_ids=None):
+    """Return Snowflake program metadata rows for course summaries."""
+    table_name = get_qualified_table_name(COURSE_PROGRAM_METADATA_TABLE)
+    where_clause, params = _in_filter('course_id', 'course_id', course_ids)
+    sql = """
+SELECT
+    course_id,
+    program_id,
+    program_type,
+    program_title,
+    created
+FROM {table_name}
+{where_clause}
+ORDER BY course_id, program_id
+""".format(table_name=table_name, where_clause=where_clause)
+
+    return fetch_all(sql, params)
+
+
+def get_course_recent_enrollment_rows(course_ids=None, recent_date=None):
+    """Return Snowflake course enrollment rows for the requested recent date."""
+    table_name = get_qualified_table_name(COURSE_ENROLLMENT_DAILY_TABLE)
+    course_filter, course_params = _in_filter('course_id', 'course_id', course_ids, prefix='AND')
+    params = {
+        'recent_date': _date_value(recent_date),
+    }
+    params.update(course_params)
+    sql = """
+SELECT
+    course_id,
+    "DATE" AS date,
+    "COUNT" AS count,
+    created
+FROM {table_name}
+WHERE "DATE" = %(recent_date)s
+{course_filter}
+ORDER BY course_id
+""".format(table_name=table_name, course_filter=course_filter)
+
+    return fetch_all(sql, params)

analytics_data_api/insights_snowflake/queries/programs.py

@@ -0,0 +1,39 @@
+"""Snowflake queries for program metadata."""
+
+from analytics_data_api.insights_snowflake.client import fetch_all, get_qualified_table_name
+
+COURSE_PROGRAM_METADATA_TABLE = 'COURSE_PROGRAM_METADATA'
+
+
+def _in_filter(column_name, param_prefix, values):
+    """Return a parameterized Snowflake IN filter for controlled columns."""
+    if not values:
+        return '', {}
+
+    params = {}
+    placeholders = []
+    for index, value in enumerate(values):
+        param_name = '{}_{}'.format(param_prefix, index)
+        params[param_name] = value
+        placeholders.append('%({})s'.format(param_name))
+
+    return 'WHERE {} IN ({})'.format(column_name, ', '.join(placeholders)), params
+
+
+def get_program_metadata_rows(program_ids=None):
+    """Return Snowflake rows for course program metadata."""
+    table_name = get_qualified_table_name(COURSE_PROGRAM_METADATA_TABLE)
+    where_clause, params = _in_filter('program_id', 'program_id', program_ids)
+    sql = """
+SELECT
+    program_id,
+    program_type,
+    program_title,
+    course_id,
+    created
+FROM {table_name}
+{where_clause}
+ORDER BY program_id, course_id
+""".format(table_name=table_name, where_clause=where_clause)
+
+    return fetch_all(sql, params)

analytics_data_api/insights_snowflake/service.py

@@ -1,21 +1,29 @@
 """Service functions for Snowflake-backed Insights endpoints."""
 
 from analytics_data_api.insights_snowflake.mappers.activity import map_course_activity_weekly_rows
+from analytics_data_api.insights_snowflake.mappers.course_summaries import map_course_summary_rows
 from analytics_data_api.insights_snowflake.mappers.enrollment import (
     map_course_enrollment_daily_rows,
     map_course_enrollment_education_rows,
     map_course_enrollment_gender_rows,
     map_course_enrollment_location_rows,
     map_course_enrollment_mode_rows,
 )
+from analytics_data_api.insights_snowflake.mappers.programs import map_program_metadata_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,
+    get_course_summary_program_rows,
+    get_course_summary_rows,
+)
 from analytics_data_api.insights_snowflake.queries.enrollment import (
     get_course_enrollment_daily_rows,
     get_course_enrollment_education_rows,
     get_course_enrollment_gender_rows,
     get_course_enrollment_location_rows,
     get_course_enrollment_mode_rows,
 )
+from analytics_data_api.insights_snowflake.queries.programs import get_program_metadata_rows
 
 
 def get_course_activity_weekly(course_id, start_date=None, end_date=None):
@@ -52,3 +60,26 @@ def get_course_enrollment_location(course_id, start_date=None, end_date=None):
     """Return course enrollment location counts in the existing API response shape."""
     rows = get_course_enrollment_location_rows(course_id, start_date=start_date, end_date=end_date)
     return map_course_enrollment_location_rows(rows)
+
+
+def get_program_metadata(program_ids=None):
+    """Return program metadata in the existing API response shape."""
+    rows = get_program_metadata_rows(program_ids=program_ids)
+    return map_program_metadata_rows(rows)
+
+
+def get_course_summaries(course_ids=None, include_programs=False, recent_date=None, exclude=None):
+    """Return course summaries in the existing API response shape."""
+    summary_rows = get_course_summary_rows(course_ids=course_ids)
+    program_rows = get_course_summary_program_rows(course_ids=course_ids) if include_programs else None
+    recent_rows = get_course_recent_enrollment_rows(
+        course_ids=course_ids,
+        recent_date=recent_date,
+    ) if recent_date else None
+
+    return map_course_summary_rows(
+        summary_rows,
+        program_rows=program_rows,
+        recent_rows=recent_rows,
+        exclude=exclude,
+    )