feat: route Insights enrollment APIs through Snowflake

analytics_data_api/insights_snowflake/mappers/enrollment.py

@@ -0,0 +1,149 @@
+"""Map Snowflake enrollment rows into the existing API response shapes."""
+
+from itertools import groupby
+
+from analytics_data_api.constants import enrollment_modes, genders
+from analytics_data_api.v0 import models
+
+GENDER_FIELD_MAP = {
+    'f': genders.FEMALE,
+    genders.FEMALE: genders.FEMALE,
+    'm': genders.MALE,
+    genders.MALE: genders.MALE,
+    'o': genders.OTHER,
+    genders.OTHER: genders.OTHER,
+}
+
+
+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 _copy_fields(row, field_names):
+    """Return an API-shaped dictionary with selected fields from a Snowflake row."""
+    return {
+        field_name: _row_value(row, field_name)
+        for field_name in field_names
+    }
+
+
+def _gender_field(gender):
+    """Return the existing API gender field for a Snowflake gender value."""
+    if gender is None:
+        return genders.UNKNOWN
+    return GENDER_FIELD_MAP.get(gender.lower(), genders.UNKNOWN)
+
+
+def map_course_enrollment_daily_rows(rows):
+    """Map course enrollment count rows into the existing API shape."""
+    return [
+        _copy_fields(row, ['course_id', 'date', 'count', 'created'])
+        for row in rows or []
+    ]
+
+
+def map_course_enrollment_education_rows(rows):
+    """Map enrollment education rows into the existing API shape."""
+    return [
+        _copy_fields(row, ['course_id', 'date', 'education_level', 'count', 'created'])
+        for row in rows or []
+    ]
+
+
+def map_course_enrollment_mode_rows(rows):
+    """Pivot enrollment mode rows into the existing API shape."""
+    rows = sorted(rows or [], key=lambda row: (_row_value(row, 'course_id'), _row_value(row, 'date')))
+    formatted_data = []
+
+    for key, group in groupby(rows, lambda row: (_row_value(row, 'course_id'), _row_value(row, 'date'))):
+        item = {
+            'course_id': key[0],
+            'date': key[1],
+            'created': None,
+        }
+        total = 0
+        cumulative_total = 0
+
+        for row in group:
+            mode = _row_value(row, 'mode')
+            count = int(_row_value(row, 'count'))
+            cumulative_count = int(_row_value(row, 'cumulative_count'))
+            created = _row_value(row, 'created')
+            item[mode] = item.get(mode, 0) + count
+            item['created'] = max(created, item['created']) if item['created'] else created
+            total += count
+            cumulative_total += cumulative_count
+
+        item[enrollment_modes.PROFESSIONAL] = item.get(enrollment_modes.PROFESSIONAL, 0) + item.pop(
+            enrollment_modes.PROFESSIONAL_NO_ID,
+            0,
+        )
+        item['count'] = total
+        item['cumulative_count'] = cumulative_total
+        formatted_data.append(item)
+
+    return formatted_data
+
+
+def map_course_enrollment_gender_rows(rows):
+    """Pivot enrollment gender rows into the existing API shape."""
+    rows = sorted(rows or [], key=lambda row: (_row_value(row, 'course_id'), _row_value(row, 'date')))
+    formatted_data = []
+
+    for key, group in groupby(rows, lambda row: (_row_value(row, 'course_id'), _row_value(row, 'date'))):
+        item = {
+            'course_id': key[0],
+            'date': key[1],
+            'created': None,
+            genders.MALE: 0,
+            genders.FEMALE: 0,
+            genders.OTHER: 0,
+            genders.UNKNOWN: 0,
+        }
+
+        for row in group:
+            gender = _gender_field(_row_value(row, 'gender'))
+            created = _row_value(row, 'created')
+            item[gender] += int(_row_value(row, 'count'))
+            item['created'] = max(created, item['created']) if item['created'] else created
+
+        formatted_data.append(item)
+
+    return formatted_data
+
+
+def map_course_enrollment_location_rows(rows):
+    """Map enrollment location rows into model instances used by the current serializer."""
+    items = [
+        models.CourseEnrollmentByCountry(
+            course_id=_row_value(row, 'course_id'),
+            date=_row_value(row, 'date'),
+            country_code=_row_value(row, 'country_code'),
+            count=int(_row_value(row, 'count')),
+            created=_row_value(row, 'created'),
+        )
+        for row in rows or []
+    ]
+    items = sorted(items, key=lambda item: (item.date, item.country.alpha2 or '', item.course_id))
+    returned_items = []
+
+    for key, group in groupby(items, lambda item: (item.date, item.country.alpha2, item.course_id)):
+        count = 0
+        created = None
+
+        for item in group:
+            created = max(created, item.created) if created else item.created
+            count += item.count
+
+        returned_items.append(models.CourseEnrollmentByCountry(
+            course_id=key[2],
+            date=key[0],
+            country_code=key[1],
+            count=count,
+            created=created,
+        ))
+
+    return returned_items

analytics_data_api/insights_snowflake/queries/enrollment.py

@@ -0,0 +1,117 @@
+"""Snowflake queries for course enrollment metrics."""
+
+import datetime
+
+from analytics_data_api.insights_snowflake.client import fetch_all, get_qualified_table_name
+
+COURSE_ENROLLMENT_DAILY_TABLE = 'COURSE_ENROLLMENT_DAILY'
+COURSE_ENROLLMENT_MODE_DAILY_TABLE = 'COURSE_ENROLLMENT_MODE_DAILY'
+COURSE_ENROLLMENT_EDUCATION_LEVEL_CURRENT_TABLE = 'COURSE_ENROLLMENT_EDUCATION_LEVEL_CURRENT'
+COURSE_ENROLLMENT_GENDER_DAILY_TABLE = 'COURSE_ENROLLMENT_GENDER_DAILY'
+COURSE_ENROLLMENT_LOCATION_CURRENT_TABLE = 'COURSE_ENROLLMENT_LOCATION_CURRENT'
+
+
+def _date_value(value):
+    """Return a date value for date-filtered Snowflake enrollment queries."""
+    if isinstance(value, datetime.datetime):
+        return value.date()
+    return value
+
+
+def _get_course_enrollment_rows(table, columns, order_by, course_id, start_date=None, end_date=None):
+    """Return Snowflake enrollment rows for one controlled table."""
+    table_name = get_qualified_table_name(table)
+    select_columns = ',\n    '.join(columns)
+    params = {
+        'course_id': course_id,
+    }
+
+    if start_date or end_date:
+        params.update({
+            'start_date': _date_value(start_date),
+            'end_date': _date_value(end_date),
+        })
+        sql = """
+SELECT
+    {select_columns}
+FROM {table_name}
+WHERE course_id = %(course_id)s
+  AND (%(start_date)s IS NULL OR "DATE" >= %(start_date)s)
+  AND (%(end_date)s IS NULL OR "DATE" < %(end_date)s)
+ORDER BY {order_by}
+""".format(select_columns=select_columns, table_name=table_name, order_by=order_by)
+    else:
+        sql = """
+SELECT
+    {select_columns}
+FROM {table_name}
+WHERE course_id = %(course_id)s
+  AND "DATE" = (
+      SELECT MAX("DATE")
+      FROM {table_name}
+      WHERE course_id = %(course_id)s
+  )
+ORDER BY {order_by}
+""".format(select_columns=select_columns, table_name=table_name, order_by=order_by)
+
+    return fetch_all(sql, params)
+
+
+def get_course_enrollment_daily_rows(course_id, start_date=None, end_date=None):
+    """Return Snowflake rows for course enrollment counts."""
+    return _get_course_enrollment_rows(
+        COURSE_ENROLLMENT_DAILY_TABLE,
+        ['course_id', '"DATE" AS date', '"COUNT" AS count', 'created'],
+        'course_id, date',
+        course_id,
+        start_date=start_date,
+        end_date=end_date,
+    )
+
+
+def get_course_enrollment_mode_rows(course_id, start_date=None, end_date=None):
+    """Return Snowflake rows for enrollment mode counts."""
+    return _get_course_enrollment_rows(
+        COURSE_ENROLLMENT_MODE_DAILY_TABLE,
+        ['course_id', '"DATE" AS date', 'mode', '"COUNT" AS count', 'cumulative_count', 'created'],
+        'course_id, date, mode',
+        course_id,
+        start_date=start_date,
+        end_date=end_date,
+    )
+
+
+def get_course_enrollment_education_rows(course_id, start_date=None, end_date=None):
+    """Return Snowflake rows for enrollment education counts."""
+    return _get_course_enrollment_rows(
+        COURSE_ENROLLMENT_EDUCATION_LEVEL_CURRENT_TABLE,
+        ['course_id', '"DATE" AS date', 'education_level', '"COUNT" AS count', 'created'],
+        'course_id, date, education_level',
+        course_id,
+        start_date=start_date,
+        end_date=end_date,
+    )
+
+
+def get_course_enrollment_gender_rows(course_id, start_date=None, end_date=None):
+    """Return Snowflake rows for enrollment gender counts."""
+    return _get_course_enrollment_rows(
+        COURSE_ENROLLMENT_GENDER_DAILY_TABLE,
+        ['course_id', '"DATE" AS date', 'gender', '"COUNT" AS count', 'created'],
+        'course_id, date, gender',
+        course_id,
+        start_date=start_date,
+        end_date=end_date,
+    )
+
+
+def get_course_enrollment_location_rows(course_id, start_date=None, end_date=None):
+    """Return Snowflake rows for enrollment location counts."""
+    return _get_course_enrollment_rows(
+        COURSE_ENROLLMENT_LOCATION_CURRENT_TABLE,
+        ['course_id', '"DATE" AS date', 'country_code', '"COUNT" AS count', 'created'],
+        'course_id, date, country_code',
+        course_id,
+        start_date=start_date,
+        end_date=end_date,
+    )

analytics_data_api/insights_snowflake/service.py

@@ -1,10 +1,54 @@
 """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.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.queries.activity import get_course_activity_weekly_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,
+)
 
 
 def get_course_activity_weekly(course_id, start_date=None, end_date=None):
     """Return course activity in the existing API response shape."""
     rows = get_course_activity_weekly_rows(course_id, start_date=start_date, end_date=end_date)
     return map_course_activity_weekly_rows(rows)
+
+
+def get_course_enrollment(course_id, start_date=None, end_date=None):
+    """Return course enrollment counts in the existing API response shape."""
+    rows = get_course_enrollment_daily_rows(course_id, start_date=start_date, end_date=end_date)
+    return map_course_enrollment_daily_rows(rows)
+
+
+def get_course_enrollment_mode(course_id, start_date=None, end_date=None):
+    """Return course enrollment mode counts in the existing API response shape."""
+    rows = get_course_enrollment_mode_rows(course_id, start_date=start_date, end_date=end_date)
+    return map_course_enrollment_mode_rows(rows)
+
+
+def get_course_enrollment_education(course_id, start_date=None, end_date=None):
+    """Return course enrollment education counts in the existing API response shape."""
+    rows = get_course_enrollment_education_rows(course_id, start_date=start_date, end_date=end_date)
+    return map_course_enrollment_education_rows(rows)
+
+
+def get_course_enrollment_gender(course_id, start_date=None, end_date=None):
+    """Return course enrollment gender counts in the existing API response shape."""
+    rows = get_course_enrollment_gender_rows(course_id, start_date=start_date, end_date=end_date)
+    return map_course_enrollment_gender_rows(rows)
+
+
+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)