Merge pull request #205 from sarendsen/develop

Add LimitOffsetPagination

Author: jerel

example/tests/unit/test_pagination.py

@@ -0,0 +1,79 @@
+from collections import OrderedDict
+
+from rest_framework.request import Request
+from rest_framework.test import APIRequestFactory
+from rest_framework.utils.urls import replace_query_param
+
+from rest_framework_json_api.pagination import LimitOffsetPagination
+
+
+factory = APIRequestFactory()
+
+
+class TestLimitOffset:
+    """
+    Unit tests for `pagination.LimitOffsetPagination`.
+    """
+
+    def setup(self):
+        class ExamplePagination(LimitOffsetPagination):
+            default_limit = 10
+            max_limit = 15
+
+        self.pagination = ExamplePagination()
+        self.queryset = range(1, 101)
+        self.base_url = 'http://testserver/'
+
+    def paginate_queryset(self, request):
+        return list(self.pagination.paginate_queryset(self.queryset, request))
+
+    def get_paginated_content(self, queryset):
+        response = self.pagination.get_paginated_response(queryset)
+        return response.data
+
+    def get_test_request(self, arguments):
+        return Request(factory.get('/', arguments))
+
+    def test_valid_offset_limit(self):
+        """
+        Basic test, assumes offset and limit are given.
+        """
+        offset = 10
+        limit = 5
+        count = len(self.queryset)
+        last_offset = count - limit
+        next_offset = 15
+        prev_offset = 5
+
+        request = self.get_test_request({
+            self.pagination.limit_query_param: limit,
+            self.pagination.offset_query_param: offset
+        })
+        base_url = replace_query_param(self.base_url, self.pagination.limit_query_param, limit)
+        last_url = replace_query_param(base_url, self.pagination.offset_query_param, last_offset)
+        first_url = base_url
+        next_url = replace_query_param(base_url, self.pagination.offset_query_param, next_offset)
+        prev_url = replace_query_param(base_url, self.pagination.offset_query_param, prev_offset)
+        queryset = self.paginate_queryset(request)
+        content = self.get_paginated_content(queryset)
+        next_offset = offset + limit
+
+        expected_content = {
+            'results': list(range(offset + 1, next_offset + 1)),
+            'links': OrderedDict([
+                ('first', first_url),
+                ('last', last_url),
+                ('next', next_url),
+                ('prev', prev_url),
+            ]),
+            'meta': {
+                'pagination': OrderedDict([
+                    ('count', count),
+                    ('limit', limit),
+                    ('offset', offset),
+                ])
+            }
+        }
+
+        assert queryset == list(range(offset + 1, next_offset + 1))
+        assert content == expected_content

rest_framework_json_api/pagination.py

@@ -4,8 +4,8 @@
 from collections import OrderedDict
 from rest_framework import serializers
 from rest_framework.views import Response
-from rest_framework.pagination import PageNumberPagination
-from rest_framework.templatetags.rest_framework import replace_query_param
+from rest_framework.pagination import PageNumberPagination, LimitOffsetPagination
+from rest_framework.utils.urls import remove_query_param, replace_query_param
 
 
 class PageNumberPagination(PageNumberPagination):
@@ -47,3 +47,52 @@ def get_paginated_response(self, data):
                 ('prev', self.build_link(previous))
             ])
         })
+
+
+class LimitOffsetPagination(LimitOffsetPagination):
+    """
+    A limit/offset based style. For example:
+    http://api.example.org/accounts/?page[limit]=100
+    http://api.example.org/accounts/?page[offset]=400&page[limit]=100
+    """
+    limit_query_param = 'page[limit]'
+    offset_query_param = 'page[offset]'
+
+    def get_last_link(self):
+        if self.count == 0:
+            return None
+
+        url = self.request.build_absolute_uri()
+        url = replace_query_param(url, self.limit_query_param, self.limit)
+
+        offset = self.count - self.limit
+
+        if offset <= 0:
+            return remove_query_param(url, self.offset_query_param)
+
+        return replace_query_param(url, self.offset_query_param, offset)
+
+    def get_first_link(self):
+        if self.count == 0:
+            return None
+        
+        url = self.request.build_absolute_uri()
+        return remove_query_param(url, self.offset_query_param)
+    
+    def get_paginated_response(self, data):
+        return Response({
+            'results': data,
+            'meta': {
+                'pagination': OrderedDict([
+                    ('count', self.count),
+                    ('limit', self.limit),
+                    ('offset', self.offset),
+                ])
+            },
+            'links': OrderedDict([
+                ('first', self.get_first_link()),
+                ('last', self.get_last_link()),
+                ('next', self.get_next_link()),
+                ('prev', self.get_previous_link())
+            ])
+        })