feat: add aggregated data collector API (#314)

* feat: add API to collect data

* feat: update version and changelog

* fix: process results

* chore: change IsGitHubAction class name

* chore: block non-SELECT queries

* feat: add setting AGGREGATED_DATA_COLLECTOR_API_ENABLED to disable data collector endpoint

* refactor: execute all predefined queries and remove query selection

* chore: rename EOX_CORE_DATA_COLLECTOR_ENABLED setting to AGGREGATED_DATA_COLLECTOR_API_ENABLED

* chore: COUNTDOWN and MAX_RETRIES constants

* refactor: merge serialize_data and process_query_results into post_process_query_results

* refactor: remove validate Only SELECT queries are allowed

* chore: rename EOX_CORE_SAVE_DATA_API_CLIENT_ID and EOX_CORE_SAVE_DATA_API_CLIENT_SECRET

* chore: Updated DataCollectorView to retrieve destination_url and token_generation_url from settings.

* refactor: move DatacollectorPermission to a dedicated permissions.py file

* feat: add aggregated collector test

---------

Co-authored-by: Luis Felipe Castano <[email protected]>

Author: magajh

CHANGELOG.md

@@ -11,6 +11,12 @@ Please do not update the unreleased notes.
 
 <!-- Content should be placed here -->
 
+## [v11.2.0](https://github.com/eduNEXT/eox-core/compare/v11.1.0...v11.2.0) - (2025-01-20)
+
+### Added
+
+- New API endpoint to support data collection and report generation (if the feature is enabled).
+
 ## [v11.1.0](https://github.com/eduNEXT/eox-core/compare/v11.0.0...v11.1.0) - (2024-11-21)
 
 ### Changed

eox_core/__init__.py

@@ -1,4 +1,4 @@
 """
 Init for main eox-core app
 """
-__version__ = '11.1.0'
+__version__ = '11.2.0'

eox_core/api/data/aggregated_collector/__init__.py

@@ -0,0 +1,123 @@
+"""
+This module contains SQL queries to retrieve platform-related metrics.
+
+These queries are used to extract information about user activity, course
+engagement, and certificate issuance. The data retrieval is conditioned on
+the feature being enabled and the necessary authentication credentials
+being set.
+"""
+
+# This query counts the total number of users in the platform.
+TOTAL_USERS_QUERY = """
+SELECT
+  COUNT(*)
+FROM
+  auth_user as au;
+"""
+
+# This query counts the number of unique active users in the last month.
+# A user is considered active if they have modified a student module within the last month.
+ACTIVE_USERS_LAST_MONTH_QUERY = """
+SELECT
+  YEAR(cs.modified) AS 'Year',
+  MONTH(cs.modified) AS 'Month',
+    COUNT(DISTINCT cs.student_id) AS 'Active Users'
+FROM
+    courseware_studentmodule AS cs
+WHERE
+    cs.modified >= DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 MONTH), '%Y-%m-01')
+    AND cs.modified < DATE_FORMAT(CURDATE(), '%Y-%m-01')
+GROUP BY
+    YEAR(cs.modified), MONTH(cs.modified);
+"""
+
+# This query counts the number of unique active users in the last 7 days.
+# A user is considered active if they have modified a student module within the last 7 days.
+ACTIVE_USERS_LAST_7_DAYS_QUERY = """
+SELECT
+  COUNT(DISTINCT cs.student_id)
+FROM
+  courseware_studentmodule AS cs
+WHERE
+  cs.modified >= DATE_SUB(CURDATE(), INTERVAL 7 DAY);
+"""
+
+# This query counts the total number of courses created on the platform.
+TOTAL_COURSES_CREATED_QUERY = """
+SELECT
+  COUNT(*)
+FROM
+  course_overviews_courseoverview as coc;
+"""
+
+# This query counts the number of CourseOverviews objects that started before
+# now and have not yet ended
+ACTIVE_COURSES_COUNT_QUERY = """
+SELECT
+  COUNT(*)
+FROM
+  course_overviews_courseoverview as coc
+WHERE
+  coc.start <= NOW()
+  AND (
+    coc.end >= NOW()
+    OR coc.end IS NULL
+  );
+"""
+
+# This query counts the number of courses that have at least one issued certificate.
+COURSES_WITH_ACTIVE_CERTIFICATES_QUERY = """
+SELECT
+  COUNT(DISTINCT coc.id)
+FROM
+  course_overviews_courseoverview AS coc
+JOIN
+  certificates_generatedcertificate AS cg
+ON
+  coc.id = cg.course_id;
+"""
+
+# This query counts the number of new enrollments in the last month.
+ENROLLMENTS_LAST_MONTH_QUERY = """
+SELECT
+  YEAR(sc.created) AS 'Year',
+  MONTH(sc.created) AS 'Month',
+  COUNT(DISTINCT sc.id) AS 'Enrollments'
+FROM
+  student_courseenrollment AS sc
+WHERE
+  sc.created >= DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 MONTH), '%Y-%m-01')
+  AND sc.created < DATE_FORMAT(CURDATE(), '%Y-%m-01')
+GROUP BY
+  YEAR(sc.created), MONTH(sc.created);
+"""
+
+# This query counts the number of new enrollments in the last 7 days.
+ENROLLMENTS_LAST_7_DAYS_QUERY = """
+SELECT
+  COUNT(DISTINCT sc.id)
+FROM
+  student_courseenrollment AS sc
+WHERE
+  sc.created >= DATE_SUB(CURDATE(), INTERVAL 7 DAY);
+"""
+
+# This query counts the total number of certificates issued on the platform.
+CERTIFICATES_ISSUED_QUERY = """
+SELECT
+  COUNT(*)
+FROM
+  certificates_generatedcertificate as cg;
+"""
+
+PREDEFINED_QUERIES = {
+    "Total Users": TOTAL_USERS_QUERY,
+    "Active Users Last Month": ACTIVE_USERS_LAST_MONTH_QUERY,
+    "Active users in the last 7 days": ACTIVE_USERS_LAST_7_DAYS_QUERY,
+    "Total Courses Created": TOTAL_COURSES_CREATED_QUERY,
+    "Active Courses Count": ACTIVE_COURSES_COUNT_QUERY,
+    "Courses With Active Certificates": COURSES_WITH_ACTIVE_CERTIFICATES_QUERY,
+    "Enrollments Last Month": ENROLLMENTS_LAST_MONTH_QUERY,
+    "Enrollments Last 7 Days": ENROLLMENTS_LAST_7_DAYS_QUERY,
+    "Certificates Issued": CERTIFICATES_ISSUED_QUERY,
+}

eox_core/api/data/aggregated_collector/queries.py

@@ -0,0 +1,52 @@
+"""
+Async task for generating reports by executing database queries
+and posting the results to the Shipyard API.
+"""
+
+import logging
+
+from celery import shared_task
+from django.db.utils import DatabaseError, OperationalError
+
+from eox_core.api.data.aggregated_collector.queries import PREDEFINED_QUERIES
+from eox_core.api.data.aggregated_collector.utils import execute_query, post_data_to_api, post_process_query_results
+
+logger = logging.getLogger(__name__)
+
+COUNTDOWN = 60
+MAX_RETRIES = 3
+
+
+@shared_task(bind=True)
+def generate_report(self, destination_url: str, token_generation_url: str, current_host: str):
+    """
+    Async task to generate a report:
+    1. Executes all predefined queries.
+    2. Sends the results to the Shipyard API.
+
+    Args:
+        self (Task): The Celery task instance.
+        destination_url (str): URL to send the results.
+
+    Raises:
+        Retry: If an error occurs, the task retries up to 3 times with a 60-second delay.
+    """
+    try:
+        report_data = {}
+        for query_name, query_sql in PREDEFINED_QUERIES.items():
+            logger.info("Executing query: %s", query_name)
+            try:
+                result = execute_query(query_sql)
+
+                processed_result = post_process_query_results(result)
+                report_data[query_name] = processed_result
+            except (DatabaseError, OperationalError) as e:
+                logger.error("Failed to execute query '%s': %s", query_name, e)
+                continue
+
+        post_data_to_api(destination_url, report_data, token_generation_url, current_host)
+
+        logger.info("Report generation task completed successfully.")
+    except Exception as e:
+        logger.error("An error occurred in the report generation task: '%s'. Retrying", e)
+        raise self.retry(exc=e, countdown=COUNTDOWN, max_retries=MAX_RETRIES)

eox_core/api/data/aggregated_collector/tasks.py

@@ -0,0 +1,14 @@
+"""
+URL configuration for the Microsite API.
+
+This module defines the URL patterns for the Microsite API,
+including versioned endpoints for aggregated_collector.
+"""
+from django.urls import include, re_path
+
+app_name = 'eox_core'  # pylint: disable=invalid-name
+
+
+urlpatterns = [  # pylint: disable=invalid-name
+    re_path(r'^v1/', include('eox_core.api.data.aggregated_collector.v1.urls', namespace='eox-data-api-collector-v1')),
+]

eox_core/api/data/aggregated_collector/urls.py

@@ -0,0 +1,98 @@
+"""
+Utility functions for report generation, including query execution and data posting.
+"""
+
+import logging
+from datetime import datetime
+
+import requests
+from django.conf import settings
+from django.db import connection
+
+from eox_core.utils import get_access_token
+
+logger = logging.getLogger(__name__)
+
+
+def execute_query(sql_query: str):
+    """
+    Execute a raw SQL query and return the results in a structured format.
+
+    Args:
+        sql_query (str): The raw SQL query to execute.
+
+    Returns:
+        list or dict: Structured query results.
+
+    Example:
+        >>> execute_query("SELECT id, username FROM auth_user WHERE is_active = 1;")
+        [
+            {"id": 1, "username": "john_doe"},
+            {"id": 2, "username": "jane_doe"}
+        ]
+    """
+    with connection.cursor() as cursor:
+        cursor.execute(sql_query)
+        rows = cursor.fetchall()
+        # If the query returns more than one column, return rows as is.
+        if cursor.description:
+            columns = [col[0] for col in cursor.description]
+            if len(columns) == 1:
+                return [row[0] for row in rows]  # Return single-column results as a list
+            return [dict(zip(columns, row)) for row in rows]  # Multi-column results as a list of dicts
+        return rows
+
+
+def post_process_query_results(data: any):
+    """
+    Cleans and processes query results by:
+    - Serializing datetime objects into strings.
+    - Extracting scalar values from single-item lists.
+    - Returning structured data for further use.
+
+    Args:
+        data (dict, list, datetime, or scalar): The query result data.
+
+    Returns:
+        dict, list, or scalar: The processed query result.
+    """
+    if isinstance(data, dict):
+        return {key: post_process_query_results(value) for key, value in data.items()}
+    if isinstance(data, list):
+        # If it's a list with one item, return just the item
+        if len(data) == 1:
+            return post_process_query_results(data[0])
+        return [post_process_query_results(item) for item in data]
+    if isinstance(data, datetime):
+        return data.isoformat()
+    return data
+
+
+def post_data_to_api(api_url: str, report_data: dict, token_generation_url: str, current_host: str):
+    """
+    Sends the generated report data to the Shipyard API.
+
+    Args:
+        report_data (dict): The data to be sent to the Shipyard API.
+
+    Raises:
+        Exception: If the API request fails.
+    """
+    token = get_access_token(
+        token_generation_url,
+        settings.EOX_CORE_AGGREGATED_COLLECTOR_TARGET_CLIENT_ID,
+        settings.EOX_CORE_AGGREGATED_COLLECTOR_TARGET_CLIENT_SECRET,
+    )
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "Content-Type": "application/json",
+    }
+    payload = {"instance_domain": current_host, "data": report_data}
+
+    try:
+        response = requests.post(api_url, json=payload, headers=headers, timeout=10)
+        response.raise_for_status()
+    except requests.Timeout as exc:
+        raise requests.Timeout("The request to API timed out.") from exc
+    except requests.RequestException as e:
+        raise requests.RequestException(f"Failed to post data to API: {e}")

eox_core/api/data/aggregated_collector/utils.py

@@ -0,0 +1,34 @@
+"""
+Custom permission classes for the Aggregated Collector API.
+"""
+
+import logging
+
+from django.conf import settings
+from rest_framework.authentication import get_authorization_header
+from rest_framework.permissions import BasePermission
+
+logger = logging.getLogger(__name__)
+
+
+class AggregatedCollectorPermission(BasePermission):
+    """
+    Permission class to allow access only if:
+    - The AGGREGATED_DATA_COLLECTOR_API_ENABLED setting is True.
+    - The request contains a valid GitHub Action token.
+    """
+
+    def has_permission(self, request, view):
+        """"
+        Determines if the request has permission to access the Aggregated Collector API.
+        """
+        # Check if the API is enabled
+        if not getattr(settings, "AGGREGATED_DATA_COLLECTOR_API_ENABLED", False):
+            return False
+
+        # Check if the request contains a valid token
+        auth_header = get_authorization_header(request).decode('utf-8')
+        auth_token = settings.EOX_CORE_AGGREGATED_COLLECTOR_AUTH_TOKEN
+        if auth_header and auth_header == f"Bearer {auth_token}":
+            return True
+        return False