import logging import urllib.parse as urlparse from urllib.parse import parse_qs from rest_framework.exceptions import ValidationError from rest_framework.pagination import CursorPagination from rest_framework.response import Response logger = logging.getLogger(__name__) class LinkHeaderPagination(CursorPagination): """Inform the user of pagination links via response headers, similar to what's described in https://developer.github.com/guides/traversing-with-pagination/. """ page_size_query_param = "limit" max_hits = 1000 def paginate_queryset(self, queryset, request, view=None): try: page = super().paginate_queryset(queryset, request, view) if self.has_next: self.count = self.get_count(queryset) else: self.count = len(page) return page except ValueError as err: # https://gitlab.com/glitchtip/glitchtip-backend/-/issues/136 logging.warning("Pagination received invalid cursor", exc_info=True) raise ValidationError("Invalid page cursor") from err def get_count(self, queryset): """Count with max limit, to prevent slowdown""" # Remove order by, it can slow down counts but has no effect return queryset.order_by()[: self.max_hits].count() def get_paginated_response(self, data): next_url = self.get_next_link() previous_url = self.get_previous_link() links = [] for url, label in ( (previous_url, "previous"), (next_url, "next"), ): if url is not None: parsed = urlparse.urlparse(url) cursor = parse_qs(parsed.query).get(self.cursor_query_param, [""])[0] links.append( '<{}>; rel="{}"; results="true"; cursor="{}"'.format( url, label, cursor ) ) else: links.append( '<{}>; rel="{}"; results="false"'.format(self.base_url, label) ) headers = {"Link": ", ".join(links)} if links else {} headers["X-Max-Hits"] = self.max_hits headers["X-Hits"] = self.count return Response(data, headers=headers)