1
1
from collections import OrderedDict
2
+ from typing import Optional , Sequence , Tuple , Type , Union
2
3
from urllib .parse import parse_qs , urlparse
3
4
4
- from rest_framework .pagination import LimitOffsetPagination as _LimitOffsetPagination
5
+ from django .db .models import QuerySet
6
+ from rest_framework import serializers
7
+ from rest_framework .pagination import BasePagination
5
8
from rest_framework .pagination import CursorPagination as _CursorPagination
9
+ from rest_framework .pagination import LimitOffsetPagination as _LimitOffsetPagination
10
+ from rest_framework .request import Request
6
11
from rest_framework .response import Response
12
+ from rest_framework .views import APIView
7
13
8
14
9
- def get_paginated_response ( * , pagination_class , serializer_class , queryset , request , view ):
10
- paginator = pagination_class ( )
15
+ class _TurnOffPaginationSerializer ( serializers . Serializer ):
16
+ paginate = serializers . BooleanField ( default = True )
11
17
12
- page = paginator .paginate_queryset (queryset , request , view = view )
13
18
14
- if page is not None :
15
- serializer = serializer_class (page , many = True )
16
- return paginator .get_paginated_response (serializer .data )
17
-
18
- serializer = serializer_class (queryset , many = True )
19
+ def turn_off_pagination (data ):
20
+ serializer = _TurnOffPaginationSerializer (data = data )
21
+ serializer .is_valid (raise_exception = True )
19
22
20
- return Response ( data = serializer .data )
23
+ return serializer .validated_data [ "paginate" ]
21
24
22
25
23
26
class LimitOffsetPagination (_LimitOffsetPagination ):
24
27
default_limit = 10
25
28
max_limit = 50
26
29
27
- def get_count (self , queryset ) -> int :
30
+ def get_count (self , queryset : Union [ QuerySet , Sequence ] ) -> int :
28
31
"""
29
32
Determine an object count, supporting either querysets or regular lists.
30
33
"""
@@ -42,44 +45,52 @@ def get_count(self, queryset) -> int:
42
45
return len (queryset )
43
46
44
47
def get_paginated_data (self , data ):
45
- return OrderedDict ([
46
- ('limit' , self .limit ),
47
- ('offset' , self .offset ),
48
- ('count' , self .count ),
49
- ('next' , self .get_next_link ()),
50
- ('previous' , self .get_previous_link ()),
51
- ('results' , data )
52
- ])
48
+ return OrderedDict (
49
+ [
50
+ ("limit" , self .limit ),
51
+ ("offset" , self .offset ),
52
+ ("count" , self .count ),
53
+ ("next" , self .get_next_link ()),
54
+ ("previous" , self .get_previous_link ()),
55
+ ("results" , data ),
56
+ ]
57
+ )
53
58
54
59
def get_paginated_response (self , data ):
55
60
"""
56
61
We redefine this method in order to return `limit` and `offset`.
57
62
This is used by the frontend to construct the pagination itself.
58
63
"""
59
- return Response (OrderedDict ([
60
- ('limit' , self .limit ),
61
- ('offset' , self .offset ),
62
- ('count' , self .count ),
63
- ('next' , self .get_next_link ()),
64
- ('previous' , self .get_previous_link ()),
65
- ('results' , data )
66
- ]))
64
+ return Response (
65
+ OrderedDict (
66
+ [
67
+ ("limit" , self .limit ),
68
+ ("offset" , self .offset ),
69
+ ("count" , self .count ),
70
+ ("next" , self .get_next_link ()),
71
+ ("previous" , self .get_previous_link ()),
72
+ ("results" , data ),
73
+ ]
74
+ )
75
+ )
67
76
68
77
69
78
class CursorPagination (_CursorPagination ):
70
79
page_size = 50 # Return 50 items by default
71
80
72
- def __init__ (self , ordering ):
81
+ def __init__ (self , ordering : Optional [ str ] ):
73
82
self .ordering : str = ordering or "-created_at"
74
83
75
- def get_ordering (self , request , queryset , view ):
84
+ def get_ordering (
85
+ self , request : Request , queryset : QuerySet , view : APIView
86
+ ) -> Tuple [str ]:
76
87
# The DRF CursorPagination expects the ordering as a tuple
77
88
if isinstance (self .ordering , str ):
78
89
return (self .ordering ,)
79
90
80
91
return tuple (self .ordering )
81
92
82
- def _get_cursor (self , url ) :
93
+ def _get_cursor (self , url : Optional [ str ]) -> Optional [ str ] :
83
94
if not url :
84
95
return None
85
96
@@ -109,3 +120,42 @@ def get_paginated_response(self, data):
109
120
]
110
121
)
111
122
)
123
+
124
+
125
+ def _init_pagination_class (
126
+ pagination_class : Type [BasePagination ],
127
+ ordering : Optional [str ],
128
+ ) -> BasePagination :
129
+ if isinstance (pagination_class , CursorPagination ):
130
+ return pagination_class (ordering = ordering )
131
+
132
+ return pagination_class ()
133
+
134
+
135
+ def response_paginate (
136
+ * ,
137
+ pagination_class : Type [BasePagination ],
138
+ serializer_class : Type [serializers .Serializer ],
139
+ queryset : QuerySet ,
140
+ request : Request ,
141
+ view : APIView ,
142
+ ordering : Optional [str ] = "-created_at"
143
+ ) -> Response :
144
+ paginate = turn_off_pagination (data = request .GET )
145
+
146
+ if not paginate :
147
+ data = serializer_class (queryset , many = True ).data
148
+
149
+ return Response (data = data )
150
+
151
+ paginator = _init_pagination_class (pagination_class , ordering )
152
+
153
+ page = paginator .paginate_queryset (queryset , request , view = view )
154
+
155
+ if page is not None :
156
+ serializer = serializer_class (page , many = True )
157
+ return paginator .get_paginated_response (serializer .data )
158
+
159
+ serializer = serializer_class (queryset , many = True )
160
+
161
+ return Response (data = serializer .data )
0 commit comments