Add support for page_size parameter in CursorPaginator class

This commit is contained in:
Kris Dorosz 2017-07-05 10:35:17 +02:00 committed by Carlton Gibson
parent aecca9d8e8
commit 60b9e58a12
2 changed files with 195 additions and 1 deletions

View File

@ -482,6 +482,15 @@ class CursorPagination(BasePagination):
ordering = '-created'
template = 'rest_framework/pagination/previous_and_next.html'
# Client can control the page size using this query parameter.
# Default is 'None'. Set to eg 'page_size' to enable usage.
page_size_query_param = None
page_size_query_description = _('Number of results to return per page.')
# Set to an integer to limit the maximum page size the client may request.
# Only relevant if 'page_size_query_param' has also been set.
max_page_size = None
# The offset in the cursor is used in situations where we have a
# nearly-unique index. (Eg millisecond precision creation timestamps)
# We guard against malicious users attempting to cause expensive database
@ -566,6 +575,16 @@ class CursorPagination(BasePagination):
return self.page
def get_page_size(self, request):
if self.page_size_query_param:
try:
return _positive_int(
request.query_params[self.page_size_query_param],
strict=True,
cutoff=self.max_page_size
)
except (KeyError, ValueError):
pass
return self.page_size
def get_next_link(self):
@ -779,7 +798,7 @@ class CursorPagination(BasePagination):
def get_schema_fields(self, view):
assert coreapi is not None, 'coreapi must be installed to use `get_schema_fields()`'
assert coreschema is not None, 'coreschema must be installed to use `get_schema_fields()`'
return [
fields = [
coreapi.Field(
name=self.cursor_query_param,
required=False,
@ -790,3 +809,16 @@ class CursorPagination(BasePagination):
)
)
]
if self.page_size_query_param is not None:
fields.append(
coreapi.Field(
name=self.page_size_query_param,
required=False,
location='query',
schema=coreschema.Integer(
title='Page size',
description=force_text(self.page_size_query_description)
)
)
)
return fields

View File

@ -633,6 +633,164 @@ class CursorPaginationTestsMixin:
assert isinstance(self.pagination.to_html(), type(''))
def test_cursor_pagination_with_page_size(self):
(previous, current, next, previous_url, next_url) = self.get_pages('/?page_size=20')
assert previous is None
assert current == [1, 1, 1, 1, 1, 1, 2, 3, 4, 4, 4, 4, 5, 6, 7, 7, 7, 7, 7, 7]
assert next == [7, 7, 7, 8, 9, 9, 9, 9, 9, 9]
(previous, current, next, previous_url, next_url) = self.get_pages(next_url)
assert previous == [1, 1, 1, 1, 1, 1, 2, 3, 4, 4, 4, 4, 5, 6, 7, 7, 7, 7, 7, 7]
assert current == [7, 7, 7, 8, 9, 9, 9, 9, 9, 9]
assert next is None
def test_cursor_pagination_with_page_size_over_limit(self):
(previous, current, next, previous_url, next_url) = self.get_pages('/?page_size=30')
assert previous is None
assert current == [1, 1, 1, 1, 1, 1, 2, 3, 4, 4, 4, 4, 5, 6, 7, 7, 7, 7, 7, 7]
assert next == [7, 7, 7, 8, 9, 9, 9, 9, 9, 9]
(previous, current, next, previous_url, next_url) = self.get_pages(next_url)
assert previous == [1, 1, 1, 1, 1, 1, 2, 3, 4, 4, 4, 4, 5, 6, 7, 7, 7, 7, 7, 7]
assert current == [7, 7, 7, 8, 9, 9, 9, 9, 9, 9]
assert next is None
def test_cursor_pagination_with_page_size_zero(self):
(previous, current, next, previous_url, next_url) = self.get_pages('/?page_size=0')
assert previous is None
assert current == [1, 1, 1, 1, 1]
assert next == [1, 2, 3, 4, 4]
(previous, current, next, previous_url, next_url) = self.get_pages(next_url)
assert previous == [1, 1, 1, 1, 1]
assert current == [1, 2, 3, 4, 4]
assert next == [4, 4, 5, 6, 7]
(previous, current, next, previous_url, next_url) = self.get_pages(next_url)
assert previous == [1, 2, 3, 4, 4]
assert current == [4, 4, 5, 6, 7]
assert next == [7, 7, 7, 7, 7]
(previous, current, next, previous_url, next_url) = self.get_pages(next_url)
assert previous == [4, 4, 4, 5, 6] # Paging artifact
assert current == [7, 7, 7, 7, 7]
assert next == [7, 7, 7, 8, 9]
(previous, current, next, previous_url, next_url) = self.get_pages(next_url)
assert previous == [7, 7, 7, 7, 7]
assert current == [7, 7, 7, 8, 9]
assert next == [9, 9, 9, 9, 9]
(previous, current, next, previous_url, next_url) = self.get_pages(next_url)
assert previous == [7, 7, 7, 8, 9]
assert current == [9, 9, 9, 9, 9]
assert next is None
(previous, current, next, previous_url, next_url) = self.get_pages(previous_url)
assert previous == [7, 7, 7, 7, 7]
assert current == [7, 7, 7, 8, 9]
assert next == [9, 9, 9, 9, 9]
(previous, current, next, previous_url, next_url) = self.get_pages(previous_url)
assert previous == [4, 4, 5, 6, 7]
assert current == [7, 7, 7, 7, 7]
assert next == [8, 9, 9, 9, 9] # Paging artifact
(previous, current, next, previous_url, next_url) = self.get_pages(previous_url)
assert previous == [1, 2, 3, 4, 4]
assert current == [4, 4, 5, 6, 7]
assert next == [7, 7, 7, 7, 7]
(previous, current, next, previous_url, next_url) = self.get_pages(previous_url)
assert previous == [1, 1, 1, 1, 1]
assert current == [1, 2, 3, 4, 4]
assert next == [4, 4, 5, 6, 7]
(previous, current, next, previous_url, next_url) = self.get_pages(previous_url)
assert previous is None
assert current == [1, 1, 1, 1, 1]
assert next == [1, 2, 3, 4, 4]
def test_cursor_pagination_with_page_size_negative(self):
(previous, current, next, previous_url, next_url) = self.get_pages('/?page_size=-5')
assert previous is None
assert current == [1, 1, 1, 1, 1]
assert next == [1, 2, 3, 4, 4]
(previous, current, next, previous_url, next_url) = self.get_pages(next_url)
assert previous == [1, 1, 1, 1, 1]
assert current == [1, 2, 3, 4, 4]
assert next == [4, 4, 5, 6, 7]
(previous, current, next, previous_url, next_url) = self.get_pages(next_url)
assert previous == [1, 2, 3, 4, 4]
assert current == [4, 4, 5, 6, 7]
assert next == [7, 7, 7, 7, 7]
(previous, current, next, previous_url, next_url) = self.get_pages(next_url)
assert previous == [4, 4, 4, 5, 6] # Paging artifact
assert current == [7, 7, 7, 7, 7]
assert next == [7, 7, 7, 8, 9]
(previous, current, next, previous_url, next_url) = self.get_pages(next_url)
assert previous == [7, 7, 7, 7, 7]
assert current == [7, 7, 7, 8, 9]
assert next == [9, 9, 9, 9, 9]
(previous, current, next, previous_url, next_url) = self.get_pages(next_url)
assert previous == [7, 7, 7, 8, 9]
assert current == [9, 9, 9, 9, 9]
assert next is None
(previous, current, next, previous_url, next_url) = self.get_pages(previous_url)
assert previous == [7, 7, 7, 7, 7]
assert current == [7, 7, 7, 8, 9]
assert next == [9, 9, 9, 9, 9]
(previous, current, next, previous_url, next_url) = self.get_pages(previous_url)
assert previous == [4, 4, 5, 6, 7]
assert current == [7, 7, 7, 7, 7]
assert next == [8, 9, 9, 9, 9] # Paging artifact
(previous, current, next, previous_url, next_url) = self.get_pages(previous_url)
assert previous == [1, 2, 3, 4, 4]
assert current == [4, 4, 5, 6, 7]
assert next == [7, 7, 7, 7, 7]
(previous, current, next, previous_url, next_url) = self.get_pages(previous_url)
assert previous == [1, 1, 1, 1, 1]
assert current == [1, 2, 3, 4, 4]
assert next == [4, 4, 5, 6, 7]
(previous, current, next, previous_url, next_url) = self.get_pages(previous_url)
assert previous is None
assert current == [1, 1, 1, 1, 1]
assert next == [1, 2, 3, 4, 4]
class TestCursorPagination(CursorPaginationTestsMixin):
"""
@ -671,6 +829,8 @@ class TestCursorPagination(CursorPaginationTestsMixin):
class ExamplePagination(pagination.CursorPagination):
page_size = 5
page_size_query_param = 'page_size'
max_page_size = 20
ordering = 'created'
self.pagination = ExamplePagination()
@ -727,6 +887,8 @@ class TestCursorPaginationWithValueQueryset(CursorPaginationTestsMixin, TestCase
def setUp(self):
class ExamplePagination(pagination.CursorPagination):
page_size = 5
page_size_query_param = 'page_size'
max_page_size = 20
ordering = 'created'
self.pagination = ExamplePagination()