fix(graphql): Validate and apply limits when paginating evaluated querysets

Address review feedback on the pagination of evaluated querysets:

- Paginate an unevaluated clone using strawberry-django's apply() rather
  than re-implementing its slicing logic. This rejects a negative offset
  (as Django does for any other queryset) and applies the limits to the
  clone's query, so that re-evaluating it cannot return unpaginated
  results.
- Return the queryset unchanged when pagination would not exclude any
  results, avoiding a clone per object in the common case.
- Extend the test to cover multiple objects with differing tags, query
  batching, and rejection of a negative offset.

Refs #23383

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

netbox/netbox/graphql/pagination.py

@@ -10,6 +10,7 @@ __all__ = (
     'OffsetPaginationInfo',
     'OffsetPaginationInput',
     'apply_distinct_window_pagination',
+    'apply_evaluated_pagination',
     'apply_pagination',
 )
 
@@ -136,17 +137,34 @@ def apply_pagination(
             limit=pagination.limit,
         )
 
-    # Slicing a queryset which has already been evaluated (e.g. by a `prefetch_related` hint on the field)
-    # returns a list rather than a QuerySet, which strawberry-django's list hook cannot handle. Instead, slice
-    # the cached results and return them on a clone of the queryset.
     if pagination is not None and related_field_id is None and queryset._result_cache is not None:
-        start = pagination.offset
-        limit = _resolve_limit(pagination.limit)
-        stop = start + limit if limit is not None and limit >= 0 else None
-        results = queryset._result_cache[start:stop]
-        queryset = queryset._chain()
-        queryset._result_cache = results
-        queryset._prefetch_done = True
-        return queryset
+        return apply_evaluated_pagination(pagination, queryset)
 
     return apply(pagination, queryset, related_field_id=related_field_id)
+
+
+def apply_evaluated_pagination(pagination: OffsetPaginationInput, queryset: _QS) -> _QS:
+    """
+    Paginate a queryset which has already been evaluated (e.g. by a `prefetch_related` hint on the field).
+
+    strawberry-django's `apply()` slices the queryset directly, which for an evaluated queryset returns a list
+    rather than a QuerySet, and strawberry-django's list hook cannot handle a list. Instead, paginate an
+    unevaluated clone of the queryset, and populate its result cache by slicing the cached results to match.
+    """
+    # TODO: Remove this once strawberry-django's `apply()` returns a QuerySet for an evaluated queryset.
+    results = queryset._result_cache
+
+    # Avoid cloning the queryset if the pagination would not exclude any results (the common case, where
+    # only the implicit MAX_PAGE_SIZE limit has been applied).
+    if pagination.offset == 0:
+        limit = _resolve_limit(pagination.limit)
+        if limit is None or limit < 0 or limit >= len(results):
+            return queryset
+
+    # Paginating the clone validates the pagination (e.g. rejecting a negative offset) and applies the limits
+    # to its query, so that the results remain consistent should the queryset be cloned again later.
+    paginated = apply(pagination, queryset._chain())
+    paginated._result_cache = results[paginated.query.low_mark:paginated.query.high_mark]
+    paginated._prefetch_done = True
+
+    return paginated

netbox/netbox/tests/test_graphql.py

@@ -615,7 +615,8 @@ class GraphQLAPITestCase(APITestCase):
 
     def test_graphql_device_list_tags_pagination(self):
         """
-        Pagination can be applied to a prefetched list field (tags).
+        Pagination is applied independently to a prefetched list field (tags) for each object, without
+        resorting to a query per object.
         """
         self.add_permissions('dcim.view_device', 'extras.view_tag')
 
@@ -626,52 +627,75 @@ class GraphQLAPITestCase(APITestCase):
             slug='pagination-model',
         )
         device_role = DeviceRole.objects.create(name='Pagination Role', slug='pagination-role')
-        device = Device.objects.create(
-            name='Pagination Device',
-            device_type=device_type,
-            role=device_role,
-            site=Site.objects.first(),
-        )
-        device.tags.set([
-            Tag.objects.create(name=f'Pagination Tag {index}', slug=f'pagination-tag-{index}')
-            for index in range(5)
+        site = Site.objects.first()
+        devices = Device.objects.bulk_create([
+            Device(name=f'Pagination Device {index}', device_type=device_type, role=device_role, site=site)
+            for index in range(2)
         ])
+        tags = [
+            Tag.objects.create(name=f'Pagination Tag {index}', slug=f'pagination-tag-{index}')
+            for index in range(8)
+        ]
+        devices[0].tags.set(tags[:5])
+        devices[1].tags.set(tags[3:])
+        url = reverse('graphql')
 
-        query = """
-        {
-            device_list(filters: {role: {slug: {exact: "pagination-role"}}}) {
-                tags(pagination: {offset: 1, limit: 2}) {
-                    slug
+        def get_device_tags(tags_field):
+            query = """
+            {
+                device_list(filters: {role: {slug: {exact: "pagination-role"}}}) {
+                    name
+                    TAGS_FIELD {
+                        slug
+                    }
                 }
             }
-        }
-        """
-        url = reverse('graphql')
-        response = self.client.post(url, data={'query': query}, format='json', **self.header)
-        self.assertHttpStatus(response, status.HTTP_200_OK)
-        data = json.loads(response.content)
-        self.assertNotIn('errors', data)
-        self.assertEqual(
-            [tag['slug'] for tag in data['data']['device_list'][0]['tags']],
-            ['pagination-tag-1', 'pagination-tag-2'],
-        )
+            """.replace('TAGS_FIELD', tags_field)
+            with CaptureQueriesContext(connection) as context:
+                response = self.client.post(url, data={'query': query}, format='json', **self.header)
+            self.assertHttpStatus(response, status.HTTP_200_OK)
+            data = json.loads(response.content)
+            self.assertNotIn('errors', data)
+            tag_queries = sum(1 for record in context.captured_queries if 'extras_tag' in record['sql'])
+            self.assertLessEqual(tag_queries, 2, msg=f'Expected batched tag prefetch, got {tag_queries} tag queries')
+            return {
+                device['name']: [tag['slug'] for tag in device['tags']]
+                for device in data['data']['device_list']
+            }
+
+        # Explicit pagination
+        self.assertEqual(get_device_tags('tags(pagination: {offset: 1, limit: 2})'), {
+            'Pagination Device 0': ['pagination-tag-1', 'pagination-tag-2'],
+            'Pagination Device 1': ['pagination-tag-4', 'pagination-tag-5'],
+        })
+
+        # No pagination: all tags are returned
+        self.assertEqual(get_device_tags('tags'), {
+            'Pagination Device 0': [tag.slug for tag in tags[:5]],
+            'Pagination Device 1': [tag.slug for tag in tags[3:]],
+        })
 
         # MAX_PAGE_SIZE is enforced when no pagination is specified
+        with override_settings(MAX_PAGE_SIZE=3):
+            self.assertEqual(get_device_tags('tags'), {
+                'Pagination Device 0': [tag.slug for tag in tags[:3]],
+                'Pagination Device 1': [tag.slug for tag in tags[3:6]],
+            })
+
+        # A negative offset is rejected
         query = """
         {
             device_list(filters: {role: {slug: {exact: "pagination-role"}}}) {
-                tags {
+                tags(pagination: {offset: -2}) {
                     slug
                 }
             }
         }
         """
-        with override_settings(MAX_PAGE_SIZE=3):
-            response = self.client.post(url, data={'query': query}, format='json', **self.header)
-        self.assertHttpStatus(response, status.HTTP_200_OK)
+        response = self.client.post(url, data={'query': query}, format='json', **self.header)
         data = json.loads(response.content)
-        self.assertNotIn('errors', data)
-        self.assertEqual(len(data['data']['device_list'][0]['tags']), 3)
+        self.assertIn('errors', data)
+        self.assertIn('Negative indexing is not supported', data['errors'][0]['message'])
 
     def test_graphql_ip_address_list_assigned_object(self):
         """