From 2a806ef3eb78efcdf4f06cc58e781605d08f5c93 Mon Sep 17 00:00:00 2001 From: sergiusnick Date: Thu, 24 Oct 2024 02:36:40 +0500 Subject: [PATCH] pagination pagination --- adrf/pagination.py | 151 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 151 insertions(+) create mode 100644 adrf/pagination.py diff --git a/adrf/pagination.py b/adrf/pagination.py new file mode 100644 index 0000000..51ea742 --- /dev/null +++ b/adrf/pagination.py @@ -0,0 +1,151 @@ +""" +Pagination serializers determine the structure of the output that should +be used for paginated responses. +""" +from django.core.paginator import InvalidPage +from rest_framework.exceptions import NotFound +from rest_framework.pagination import _reverse_ordering, \ + PageNumberPagination as DRFPageNumberPagination, \ + LimitOffsetPagination as DRFLimitOffsetPagination, \ + CursorPagination as DRFCursorPagination + + +class PageNumberPagination(DRFPageNumberPagination): + async def paginate_queryset(self, queryset, request, view=None): + """ + Paginate a queryset if required, either returning a + page object, or `None` if pagination is not configured for this view. + """ + self.request = request + page_size = self.get_page_size(request) + if not page_size: + return None + + paginator = self.django_paginator_class(queryset, page_size) + page_number = self.get_page_number(request, paginator) + + try: + self.page = paginator.page(page_number) + except InvalidPage as exc: + msg = self.invalid_page_message.format( + page_number=page_number, message=str(exc) + ) + raise NotFound(msg) + + if paginator.num_pages > 1 and self.template is not None: + # The browsable API should display pagination controls. + self.display_page_controls = True + + return self.page + + +class LimitOffsetPagination(DRFLimitOffsetPagination): + async def aget_count(self, queryset): + """ + Determine an object count, supporting either querysets or regular lists. + """ + try: + return (await queryset.acount()) + except (AttributeError, TypeError): + return len(queryset) + + async def paginate_queryset(self, queryset, request, view=None): + self.request = request + self.limit = self.get_limit(request) + if self.limit is None: + return None + + self.count = await self.aget_count(queryset) + self.offset = self.get_offset(request) + if self.count > self.limit and self.template is not None: + self.display_page_controls = True + + if self.count == 0 or self.offset > self.count: + return [] + return queryset[self.offset:self.offset + self.limit] + + +class CursorPagination(DRFCursorPagination): + async def paginate_queryset(self, queryset, request, view=None): + self.request = request + self.page_size = self.get_page_size(request) + if not self.page_size: + return None + + self.base_url = request.build_absolute_uri() + self.ordering = self.get_ordering(request, queryset, view) + + self.cursor = self.decode_cursor(request) + if self.cursor is None: + (offset, reverse, current_position) = (0, False, None) + else: + (offset, reverse, current_position) = self.cursor + + # Cursor pagination always enforces an ordering. + if reverse: + queryset = queryset.order_by(*_reverse_ordering(self.ordering)) + else: + queryset = queryset.order_by(*self.ordering) + + # If we have a cursor with a fixed position then filter by that. + if current_position is not None: + order = self.ordering[0] + is_reversed = order.startswith('-') + order_attr = order.lstrip('-') + + # Test for: (cursor reversed) XOR (queryset reversed) + if self.cursor.reverse != is_reversed: + kwargs = {order_attr + '__lt': current_position} + else: + kwargs = {order_attr + '__gt': current_position} + + queryset = queryset.filter(**kwargs) + + # If we have an offset cursor then offset the entire page by that amount. + # We also always fetch an extra item in order to determine if there is a + # page following on from this one. + results = queryset[offset:offset + self.page_size + 1] + self.page = results[:self.page_size] + + # Determine the position of the final item following the page. + if (await results.acount()) > (await self.page.acount()): + has_following_position = True + num_elements = await results.acount() + async for item in queryset[offset + num_elements:offset + num_elements + 1]: + instance = item + break + following_position = self._get_position_from_instance( + instance, + self.ordering, + ) + else: + has_following_position = False + following_position = None + + if reverse: + # If we have a reverse queryset, then the query ordering was in reverse + # so we need to reverse the items again before returning them to the user. + # self.page = self.page.order_by(*_reverse_ordering(self.ordering)) + + # Determine next and previous positions for reverse cursors. + self.has_next = (current_position is not None) or (offset > 0) + self.has_previous = has_following_position + if self.has_next: + self.next_position = current_position + if self.has_previous: + self.previous_position = following_position + else: + # Determine next and previous positions for forward cursors. + self.has_next = has_following_position + self.has_previous = (current_position is not None) or (offset > 0) + if self.has_next: + self.next_position = following_position + if self.has_previous: + self.previous_position = current_position + + # Display page controls in the browsable API if there is more + # than one page. + if (self.has_previous or self.has_next) and self.template is not None: + self.display_page_controls = True + + return self.page