diff --git a/partner_catalog/api/v1/views.py b/partner_catalog/api/v1/views.py index 402d6c9..6ab0563 100644 --- a/partner_catalog/api/v1/views.py +++ b/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"] diff --git a/partner_catalog/helpers/mixins.py b/partner_catalog/helpers/mixins.py new file mode 100644 index 0000000..86f0c27 --- /dev/null +++ b/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 diff --git a/partner_catalog/helpers/renderers.py b/partner_catalog/helpers/renderers.py new file mode 100644 index 0000000..9bd54d4 --- /dev/null +++ b/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()