Add CSV export support to partner catalog list endpoints via DRF rend…

partner_catalog/api/v1/views.py

@@ -30,6 +30,7 @@
     PartnerSerializer,
 )
 from partner_catalog.api.v1.tasks import bulk_remove_invitations, bulk_upload_invitations
+from partner_catalog.helpers.mixins import CSVExportMixin
 from partner_catalog.models import (
     CatalogCourse,
     CatalogCourseEnrollment,
@@ -88,7 +89,7 @@ def get_queryset(self):
         return qs
 
 
-class PartnerCatalogViewSet(viewsets.ModelViewSet):
+class PartnerCatalogViewSet(CSVExportMixin, viewsets.ModelViewSet):
     """
     ViewSet for Corporate Partner Catalog data.
     Provides access to corporate partner catalog information.
@@ -98,6 +99,26 @@ class PartnerCatalogViewSet(viewsets.ModelViewSet):
     queryset = PartnerCatalog.objects.all()
     serializer_class = PartnerCatalogSerializer
     permission_classes = [IsPartnerCatalogManager]
+
+    csv_filename = "catalogs_report.csv"
+    csv_fields = [
+        "name", "slug", "status", "courses", "enrollments",
+        "total_learners", "active_learners", "certified", "completion_rate",
+        "available_start_date", "available_end_date",
+    ]
+    csv_labels = {
+        "name": "Catalog Name",
+        "slug": "Slug",
+        "status": "Status",
+        "courses": "Courses",
+        "enrollments": "Enrollments",
+        "total_learners": "Total Learners",
+        "active_learners": "Active Learners",
+        "certified": "Certified",
+        "completion_rate": "Completion Rate",
+        "available_start_date": "Available Start Date",
+        "available_end_date": "Available End Date",
+    }
     filter_backends = [
         DjangoFilterBackend,
         filters.SearchFilter,
@@ -218,7 +239,7 @@ def available_courses(self, request, **kwargs):
         return Response(response_data)
 
 
-class CatalogLearnerViewset(InjectNestedFKMixin, viewsets.ReadOnlyModelViewSet):
+class CatalogLearnerViewset(CSVExportMixin, InjectNestedFKMixin, viewsets.ReadOnlyModelViewSet):
     """
     ViewSet for Corporate Partner Catalog Learner data.
     Provides access to corporate partner catalog learner information.
@@ -227,6 +248,23 @@ class CatalogLearnerViewset(InjectNestedFKMixin, viewsets.ReadOnlyModelViewSet):
     queryset = CatalogLearner.objects.select_related("catalog", "user", "current_invitation")
     serializer_class = CatalogLearnerSerializer
     permission_classes = [IsPartnerCatalogManager]
+
+    csv_filename = "learners_report.csv"
+    csv_fields = [
+        "user.full_name", "user.email", "active", "invite_sent_at",
+        "accepted_at", "user.last_login", "enrollments", "certified", "removed_at",
+    ]
+    csv_labels = {
+        "user.full_name": "Full Name",
+        "user.email": "Email",
+        "active": "Active",
+        "invite_sent_at": "Invite Sent At",
+        "accepted_at": "Accepted At",
+        "user.last_login": "Last Login",
+        "enrollments": "Enrollments",
+        "certified": "Certified",
+        "removed_at": "Removed At",
+    }
     filter_backends = [
         DjangoFilterBackend,
         filters.SearchFilter,
@@ -275,6 +313,7 @@ def get_queryset(self):
 
 
 class CatalogCourseViewSet(
+    CSVExportMixin,
     InjectNestedFKMixin,
     mixins.RetrieveModelMixin,
     mixins.UpdateModelMixin,
@@ -291,6 +330,24 @@ class CatalogCourseViewSet(
     )
     serializer_class = CatalogCourseSerializer
     permission_classes = [IsPartnerCatalogManager]
+
+    csv_filename = "courses_report.csv"
+    csv_fields = [
+        "course_run.display_name", "position", "course_run.start", "course_run.end",
+        "course_run.enrollment_start", "course_run.enrollment_end",
+        "enrollments", "certified", "completion_rate",
+    ]
+    csv_labels = {
+        "course_run.display_name": "Course Name",
+        "position": "Position",
+        "course_run.start": "Start Date",
+        "course_run.end": "End Date",
+        "course_run.enrollment_start": "Enrollment Start",
+        "course_run.enrollment_end": "Enrollment End",
+        "enrollments": "Enrollments",
+        "certified": "Certified",
+        "completion_rate": "Completion Rate",
+    }
     filter_backends = [
         DjangoFilterBackend,
         filters.SearchFilter,
@@ -502,7 +559,7 @@ def bulk_invite_status(self, request, task_id=None, **kwargs):
 
 
 class CatalogCourseEnrollmentViewSet(
-    viewsets.ReadOnlyModelViewSet, InjectNestedFKMixin
+    CSVExportMixin, viewsets.ReadOnlyModelViewSet, InjectNestedFKMixin
 ):
     """
     ViewSet for Catalog Course Enrollments.
@@ -512,6 +569,20 @@ class CatalogCourseEnrollmentViewSet(
     queryset = CatalogCourseEnrollment.objects.select_related("user", "catalog_course")
     serializer_class = CatalogCourseEnrollmentSerializer
     permission_classes = [IsPartnerCatalogManager]
+
+    csv_filename = "course_enrollments_report.csv"
+    csv_fields = [
+        "user.full_name", "user.email", "active", "user.last_login",
+        "progress", "has_certificate",
+    ]
+    csv_labels = {
+        "user.full_name": "Full Name",
+        "user.email": "Email",
+        "active": "Active",
+        "user.last_login": "Last Login",
+        "progress": "Progress",
+        "has_certificate": "Has Certificate",
+    }
     filter_backends = [
         DjangoFilterBackend,
         filters.SearchFilter,
@@ -533,7 +604,7 @@ def get_queryset(self):
         return qs.filter(catalog_course_id=course_pk) if course_pk else qs
 
 
-class CatalogEnrollmentsViewSet(viewsets.ReadOnlyModelViewSet):
+class CatalogEnrollmentsViewSet(CSVExportMixin, viewsets.ReadOnlyModelViewSet):
     """
     ViewSet for retrieving enrollments across all courses in a specific corporate partner catalog.
 
@@ -544,6 +615,23 @@ class CatalogEnrollmentsViewSet(viewsets.ReadOnlyModelViewSet):
 
     serializer_class = CatalogCourseEnrollmentSerializer
     permission_classes = [IsPartnerCatalogManager]
+
+    csv_filename = "enrollments_report.csv"
+    csv_fields = [
+        "user.full_name", "user.email", "active", "user.last_login",
+        "course_overview.display_name", "course_overview.id",
+        "progress", "has_certificate",
+    ]
+    csv_labels = {
+        "user.full_name": "Full Name",
+        "user.email": "Email",
+        "active": "Active",
+        "user.last_login": "Last Login",
+        "course_overview.display_name": "Course Name",
+        "course_overview.id": "Course ID",
+        "progress": "Progress",
+        "has_certificate": "Has Certificate",
+    }
     filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
 
     filterset_fields = ["active", "user"]

partner_catalog/helpers/mixins.py

@@ -0,0 +1,61 @@
+"""View mixins for CSV export support."""
+
+from rest_framework.settings import api_settings
+
+from partner_catalog.helpers.renderers import CSVReportRenderer
+
+
+class CSVExportMixin:
+    """
+    Mixin that adds CSV export capability to list endpoints.
+
+    Usage: add to a ViewSet's bases and define ``csv_filename``,
+    ``csv_fields`` (ordered list of dot-notation keys) and
+    ``csv_labels`` (mapping of keys to human-readable headers).
+
+    Clients request CSV via ``?format=csv`` or ``Accept: text/csv``.
+    Pagination is automatically skipped for CSV responses so the full
+    dataset is exported.
+
+    Note: Open edX sets ``URL_FORMAT_OVERRIDE = None`` which disables
+    DRF's built-in ``?format=`` query parameter support. This mixin
+    works around that by explicitly checking the query parameter in
+    ``get_format_suffix()``.
+    """
+
+    csv_filename = "report.csv"
+    csv_fields = None
+    csv_labels = {}
+
+    def get_renderers(self):
+        renderers = super().get_renderers()
+        if not any(isinstance(r, CSVReportRenderer) for r in renderers):
+            renderers.append(CSVReportRenderer())
+        return renderers
+
+    def get_format_suffix(self, **kwargs):
+        """
+        Re-enable ``?format=`` query parameter for CSV requests.
+
+        Open edX sets ``URL_FORMAT_OVERRIDE = None`` globally, which
+        prevents DRF from reading the ``?format=`` query parameter.
+        We restore this behaviour only for our CSV-enabled views.
+        """
+        fmt = super().get_format_suffix(**kwargs)
+        if fmt is None:
+            key = api_settings.URL_FORMAT_OVERRIDE or "format"
+            fmt = self.request.query_params.get(key)
+        return fmt
+
+    def paginate_queryset(self, queryset):
+        """Skip pagination when the client requested CSV format."""
+        if getattr(self.request, "accepted_renderer", None) and self.request.accepted_renderer.format == "csv":
+            return None
+        return super().paginate_queryset(queryset)
+
+    def finalize_response(self, request, response, *args, **kwargs):
+        """Attach Content-Disposition header for CSV downloads."""
+        response = super().finalize_response(request, response, *args, **kwargs)
+        if getattr(response, "accepted_renderer", None) and response.accepted_renderer.format == "csv":
+            response["Content-Disposition"] = f'attachment; filename="{self.csv_filename}"'
+        return response

partner_catalog/helpers/renderers.py

@@ -0,0 +1,62 @@
+"""Custom DRF renderers for CSV export."""
+
+import csv
+import io
+
+from rest_framework.renderers import BaseRenderer
+
+
+class CSVReportRenderer(BaseRenderer):
+    """
+    Renderer that serializes data to CSV format.
+
+    Reads ``csv_fields`` and ``csv_labels`` from the view to control
+    which columns are included and how they are named in the header row.
+    Nested dictionaries are flattened using dot notation
+    (e.g. ``{"user": {"email": "a@b.c"}}`` becomes ``"user.email"``).
+    """
+
+    media_type = "text/csv"
+    format = "csv"
+    charset = "utf-8"
+
+    def flatten_item(self, item, parent_key=""):
+        """Flatten a nested dict using dot-separated keys."""
+        flat = {}
+        for key, value in item.items():
+            full_key = f"{parent_key}.{key}" if parent_key else key
+            if isinstance(value, dict):
+                flat.update(self.flatten_item(value, full_key))
+            else:
+                flat[full_key] = value if value is not None else ""
+        return flat
+
+    def render(self, data, accepted_media_type=None, renderer_context=None):
+        if data is None:
+            return ""
+
+        # Handle paginated responses.
+        if isinstance(data, dict) and "results" in data:
+            data = data["results"]
+
+        if not isinstance(data, list):
+            data = [data]
+
+        flat_data = [self.flatten_item(item) for item in data]
+
+        if not flat_data:
+            return ""
+
+        view = renderer_context.get("view") if renderer_context else None
+        csv_fields = getattr(view, "csv_fields", None) or list(flat_data[0].keys())
+        csv_labels = getattr(view, "csv_labels", {})
+
+        headers = [csv_labels.get(field, field) for field in csv_fields]
+
+        output = io.StringIO()
+        writer = csv.writer(output)
+        writer.writerow(headers)
+        for item in flat_data:
+            writer.writerow([item.get(field, "") for field in csv_fields])
+
+        return output.getvalue()