pagination

pagination

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