pagination.py 2.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051
  1. from rest_framework import status
  2. from rest_framework.pagination import CursorPagination
  3. from rest_framework.response import Response
  4. from rest_framework.utils.urls import replace_query_param
  5. class LinkHeaderCursorPagination(CursorPagination):
  6. """
  7. Inform the user of pagination links via response headers, similar to what's
  8. described in https://developer.github.com/v3/guides/traversing-with-pagination/
  9. Inspired by the django-rest-framework-link-header-pagination package.
  10. """
  11. template = None
  12. @staticmethod
  13. def construct_headers(pagination_map):
  14. links = [
  15. f'<{url}>; rel="{label}"'
  16. for label, url in pagination_map.items()
  17. if url is not None
  18. ]
  19. return {"Link": ", ".join(links)} if links else {}
  20. def get_paginated_response(self, data):
  21. pagination_required = self.has_next or self.has_previous
  22. if not pagination_required:
  23. return Response(data)
  24. url = self.request.build_absolute_uri()
  25. pagination_map = {
  26. "first": replace_query_param(url, self.cursor_query_param, "")
  27. }
  28. if self.cursor_query_param not in self.request.query_params:
  29. count = self.queryset.count()
  30. data = {
  31. "detail": f"Pagination required. You can query up to {self.page_size} items at a time ({count} total). "
  32. "Please use the `first` page link (see Link header).",
  33. }
  34. headers = self.construct_headers(pagination_map)
  35. return Response(data, headers=headers, status=status.HTTP_400_BAD_REQUEST)
  36. pagination_map.update(prev=self.get_previous_link(), next=self.get_next_link())
  37. headers = self.construct_headers(pagination_map)
  38. return Response(data, headers=headers)
  39. def paginate_queryset(self, queryset, request, view=None):
  40. self.request = request
  41. self.queryset = queryset
  42. return super().paginate_queryset(queryset, request, view)