feat: add helper functions for generating CSV reports

partner_catalog/api/v1/schemas.py

@@ -292,3 +292,55 @@ def remove_courses_schema(func):
         },
         tags=["Catalogs"]
     )(func)
+
+
+def report_schema(func):
+    """Decorator to document CSV report endpoints.
+
+    Expected behavior:
+      * 200: Returns a CSV file attachment with the selected report fields as columns.
+      * 400: Returns JSON error when there is no data available to export.
+
+    Notes:
+      * The CSV filename is always `report.csv` (may evolve later to be dynamic).
+      * The set & order of columns comes from the view's `report_fields` attribute.
+    """
+    return extend_schema(
+        summary="Download CSV report",
+        description=dedent(
+            """
+            Generate and download a CSV report for the current resource scope.
+
+            The columns included in the CSV are defined by the ViewSet's `report_fields` attribute.
+            Filtering, search, or other query parameters applied to the list endpoint
+            will also affect the dataset exported here.
+
+            Responses:
+            * 200: CSV file (text/csv) attachment named `report.csv`.
+            * 400: JSON error when there is no data to include in the report.
+            """
+        ),
+        responses={
+            200: OpenApiResponse(
+                response=OpenApiTypes.BINARY,
+                description="CSV report generated successfully (text/csv attachment)",
+                examples=[
+                    OpenApiExample(
+                        "CSV Preview (first lines)",
+                        value="id,name\n1,Sample Name\n2,Another\n",
+                    )
+                ],
+            ),
+            400: OpenApiResponse(
+                response=OpenApiTypes.OBJECT,
+                description="No data available to generate report",
+                examples=[
+                    OpenApiExample(
+                        "No Data",
+                        value={"detail": "No data to report."},
+                    )
+                ],
+            ),
+        },
+        tags=["Reports"],
+    )(func)

partner_catalog/api/v1/views.py

@@ -16,6 +16,7 @@
     bulk_status_invitations_schema,
     bulk_upload_invitations_schema,
     remove_courses_schema,
+    report_schema,
 )
 from partner_catalog.api.v1.serializers import (
     BasicCourseOverviewSerializer,
@@ -29,6 +30,7 @@
     PartnerSerializer,
 )
 from partner_catalog.api.v1.tasks import bulk_remove_invitations, bulk_upload_invitations
+from partner_catalog.helpers.reports import generate_csv_report
 from partner_catalog.models import (
     CatalogCourse,
     CatalogCourseEnrollment,
@@ -231,6 +233,29 @@ def enrollments(self, request, pk=None):
         serializer = CatalogCourseEnrollmentSerializer(enrollments, many=True)
         return Response(serializer.data)
 
+    @report_schema
+    @action(detail=True, methods=["get"], url_path="enrollments/report")
+    def enrollments_report(self, request, pk=None):
+        """Download CSV report of all course enrollments for this catalog."""
+        enrollments = CatalogCourseEnrollment.objects.filter(
+            catalog_course__catalog_id=pk
+        ).select_related("user", "catalog_course", "catalog_course__course_overview")
+
+        serializer = CatalogCourseEnrollmentSerializer(enrollments, many=True)
+        report_fields = [
+            "user__full_name",
+            "user__email",
+            "active",
+            "user__last_login",
+            "course_overview__display_name",
+            "course_overview__id",
+            "progress",
+            "has_certificate",
+        ]
+        return generate_csv_report(
+            serializer.data, report_fields, filename="enrollments_report.csv"
+        )
+
 
 class CatalogLearnerViewset(InjectNestedFKMixin, viewsets.ReadOnlyModelViewSet):
     """
@@ -279,6 +304,27 @@ def get_queryset(self):
         qs = annotate_learner_certified_count(qs)
         return qs
 
+    @report_schema
+    @action(detail=False, methods=["get"], url_path="report")
+    def report(self, request, *args, **kwargs):
+        """Download CSV report of all learners in this catalog."""
+        queryset = self.filter_queryset(self.get_queryset())
+        serializer = self.get_serializer(queryset, many=True)
+        report_fields = [
+            "user__full_name",
+            "user__email",
+            "active",
+            "invite_sent_at",
+            "accepted_at",
+            "user__last_login",
+            "enrollments",
+            "certified",
+            "removed_at",
+        ]
+        return generate_csv_report(
+            serializer.data, report_fields, filename="learners_report.csv"
+        )
+
 
 class CatalogCourseViewSet(
     InjectNestedFKMixin,
@@ -328,6 +374,27 @@ def get_queryset(self):
         qs = annotate_course_certified_count(qs)
         return qs
 
+    @report_schema
+    @action(detail=False, methods=["get"], url_path="report")
+    def report(self, request, *args, **kwargs):
+        """Download CSV report of all courses in this catalog."""
+        queryset = self.filter_queryset(self.get_queryset())
+        serializer = self.get_serializer(queryset, many=True)
+        report_fields = [
+            "course_run__display_name",
+            "position",
+            "course_run__start",
+            "course_run__end",
+            "course_run__enrollment_start",
+            "course_run__enrollment_end",
+            "enrollments",
+            "certified",
+            "completion_rate",
+        ]
+        return generate_csv_report(
+            serializer.data, report_fields, filename="courses_report.csv"
+        )
+
 
 class CatalogLearnerInvitationViewSet(
     mixins.ListModelMixin,

partner_catalog/helpers/reports.py

@@ -0,0 +1,59 @@
+"""Helper functions for generating CSV reports."""
+import csv
+
+from django.http import HttpResponse
+from rest_framework.response import Response
+
+
+def get_nested_value(item, field):
+    """
+    Get a value from a nested dictionary using '__' notation.
+
+    Args:
+        item: The dictionary to extract the value from.
+        field: The field name, which may contain '__' for nested access.
+
+    Returns:
+        The value at the nested path, or an empty string if not found.
+    """
+    if "__" not in field:
+        return item.get(field, "")
+
+    keys = field.split("__")
+    value = item
+    for key in keys:
+        if isinstance(value, dict):
+            value = value.get(key, "")
+        else:
+            return ""
+    return value
+
+
+def generate_csv_report(data, fields, filename="report.csv"):
+    """
+    Generate a CSV report from serialized data.
+
+    Args:
+        data: List of dictionaries (serialized data).
+        fields: List of field names to include in the CSV.
+        filename: Name of the CSV file to download.
+
+    Returns:
+        HttpResponse with CSV content or Response with error if no data.
+    """
+    if not data:
+        return Response({"detail": "No data to report."}, status=400)
+
+    filtered_data = [
+        {field: get_nested_value(item, field) for field in fields}
+        for item in data
+    ]
+
+    response = HttpResponse(content_type="text/csv")
+    response["Content-Disposition"] = f'attachment; filename="{filename}"'
+
+    writer = csv.DictWriter(response, fieldnames=fields)
+    writer.writeheader()
+    writer.writerows(filtered_data)
+
+    return response