Coverage for netbox/api/pagination.py: 85%
116 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1from django.db.models import QuerySet
2from django.utils.translation import gettext_lazy as _
3from rest_framework.exceptions import ValidationError
4from rest_framework.pagination import LimitOffsetPagination
5from rest_framework.utils.urls import remove_query_param, replace_query_param
7from netbox.api.exceptions import QuerySetNotOrdered
8from netbox.config import get_config
11class NetBoxPagination(LimitOffsetPagination):
12 """
13 Provides two mutually exclusive pagination mechanisms: offset-based and cursor-based.
15 Offset-based pagination employs `offset` and (optionally) `limit` parameters to page through results following the
16 model's natural order. `offset` indicates the number of results to skip. This provides very human-friendly behavior,
17 but performance can suffer when querying very large data sets due the overhead required to determine the starting
18 point in the database.
20 Cursor-based pagination employs `start` and (optionally) `limit` parameters to page through results as ordered by
21 the model's primary key (i.e. `id`). `start` indicates the numeric ID of the first object to return; `limit`
22 indicates the maximum number of objects to return beginning with the specified ID. Objects *must* be ordered by ID
23 to ensure pagination is consistent. This approach is less human-friendly but offers superior performance to
24 offset-based pagination. In cursor mode, `count` is omitted (null) for performance.
26 Offset- and cursor-based pagination are mutually exclusive: Only `offset` _or_ `start` is permitted for a request.
28 `limit` may be set to zero (`?limit=0`). This returns all objects matching a query, but retains the same format as
29 a paginated request. The limit can only be disabled if `MAX_PAGE_SIZE` has been set to 0 or None.
30 """
31 start_query_param = 'start'
33 def __init__(self):
34 self.default_limit = get_config().PAGINATE_COUNT
35 self.start = None
36 self._page_length = 0
37 self._last_pk = None
39 def paginate_queryset(self, queryset, request, view=None):
41 if isinstance(queryset, QuerySet) and not queryset.ordered: 41 ↛ 42line 41 didn't jump to line 42 because the condition on line 41 was never true
42 raise QuerySetNotOrdered(
43 "Paginating over an unordered queryset is unreliable. Ensure that a minimal "
44 "ordering has been applied to the queryset for this API endpoint."
45 )
47 self.start = self.get_start(request)
48 self.limit = self.get_limit(request)
49 self.request = request
51 # Cursor-based pagination
52 if self.start is not None:
53 if self.offset_query_param in request.query_params:
54 raise ValidationError(
55 _("'{start_param}' and '{offset_param}' are mutually exclusive.").format(
56 start_param=self.start_query_param,
57 offset_param=self.offset_query_param,
58 )
59 )
60 if 'ordering' in request.query_params:
61 raise ValidationError(_("Ordering cannot be specified in conjunction with cursor-based pagination."))
63 self.count = None
64 self.offset = 0
66 queryset = queryset.filter(pk__gte=self.start).order_by('pk')
67 results = list(queryset[:self.limit]) if self.limit else list(queryset)
69 self._page_length = len(results)
70 if results: 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true
71 self._last_pk = results[-1].pk if hasattr(results[-1], 'pk') else results[-1]['pk']
73 return results
75 # Offset-based pagination
76 if isinstance(queryset, QuerySet): 76 ↛ 80line 76 didn't jump to line 80 because the condition on line 76 was always true
77 self.count = self.get_queryset_count(queryset)
78 else:
79 # We're dealing with an iterable, not a QuerySet
80 self.count = len(queryset)
82 self.offset = self.get_offset(request)
84 if self.limit and self.count > self.limit and self.template is not None:
85 self.display_page_controls = True
87 if self.count == 0 or self.offset > self.count:
88 return list()
90 if self.limit: 90 ↛ 92line 90 didn't jump to line 92 because the condition on line 90 was always true
91 return list(queryset[self.offset:self.offset + self.limit])
92 return list(queryset[self.offset:])
94 def get_start(self, request):
95 try:
96 value = int(request.query_params[self.start_query_param])
97 if value < 0:
98 raise ValidationError(
99 _("Invalid '{param}' parameter: must be a non-negative integer.").format(
100 param=self.start_query_param,
101 )
102 )
103 return value
104 except KeyError:
105 return None
106 except (ValueError, TypeError):
107 raise ValidationError(
108 _("Invalid '{param}' parameter: must be a non-negative integer.").format(
109 param=self.start_query_param,
110 )
111 )
113 def get_limit(self, request):
114 max_limit = self.default_limit
115 MAX_PAGE_SIZE = get_config().MAX_PAGE_SIZE
116 if MAX_PAGE_SIZE: 116 ↛ 119line 116 didn't jump to line 119 because the condition on line 116 was always true
117 max_limit = min(max_limit, MAX_PAGE_SIZE)
119 if self.limit_query_param: 119 ↛ 135line 119 didn't jump to line 135 because the condition on line 119 was always true
120 try:
121 limit = int(request.query_params[self.limit_query_param])
122 if limit < 0:
123 raise ValueError()
125 if MAX_PAGE_SIZE: 125 ↛ 131line 125 didn't jump to line 131 because the condition on line 125 was always true
126 if limit == 0:
127 max_limit = MAX_PAGE_SIZE
128 else:
129 max_limit = min(MAX_PAGE_SIZE, limit)
130 else:
131 max_limit = limit
132 except (KeyError, ValueError):
133 pass
135 return max_limit
137 def get_queryset_count(self, queryset):
138 return queryset.count()
140 def get_next_link(self):
142 # Pagination has been disabled
143 if not self.limit: 143 ↛ 144line 143 didn't jump to line 144 because the condition on line 143 was never true
144 return None
146 # Cursor mode
147 if self.start is not None:
148 if self._page_length < self.limit: 148 ↛ 150line 148 didn't jump to line 150 because the condition on line 148 was always true
149 return None
150 url = self.request.build_absolute_uri()
151 url = replace_query_param(url, self.start_query_param, self._last_pk + 1)
152 url = replace_query_param(url, self.limit_query_param, self.limit)
153 url = remove_query_param(url, self.offset_query_param)
154 return url
156 return super().get_next_link()
158 def get_previous_link(self):
160 # Pagination has been disabled
161 if not self.limit: 161 ↛ 162line 161 didn't jump to line 162 because the condition on line 161 was never true
162 return None
164 # Cursor mode: forward-only
165 if self.start is not None:
166 return None
168 return super().get_previous_link()
170 def get_schema_operation_parameters(self, view):
171 parameters = super().get_schema_operation_parameters(view)
172 parameters.append({
173 'name': self.start_query_param,
174 'required': False,
175 'in': 'query',
176 'description': (
177 'Cursor-based pagination: return results with pk >= start, ordered by pk. '
178 'Mutually exclusive with offset.'
179 ),
180 'schema': {
181 'type': 'integer',
182 },
183 })
184 return parameters
187class StripCountAnnotationsPaginator(NetBoxPagination):
188 """
189 Strips the annotations on the queryset before getting the count
190 to optimize pagination of complex queries.
191 """
192 def get_queryset_count(self, queryset):
193 # Clone the queryset to avoid messing up the actual query
194 cloned_queryset = queryset.all()
195 cloned_queryset.query.annotations.clear()
197 return cloned_queryset.count()
200class LimitOffsetListPagination(LimitOffsetPagination):
201 """
202 DRF LimitOffset Paginator but for list instead of queryset
203 """
204 count = 0
205 offset = 0
207 def paginate_list(self, data, request, view=None):
208 self.request = request
209 self.limit = self.get_limit(request)
210 self.count = len(data)
211 self.offset = self.get_offset(request)
213 if self.limit is None: 213 ↛ 216line 213 didn't jump to line 216 because the condition on line 213 was always true
214 self.limit = self.count
216 if self.count == 0 or self.offset > self.count:
217 return []
219 if self.count > self.limit and self.template is not None: 219 ↛ 220line 219 didn't jump to line 220 because the condition on line 219 was never true
220 self.display_page_controls = True
222 return data[self.offset:self.offset + self.limit]