123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566 |
- 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)
|