diff --git a/api/api/admin/oauth.py b/api/api/admin/oauth.py
index 07daebf460f..e7a7a2ac1ef 100644
--- a/api/api/admin/oauth.py
+++ b/api/api/admin/oauth.py
@@ -1,12 +1,65 @@
+from django import forms
 from django.contrib import admin
 
 from oauth2_provider.models import AccessToken
 
+from api.constants.privilege import Privilege
+from api.models.oauth import ThrottledApplication
+
 
 def register(site):
+    site.register(ThrottledApplication, ThrottledApplicationAdmin)
     site.register(AccessToken, AccessTokenAdmin)
 
 
+class ThrottledApplicationAdminForm(forms.ModelForm):
+    class Meta:
+        model = ThrottledApplication
+        exclude = (
+            "client_type",
+            "redirect_uris",
+            "post_logout_redirect_uris",
+            "skip_authorization",
+            "algorithm",
+            "user",
+        )
+
+    # ArrayField doesn't have a good default field, so use a multiple choice field, but
+    # override default widget of multi-<select>, which is much more annoying to use
+    # and easy to accidentally un-select an option. The multi-checkbox is much easier to use
+    privileges = forms.MultipleChoiceField(
+        choices=Privilege.choices,
+        required=False,
+        widget=forms.CheckboxSelectMultiple,
+    )
+
+
+class ThrottledApplicationAdmin(admin.ModelAdmin):
+    form = ThrottledApplicationAdminForm
+    view_on_site = False
+
+    search_fields = ("client_id", "name", "rate_limit_model", "privileges")
+    list_display = ("client_id", "name", "created", "rate_limit_model", "privileges")
+    ordering = ("-created",)
+
+    readonly_fields = (
+        "name",
+        "created",
+        "client_id",
+        "verified",
+        "authorization_grant_type",
+        "client_secret",
+    )
+
+    def has_delete_permission(self, *args, **kwargs):
+        """
+        Disallow deleting throttled applications. Use ``revoke`` instead.
+
+        This also hides the delete button on the change view.
+        """
+        return False
+
+
 class AccessTokenAdmin(admin.ModelAdmin):
     search_fields = ("token", "id")
     list_display = ("token", "id", "created", "scope", "expires")
diff --git a/api/api/constants/privilege.py b/api/api/constants/privilege.py
new file mode 100644
index 00000000000..b892ec7d424
--- /dev/null
+++ b/api/api/constants/privilege.py
@@ -0,0 +1,72 @@
+import typing
+from dataclasses import dataclass
+
+from rest_framework.request import Request
+
+
+ANONYMOUS: typing.Literal["anonymous"] = "anonymous"
+AUTHENTICATED: typing.Literal["authenticated"] = "authenticated"
+PRIVILEGED: typing.Literal["privileged"] = "privileged"
+
+Level = typing.Literal[ANONYMOUS, AUTHENTICATED, PRIVILEGED]
+LEVELS = typing.get_args(Level)
+
+
+_PRIVILEGES = {}
+
+
+@dataclass
+class Privilege:
+    """
+    Privileges granted to applications upon approved request to Openverse maintainers.
+
+    Maintainers review requests for increased privileges on a per-case basis.
+
+    Distinct from ``rate_limit_model`` which only affects access rates rather than privileges.
+    """
+
+    slug: str
+    anonymous: typing.Any
+    authenticated: typing.Any
+    privileged: typing.Any
+
+    def __post_init__(self):
+        _PRIVILEGES[self.slug] = self
+
+    def request_level(self, request: None | Request) -> tuple[Level, typing.Any]:
+        """Retrieve the level of any request in relation to the privilege."""
+        if request is None or request.auth is None:
+            return ANONYMOUS, self.anonymous
+
+        if self.slug in request.auth.application.privileges:
+            return PRIVILEGED, self.privileged
+
+        return AUTHENTICATED, self.authenticated
+
+    @classmethod
+    @property
+    def choices(cls):
+        return ((slug,) * 2 for slug in _PRIVILEGES)
+
+
+PAGE_SIZE = Privilege(
+    "page_size",
+    anonymous=20,
+    authenticated=50,
+    # Max out privileged page size at the maximum authenticated
+    # pagination depth, otherwise privileged page size limit can
+    # contradict pagination depth limit.
+    privileged=240,
+)
+
+PAGINATION_DEPTH = Privilege(
+    "pagination_depth",
+    # 12 pages of 20 results
+    # Both anon and authed are limited to the same depth
+    # authed users can request bigger pages, but still only the same
+    # overall number of results available
+    anonymous=12 * PAGE_SIZE.anonymous,
+    authenticated=12 * PAGE_SIZE.anonymous,
+    # 20 pages of maxed out page sizes for privileged apps
+    privileged=20 * PAGE_SIZE.privileged,
+)
diff --git a/api/api/migrations/0064_throttledapplication_privileges.py b/api/api/migrations/0064_throttledapplication_privileges.py
new file mode 100644
index 00000000000..1216efec883
--- /dev/null
+++ b/api/api/migrations/0064_throttledapplication_privileges.py
@@ -0,0 +1,19 @@
+# Generated by Django 4.2.11 on 2024-05-23 01:41
+
+import django.contrib.postgres.fields
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('api', '0063_merge_20240521_0843'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='throttledapplication',
+            name='privileges',
+            field=django.contrib.postgres.fields.ArrayField(base_field=models.CharField(choices=[('page_size', 'page_size'), ('pagination_depth', 'pagination_depth')]), default=list, help_text='Privileges granted to applications upon approved request to Openverse maintainers.  Maintainers review requests for increased privileges on a per-case basis.  Distinct from ``rate_limit_model`` which only affects access rates rather than privileges.', size=None),
+        ),
+    ]
diff --git a/api/api/models/oauth.py b/api/api/models/oauth.py
index 4154f18f504..e19801381e4 100644
--- a/api/api/models/oauth.py
+++ b/api/api/models/oauth.py
@@ -1,7 +1,12 @@
+from textwrap import dedent
+
+from django.contrib.postgres.fields import ArrayField
 from django.db import models
 
 from oauth2_provider.models import AbstractApplication
 
+from api.constants import privilege
+
 
 class OAuth2Registration(models.Model):
     """Information about API key applicants."""
@@ -40,6 +45,14 @@ class ThrottledApplication(AbstractApplication):
     verified = models.BooleanField(default=False)
     revoked = models.BooleanField(default=False)
 
+    privileges = ArrayField(
+        models.CharField(
+            choices=privilege.Privilege.choices,
+        ),
+        default=list,
+        help_text=dedent(privilege.Privilege.__doc__).strip().replace("\n", " "),
+    )
+
 
 class OAuth2Verification(models.Model):
     """
diff --git a/api/api/serializers/base.py b/api/api/serializers/base.py
index 2230045a3ab..df4170e6925 100644
--- a/api/api/serializers/base.py
+++ b/api/api/serializers/base.py
@@ -1,4 +1,7 @@
+import typing
+
 from rest_framework import serializers
+from rest_framework.request import Request
 
 
 class BaseModelSerializer(serializers.ModelSerializer):
@@ -19,3 +22,14 @@ def build_property_field(self, field_name, model_class):
         if doc := getattr(model_class, field_name).__doc__:
             kwargs.setdefault("help_text", doc)
         return klass, kwargs
+
+
+class BaseRequestSerializer(serializers.Serializer):
+    class Context(typing.TypedDict):
+        request: typing.Required[Request]
+
+    context: Context
+
+    @property
+    def request(self) -> Request | None:
+        return self.context.get("request", None)
diff --git a/api/api/serializers/media_serializers.py b/api/api/serializers/media_serializers.py
index bcefcdb8a2b..f1790f400eb 100644
--- a/api/api/serializers/media_serializers.py
+++ b/api/api/serializers/media_serializers.py
@@ -1,5 +1,5 @@
 from collections import namedtuple
-from typing import TypedDict
+from math import floor
 
 from django.conf import settings
 from django.core.exceptions import ValidationError as DjangoValidationError
@@ -7,13 +7,12 @@
 from django.urls import reverse
 from rest_framework import serializers
 from rest_framework.exceptions import NotAuthenticated, ValidationError
-from rest_framework.request import Request
 
 from drf_spectacular.utils import extend_schema_serializer
 from elasticsearch_dsl.response import Hit
 from openverse_attribution.license import License
 
-from api.constants import sensitivity
+from api.constants import privilege, sensitivity
 from api.constants.licenses import LICENSE_GROUPS
 from api.constants.media_types import MediaType
 from api.constants.parameters import COLLECTION, TAG
@@ -21,7 +20,7 @@
 from api.constants.sorting import DESCENDING, RELEVANCE, SORT_DIRECTIONS, SORT_FIELDS
 from api.controllers import search_controller
 from api.models.media import AbstractMedia
-from api.serializers.base import BaseModelSerializer
+from api.serializers.base import BaseModelSerializer, BaseRequestSerializer
 from api.serializers.docs import (
     COLLECTION_HELP_TEXT,
     CREATOR_HELP_TEXT,
@@ -40,40 +39,36 @@
 #######################
 
 
-class PaginatedRequestSerializer(serializers.Serializer):
+class PaginatedRequestSerializer(BaseRequestSerializer):
     """This serializer passes pagination parameters from the query string."""
 
+    _SUBJECT_TO_PAGINATION_LIMITS = (
+        "This parameter is subject to limitations based on authentication "
+        "and special privileges. For details, refer to [the authentication "
+        "documentation](#tag/auth)."
+    )
+
     field_names = [
         "page_size",
         "page",
     ]
     page_size = serializers.IntegerField(
         label="page_size",
-        help_text=f"Number of results to return per page. "
-        f"Maximum is {settings.MAX_AUTHED_PAGE_SIZE} for authenticated "
-        f"requests, and {settings.MAX_ANONYMOUS_PAGE_SIZE} for "
-        f"unauthenticated requests.",
+        help_text=f"Number of results to return per page. {_SUBJECT_TO_PAGINATION_LIMITS}",
         required=False,
-        default=settings.MAX_ANONYMOUS_PAGE_SIZE,
+        default=privilege.PAGE_SIZE.anonymous,
         min_value=1,
     )
     page = serializers.IntegerField(
         label="page",
-        help_text="The page of results to retrieve.",
+        help_text=f"The page of results to retrieve. {_SUBJECT_TO_PAGINATION_LIMITS}",
         required=False,
         default=1,
-        max_value=settings.MAX_PAGINATION_DEPTH,
         min_value=1,
     )
 
     def validate_page_size(self, value):
-        request = self.context.get("request")
-        is_anonymous = getattr(request, "auth", None) is None
-        max_value = (
-            settings.MAX_ANONYMOUS_PAGE_SIZE
-            if is_anonymous
-            else settings.MAX_AUTHED_PAGE_SIZE
-        )
+        level, max_value = privilege.PAGE_SIZE.request_level(self.request)
 
         validator = MaxValueValidator(
             max_value,
@@ -82,19 +77,67 @@ def validate_page_size(self, value):
             ),
         )
 
-        if is_anonymous:
-            try:
-                validator(value)
-            except (ValidationError, DjangoValidationError) as e:
-                raise NotAuthenticated(
-                    detail=e.message,
-                    code=e.code,
-                )
-        else:
+        try:
             validator(value)
+        except (ValidationError, DjangoValidationError) as e:
+            if level == privilege.PRIVILEGED:
+                raise
+
+            raise NotAuthenticated(
+                detail=f"page_size may not exceed {max_value} for {level} requests",
+                code=e.code,
+            )
 
         return value
 
+    def clamp_result_count(self, real_result_count):
+        _, max_depth = privilege.PAGINATION_DEPTH.request_level(self.request)
+
+        if real_result_count > max_depth:
+            return max_depth
+
+        return real_result_count
+
+    def clamp_page_count(self, real_page_count):
+        _, max_depth = privilege.PAGINATION_DEPTH.request_level(self.request)
+
+        page_size = self.data["page_size"]
+        max_possible_page_count = max_depth / page_size
+
+        if real_page_count > max_possible_page_count:
+            return floor(max_possible_page_count)
+
+        return real_page_count
+
+    def validate(self, data):
+        data = super().validate(data)
+
+        # pagination depth is validated as a combination of page and page size,
+        # and so cannot be validated in the individual field validation methods
+        level, max_depth = privilege.PAGINATION_DEPTH.request_level(self.request)
+
+        requested_pagination_depth = data["page"] * data["page_size"]
+
+        pagination_depth_validator = MaxValueValidator(
+            max_depth,
+            message=serializers.IntegerField.default_error_messages["max_value"].format(
+                max_value=max_depth
+            ),
+        )
+
+        try:
+            pagination_depth_validator(requested_pagination_depth)
+        except (ValidationError, DjangoValidationError) as e:
+            if level == privilege.PRIVILEGED:
+                raise
+
+            raise NotAuthenticated(
+                detail=f"pagination depth may not exceed {max_depth} for {level} requests",
+                code=e.code,
+            )
+
+        return data
+
 
 @extend_schema_serializer(
     # Hide internal fields from documentation.
@@ -275,10 +318,9 @@ class MediaSearchRequestSerializer(PaginatedRequestSerializer):
         required=False,
     )
 
-    class Context(TypedDict, total=True):
+    class Context(BaseRequestSerializer.Context, total=True):
         warnings: list[dict]
         media_type: MediaType
-        request: Request
 
     context: Context
 
@@ -401,7 +443,7 @@ def validate_source(self, value):
                     f"Refer to the source list for valid options: {sources_list}."
                 )
             elif invalid_sources := (sources - valid_sources):
-                available_sources_uri = self.context["request"].build_absolute_uri(
+                available_sources_uri = self.request.build_absolute_uri(
                     reverse(f"{self.media_type}-stats")
                 )
                 self.context["warnings"].append(
diff --git a/api/api/utils/pagination.py b/api/api/utils/pagination.py
index 8c3534504ec..b70f1b42928 100644
--- a/api/api/utils/pagination.py
+++ b/api/api/utils/pagination.py
@@ -1,4 +1,3 @@
-from django.conf import settings
 from rest_framework.pagination import PageNumberPagination
 from rest_framework.response import Response
 
@@ -22,7 +21,7 @@ def __init__(self, *args, **kwargs):
     def get_paginated_response(self, data):
         response = {
             "result_count": self.result_count,
-            "page_count": min(settings.MAX_PAGINATION_DEPTH, self.page_count),
+            "page_count": self.page_count,
             "page_size": self.page_size,
             "page": self.page,
             "results": data,
diff --git a/api/api/views/media_views.py b/api/api/views/media_views.py
index 54839fbeaba..daf6688a8d3 100644
--- a/api/api/views/media_views.py
+++ b/api/api/views/media_views.py
@@ -189,8 +189,8 @@ def get_media_results(
                 filter_dead,
                 page,
             )
-            self.paginator.page_count = num_pages
-            self.paginator.result_count = num_results
+            self.paginator.page_count = params.clamp_page_count(num_pages)
+            self.paginator.result_count = params.clamp_result_count(num_results)
         except ValueError as e:
             raise APIException(getattr(e, "message", str(e)))
 
diff --git a/api/conf/settings/misc.py b/api/conf/settings/misc.py
index 0b78b985a69..22ca88759cd 100644
--- a/api/conf/settings/misc.py
+++ b/api/conf/settings/misc.py
@@ -32,7 +32,3 @@
     "OUTBOUND_USER_AGENT_TEMPLATE",
     default=f"Openverse{{purpose}}/{API_VERSION} (https://wordpress.org/openverse)",
 )
-
-MAX_ANONYMOUS_PAGE_SIZE = 20
-MAX_AUTHED_PAGE_SIZE = 500
-MAX_PAGINATION_DEPTH = 20
diff --git a/api/conf/settings/spectacular.py b/api/conf/settings/spectacular.py
index 6f67ee62bb2..61a2233a7c1 100644
--- a/api/conf/settings/spectacular.py
+++ b/api/conf/settings/spectacular.py
@@ -50,19 +50,17 @@
             "name": "auth",
             "description": dedent(
                 """
-                The API has rate-limiting and throttling in place to prevent
-                abuse. In the response for each request that is subject to
-                rate-limits, you can see the `X-RateLimit-` headers for info
-                about your permitted and available usage. Exceeding the limit
-                will result in '429: Too Many Requests' responses.
-
-                - Anonymous clients are limited to make 5 req/hour and
-                  100 req/day. Therefore we recommend registering for
-                  authenticated usage.
+                Openverse provides free and open access to the Openverse API to
+                anonymous and registered users. [Refer to the API documentation
+                site for information on how to register](https://api.openverse.org/v1/#tag/auth).
 
-                - Authenticated clients can make 100 req/minute and 10,000
-                  req/day. Additionally, you can request a higher limit to fit
-                  your application's needs.
+                All Openverse API users are subject to rate limits and restrictions
+                on how much of Openverse's dataset can be accessed through the API.
+                [Individuals should contact Openverse to request expanded access to
+                the API](https://github.com/WordPress/openverse#keep-in-touch).
+                Requests are considered on a case-by-case basis and are subject to
+                evaluation in light of [Openverse's Terms of Service](https://docs.openverse.org/terms_of_service.html).
+                Escalated access may be revoked at any time.
 
                 To authenticate yourself, you must sign up for an API key using
                 the `register` endpoint and then get an access token using the
@@ -74,6 +72,35 @@
                 ```
                 Authorization: Bearer <access_token>
                 ```
+
+                ### Rate limits
+
+                Openverse endpoints are rate limited. Anonymous requests should be
+                sufficient for most users. Indeed, https://openverse.org itself
+                operates using anonymous requests from the browser.
+
+                Registered users are automatically granted slightly higher limits.
+                Further increases to rate limits are available upon request (see above).
+
+                Every Openverse API response that was subject to rate-limits includes
+                headers outlining the permitted and available usage. Exceeding the limit
+                will result in '429: Too Many Requests' responses.
+
+                ### Pagination
+
+                Openverse's dataset is valuable, and the
+                [Terms of Service](https://docs.openverse.org/terms_of_service.html)
+                disallow scraping under all circumstances. As such, pagination
+                for anonymous users is limited, accommodating only the typical
+                usage on https://openverse.org. Pagination is limited in terms
+                of the size of individual pages and the total number of works
+                visible for a query (pagination depth).
+
+                Authenticated users are subject to the same limit of total works
+                available for a query, but may request larger individual pages.
+
+                Increases to pagination limits on page size and total depth are
+                available upon request (see above).
                 """
             ),
         },
diff --git a/api/latest_migrations/api b/api/latest_migrations/api
index b1c2f634688..48c55803d28 100644
--- a/api/latest_migrations/api
+++ b/api/latest_migrations/api
@@ -2,4 +2,4 @@
 # If you have a merge conflict in this file, it means you need to run:
 #     manage.py makemigrations --merge
 # in order to resolve the conflict between migrations.
-0063_merge_20240521_0843
+0064_throttledapplication_privileges
diff --git a/api/test/conftest.py b/api/test/conftest.py
index 04672679528..939d8b58ede 100644
--- a/api/test/conftest.py
+++ b/api/test/conftest.py
@@ -7,8 +7,19 @@
     unreachable_django_cache,
     unreachable_redis,
 )
-from test.fixtures.media_type_config import __all__ as media_type_config_all
-from test.fixtures.rest_framework import api_client, request_factory
+from test.fixtures.media_type_config import (
+    audio_media_type_config,
+    cleanup_elasticsearch_test_documents,
+    image_media_type_config,
+    media_type_config,
+)
+from test.fixtures.rest_framework import (
+    access_token,
+    anon_request,
+    api_client,
+    authed_request,
+    request_factory,
+)
 
 
 __all__ = [
@@ -21,4 +32,11 @@
     "unreachable_redis",
     "api_client",
     "request_factory",
-] + media_type_config_all
+    "access_token",
+    "authed_request",
+    "anon_request",
+    "image_media_type_config",
+    "audio_media_type_config",
+    "media_type_config",
+    "cleanup_elasticsearch_test_documents",
+]
diff --git a/api/test/fixtures/media_type_config.py b/api/test/fixtures/media_type_config.py
index 0e4089104a6..9672c091933 100644
--- a/api/test/fixtures/media_type_config.py
+++ b/api/test/fixtures/media_type_config.py
@@ -1,4 +1,5 @@
 from dataclasses import dataclass
+from typing import Iterable
 
 import pook
 import pytest
@@ -51,13 +52,13 @@ class MediaTypeConfig:
     report_serializer: MediaReportRequestSerializer
     report_factory: MediaReportFactory
     deleted_class: AbstractDeletedMedia
-    providers: list[str]
+    providers: Iterable[str]
     """providers for the media type from the sample data"""
 
-    categories: list[str]
+    categories: Iterable[str]
     """categories for the media type from the sample data"""
 
-    tags: list[str]
+    tags: Iterable[str]
     """tags for the media type from the sample data"""
 
     q: str
@@ -83,9 +84,9 @@ def indexes(self):
         report_factory=model_factories.ImageReportFactory,
         sensitive_class=SensitiveImage,
         deleted_class=DeletedImage,
-        providers=["flickr", "stocksnap"],
-        categories=["photograph"],
-        tags=["cat", "Cat"],
+        providers=("flickr", "stocksnap"),
+        categories=("photograph"),
+        tags=("cat", "Cat"),
         q="dog",
     ),
     "audio": MediaTypeConfig(
@@ -102,9 +103,9 @@ def indexes(self):
         report_factory=model_factories.AudioReportFactory,
         sensitive_class=SensitiveAudio,
         deleted_class=DeletedAudio,
-        providers=["freesound", "jamendo", "wikimedia_audio", "ccmixter"],
-        categories=["music", "pronunciation"],
-        tags=["cat"],
+        providers=("freesound", "jamendo", "wikimedia_audio", "ccmixter"),
+        categories=("music", "pronunciation"),
+        tags=("cat",),
         q="love",
     ),
 }
@@ -154,11 +155,3 @@ def cleanup_elasticsearch_test_documents(request, settings):
         query={"match": {"tags.name": CREATED_BY_FIXTURE_MARKER}},
         refresh=True,
     )
-
-
-__all__ = [
-    "image_media_type_config",
-    "audio_media_type_config",
-    "media_type_config",
-    "cleanup_elasticsearch_test_documents",
-]
diff --git a/api/test/fixtures/rest_framework.py b/api/test/fixtures/rest_framework.py
index f96eeb0aacd..404a7ee3458 100644
--- a/api/test/fixtures/rest_framework.py
+++ b/api/test/fixtures/rest_framework.py
@@ -1,7 +1,10 @@
 from rest_framework.test import APIClient, APIRequestFactory
+from rest_framework.views import APIView
 
 import pytest
 
+from test.factory.models.oauth2 import AccessTokenFactory
+
 
 @pytest.fixture
 def api_client() -> APIClient:
@@ -13,3 +16,25 @@ def request_factory() -> APIRequestFactory():
     request_factory = APIRequestFactory(defaults={"REMOTE_ADDR": "192.0.2.1"})
 
     return request_factory
+
+
+@pytest.fixture
+def access_token():
+    token = AccessTokenFactory.create()
+    token.application.verified = True
+    token.application.save()
+    return token
+
+
+@pytest.fixture
+def authed_request(access_token, request_factory):
+    request = request_factory.get(
+        "/", HTTP_AUTHORIZATION=f"Bearer {access_token.token}"
+    )
+
+    return APIView().initialize_request(request)
+
+
+@pytest.fixture
+def anon_request(request_factory):
+    return APIView().initialize_request(request_factory.get("/"))
diff --git a/api/test/integration/test_auth.py b/api/test/integration/test_auth.py
index e5dbe2f0b06..8d254735539 100644
--- a/api/test/integration/test_auth.py
+++ b/api/test/integration/test_auth.py
@@ -7,6 +7,7 @@
 import pytest
 from oauth2_provider.models import AccessToken
 
+from api.constants import privilege
 from api.models import OAuth2Verification, ThrottledApplication
 
 
@@ -249,34 +250,6 @@ def test_authority_authed(
     assert source == exp_source
 
 
-@pytest.mark.django_db
-def test_page_size_limit_unauthed(api_client):
-    query_params = {"page_size": 20}
-    res = api_client.get("/v1/images/", query_params)
-    assert res.status_code == 200
-    query_params["page_size"] = 21
-    res = api_client.get("/v1/images/", query_params)
-    assert res.status_code == 401
-
-
-@pytest.mark.django_db
-def test_page_size_limit_authed(api_client, test_auth_token_exchange):
-    time.sleep(1)
-    token = test_auth_token_exchange["access_token"]
-    query_params = {"page_size": 21}
-    res = api_client.get(
-        "/v1/images/", query_params, HTTP_AUTHORIZATION=f"Bearer {token}"
-    )
-
-    assert res.status_code == 200
-
-    query_params = {"page_size": 500}
-    res = api_client.get(
-        "/v1/images/", query_params, HTTP_AUTHORIZATION=f"Bearer {token}"
-    )
-    assert res.status_code == 200
-
-
 @pytest.mark.django_db
 def test_invalid_credentials_401(api_client):
     res = api_client.get(
@@ -302,3 +275,102 @@ def test_revoked_application_access(api_client, test_auth_token_exchange):
     res = api_client.get("/v1/images/", HTTP_AUTHORIZATION=f"Bearer {token}")
 
     assert res.status_code == 403
+
+
+@pytest.mark.parametrize(
+    "level, page_size_modification, allowed",
+    (
+        config
+        for level in privilege.LEVELS
+        for config in (
+            (level, -1, True),
+            (level, 0, True),
+            (level, +1, False),
+        )
+    ),
+)
+@pytest.mark.django_db
+def test_page_size_privileges(
+    api_client, test_auth_token_exchange, level, page_size_modification, allowed
+):
+    if level == privilege.ANONYMOUS:
+        authorization = ""
+    else:
+        token = test_auth_token_exchange["access_token"]
+        application = AccessToken.objects.get(token=token).application
+
+        if level == privilege.PRIVILEGED:
+            application.privileges.append(privilege.PAGE_SIZE.slug)
+
+        application.save()
+        authorization = f"Bearer {token}"
+
+    limit = getattr(privilege.PAGE_SIZE, level)
+
+    res = api_client.get(
+        "/v1/images/",
+        {"page_size": limit + page_size_modification},
+        HTTP_AUTHORIZATION=authorization,
+    )
+
+    if allowed:
+        assert res.status_code == 200
+    elif level != privilege.PRIVILEGED:
+        assert res.status_code == 401
+    else:
+        # When privileged, the request errors on bad request, rather than unauthorized
+        # maybe we should just always use bad request instead?
+        assert res.status_code == 400
+
+
+@pytest.mark.parametrize(
+    "level, pagination_depth_modification, allowed",
+    (
+        config
+        for level in privilege.LEVELS
+        for config in (
+            (level, -1, True),
+            (level, 0, True),
+            (level, +1, False),
+        )
+    ),
+)
+@pytest.mark.django_db
+def test_pagination_depth_privileges(
+    api_client, test_auth_token_exchange, level, pagination_depth_modification, allowed
+):
+    if level == privilege.ANONYMOUS:
+        authorization = ""
+    else:
+        token = test_auth_token_exchange["access_token"]
+        application = AccessToken.objects.get(token=token).application
+
+        if level == privilege.PRIVILEGED:
+            application.privileges.append(privilege.PAGINATION_DEPTH.slug)
+
+        application.save()
+        authorization = f"Bearer {token}"
+
+    depth_limit = getattr(privilege.PAGINATION_DEPTH, level)
+
+    page_size_limit = privilege.PAGE_SIZE.anonymous
+
+    last_page = int(depth_limit / page_size_limit)
+
+    res = api_client.get(
+        "/v1/images/",
+        {
+            "page_size": page_size_limit,
+            "page": last_page + pagination_depth_modification,
+        },
+        HTTP_AUTHORIZATION=authorization,
+    )
+
+    if allowed:
+        assert res.status_code == 200
+    elif level != privilege.PRIVILEGED:
+        assert res.status_code == 401
+    else:
+        # When privileged, the request errors on bad request, rather than unauthorized
+        # maybe we should just always use bad request instead?
+        assert res.status_code == 400
diff --git a/api/test/integration/test_dead_link_filter.py b/api/test/integration/test_dead_link_filter.py
index 6fe7de4cca4..1afb2c028bd 100644
--- a/api/test/integration/test_dead_link_filter.py
+++ b/api/test/integration/test_dead_link_filter.py
@@ -1,10 +1,9 @@
 from unittest.mock import patch
 from uuid import uuid4
 
-from django.conf import settings
-
 import pytest
 
+from api.constants import privilege
 from api.controllers.elasticsearch.helpers import DEAD_LINK_RATIO
 
 
@@ -179,8 +178,8 @@ def test_page_size_removing_dead_links(search_without_dead_links):
 def test_page_consistency_removing_dead_links(search_without_dead_links):
     """Test that results in consecutive pages don't repeat when filtering dead links."""
 
-    total_pages = settings.MAX_PAGINATION_DEPTH
     page_size = 5
+    total_pages = int(privilege.PAGINATION_DEPTH.anonymous / page_size)
 
     page_results = []
     for page in range(1, total_pages + 1):
@@ -198,11 +197,3 @@ def no_duplicates(xs):
     ids = list(map(lambda x: x["id"], page_results))
     # No results should be repeated so we should have no duplicate ids
     assert no_duplicates(ids)
-
-
-@pytest.mark.django_db
-def test_max_page_count(api_client):
-    response = api_client.get(
-        "/v1/images/", {"page": settings.MAX_PAGINATION_DEPTH + 1}
-    )
-    assert response.status_code == 400
diff --git a/api/test/unit/controllers/test_search_controller_search_query.py b/api/test/unit/controllers/test_search_controller_search_query.py
index 5a03e672705..b97084d5df3 100644
--- a/api/test/unit/controllers/test_search_controller_search_query.py
+++ b/api/test/unit/controllers/test_search_controller_search_query.py
@@ -32,9 +32,10 @@ def enabled_sources_cache(django_cache, monkeypatch):
     cache.delete(ENABLED_SOURCES_CACHE_KEY, version=ENABLED_SOURCES_CACHE_VERSION)
 
 
-def test_create_search_query_empty(media_type_config):
+def test_create_search_query_empty(media_type_config, anon_request):
     serializer = media_type_config.search_request_serializer(
-        data={}, context={"media_type": "image"}
+        data={},
+        context={"media_type": media_type_config.media_type, "request": anon_request},
     )
     serializer.is_valid(raise_exception=True)
     search_query = search_controller.build_search_query(serializer)
@@ -49,10 +50,13 @@ def test_create_search_query_empty(media_type_config):
     }
 
 
-def test_create_search_query_empty_no_ranking(media_type_config, settings):
+def test_create_search_query_empty_no_ranking(
+    media_type_config, anon_request, settings
+):
     settings.USE_RANK_FEATURES = False
     serializer = media_type_config.search_request_serializer(
-        data={}, context={"media_type": media_type_config.media_type}
+        data={},
+        context={"media_type": media_type_config.media_type, "request": anon_request},
     )
     serializer.is_valid(raise_exception=True)
     search_query = search_controller.build_search_query(serializer)
@@ -64,9 +68,10 @@ def test_create_search_query_empty_no_ranking(media_type_config, settings):
     }
 
 
-def test_create_search_query_q_search_no_filters(media_type_config):
+def test_create_search_query_q_search_no_filters(media_type_config, anon_request):
     serializer = media_type_config.search_request_serializer(
-        data={"q": "cat"}, context={"media_type": media_type_config.media_type}
+        data={"q": "cat"},
+        context={"media_type": media_type_config.media_type, "request": anon_request},
     )
     serializer.is_valid(raise_exception=True)
     search_query = search_controller.build_search_query(serializer)
@@ -98,10 +103,12 @@ def test_create_search_query_q_search_no_filters(media_type_config):
     }
 
 
-def test_create_search_query_q_search_with_quotes_adds_raw_suffix(media_type_config):
+def test_create_search_query_q_search_with_quotes_adds_raw_suffix(
+    media_type_config, anon_request
+):
     serializer = media_type_config.search_request_serializer(
         data={"q": '"The cutest cat"'},
-        context={"media_type": media_type_config.media_type},
+        context={"media_type": media_type_config.media_type, "request": anon_request},
     )
     serializer.is_valid(raise_exception=True)
     search_query = search_controller.build_search_query(serializer)
@@ -134,7 +141,9 @@ def test_create_search_query_q_search_with_quotes_adds_raw_suffix(media_type_con
     }
 
 
-def test_create_search_query_q_search_with_filters(image_media_type_config):
+def test_create_search_query_q_search_with_filters(
+    image_media_type_config, anon_request
+):
     serializer = image_media_type_config.search_request_serializer(
         data={
             "q": "cat",
@@ -148,7 +157,10 @@ def test_create_search_query_q_search_with_filters(image_media_type_config):
             "unstable__authority_boost": "2.5",
             "unstable__include_sensitive_results": True,
         },
-        context={"media_type": image_media_type_config.media_type},
+        context={
+            "media_type": image_media_type_config.media_type,
+            "request": anon_request,
+        },
     )
     serializer.is_valid(raise_exception=True)
     search_query = search_controller.build_search_query(serializer)
@@ -185,14 +197,17 @@ def test_create_search_query_q_search_with_filters(image_media_type_config):
     }
 
 
-def test_create_search_query_non_q_query(image_media_type_config):
+def test_create_search_query_non_q_query(image_media_type_config, anon_request):
     serializer = image_media_type_config.search_request_serializer(
         data={
             "creator": "Artist From Openverse",
             "title": "kitten🐱",
             "tags": "cute",
         },
-        context={"media_type": image_media_type_config.media_type},
+        context={
+            "media_type": image_media_type_config.media_type,
+            "request": anon_request,
+        },
     )
     serializer.is_valid(raise_exception=True)
     search_query = search_controller.build_search_query(serializer)
@@ -231,13 +246,17 @@ def test_create_search_query_non_q_query(image_media_type_config):
 
 def test_create_search_query_q_search_license_license_type_creates_2_terms_filters(
     image_media_type_config,
+    anon_request,
 ):
     serializer = image_media_type_config.search_request_serializer(
         data={
             "license": "by-nc",
             "license_type": "commercial",
         },
-        context={"media_type": image_media_type_config.media_type},
+        context={
+            "media_type": image_media_type_config.media_type,
+            "request": anon_request,
+        },
     )
     serializer.is_valid(raise_exception=True)
     search_query = search_controller.build_search_query(serializer)
@@ -271,9 +290,14 @@ def test_create_search_query_q_search_license_license_type_creates_2_terms_filte
 def test_create_search_query_empty_with_dynamically_enabled_sources(
     image_media_type_config,
     enabled_sources_cache,
+    anon_request,
 ):
     serializer = image_media_type_config.search_request_serializer(
-        data={}, context={"media_type": image_media_type_config.media_type}
+        data={},
+        context={
+            "media_type": image_media_type_config.media_type,
+            "request": anon_request,
+        },
     )
     serializer.is_valid(raise_exception=True)
 
@@ -322,9 +346,15 @@ def test_create_search_query_empty_with_dynamically_enabled_sources(
         ),
     ],
 )
-def test_build_collection_query(image_media_type_config, data, expected_query_filter):
+def test_build_collection_query(
+    image_media_type_config, data, expected_query_filter, anon_request
+):
     serializer = image_media_type_config.search_request_serializer(
-        data=data, context={"media_type": image_media_type_config.media_type}
+        data=data,
+        context={
+            "media_type": image_media_type_config.media_type,
+            "request": anon_request,
+        },
     )
     serializer.is_valid(raise_exception=True)
     actual_query = search_controller.build_collection_query(serializer)
diff --git a/api/test/unit/serializers/test_audio_serializers.py b/api/test/unit/serializers/test_audio_serializers.py
index 365be087a04..d96a5e77467 100644
--- a/api/test/unit/serializers/test_audio_serializers.py
+++ b/api/test/unit/serializers/test_audio_serializers.py
@@ -1,8 +1,5 @@
 import uuid
 
-from rest_framework.request import Request
-from rest_framework.test import APIRequestFactory
-
 import pytest
 
 from api.models.audio import Audio
@@ -21,11 +18,8 @@ def audio_fixture():
 
 
 @pytest.mark.django_db
-def test_audio_serializer_omit_peaks_by_default(audio_fixture):
-    factory = APIRequestFactory()
-    request = factory.get(f"audio/{audio_fixture.identifier}")
-    request = Request(request)
-    mock_ctx = {"request": request}
+def test_audio_serializer_omit_peaks_by_default(audio_fixture, anon_request):
+    mock_ctx = {"request": anon_request}
 
     audio_serializer = AudioSerializer(instance=audio_fixture, context=mock_ctx)
     assert "peaks" not in audio_serializer.data
@@ -33,11 +27,8 @@ def test_audio_serializer_omit_peaks_by_default(audio_fixture):
 
 @pytest.mark.django_db
 @pytest.mark.parametrize("include_peaks", [True, False])
-def test_audio_serializer_with_peaks_param(audio_fixture, include_peaks):
-    factory = APIRequestFactory()
-    request = factory.get(f"audio/{audio_fixture.identifier}/?peaks={include_peaks}")
-    request = Request(request)
-    mock_ctx = {"request": request, "validated_data": {"peaks": include_peaks}}
+def test_audio_serializer_with_peaks_param(audio_fixture, anon_request, include_peaks):
+    mock_ctx = {"request": anon_request, "validated_data": {"peaks": include_peaks}}
 
     audio_serializer = AudioSerializer(instance=audio_fixture, context=mock_ctx)
     assert ("peaks" in audio_serializer.data) is include_peaks
@@ -45,7 +36,7 @@ def test_audio_serializer_with_peaks_param(audio_fixture, include_peaks):
 
 # https://github.com/WordPress/openverse/issues/3930
 @pytest.mark.django_db
-def test_audio_serializer_with_non_required_alt_audio_fields_missing():
+def test_audio_serializer_with_non_required_alt_audio_fields_missing(anon_request):
     alt_files = [
         {"bit_rate": 128, "filetype": "mp3", "url": "https://example.com/audio.mp3"}
     ]
@@ -55,10 +46,7 @@ def test_audio_serializer_with_non_required_alt_audio_fields_missing():
         alt_files=alt_files,
     )
     audio.save()
-    factory = APIRequestFactory()
-    request = factory.get(f"audio/{audio.identifier}/?peaks=false")
-    request = Request(request)
-    mock_ctx = {"request": request}
+    mock_ctx = {"request": anon_request}
 
     audio_serializer = AudioSerializer(instance=audio, context=mock_ctx)
 
diff --git a/api/test/unit/serializers/test_media_serializers.py b/api/test/unit/serializers/test_media_serializers.py
index 79059210ee4..74304d5a853 100644
--- a/api/test/unit/serializers/test_media_serializers.py
+++ b/api/test/unit/serializers/test_media_serializers.py
@@ -2,27 +2,12 @@
 import uuid
 from unittest.mock import MagicMock, patch
 
-from django.conf import settings
-from rest_framework.exceptions import NotAuthenticated
-from rest_framework.serializers import ValidationError
-from rest_framework.test import force_authenticate
-from rest_framework.views import APIView
-
 import pytest
 
 from api.constants import sensitivity
 from api.serializers.audio_serializers import AudioSearchRequestSerializer
 from api.serializers.image_serializers import ImageSearchRequestSerializer
 from api.serializers.media_serializers import MediaSearchRequestSerializer
-from test.factory.models.oauth2 import AccessTokenFactory
-
-
-@pytest.fixture
-def access_token():
-    token = AccessTokenFactory.create()
-    token.application.verified = True
-    token.application.save()
-    return token
 
 
 @pytest.fixture
@@ -35,59 +20,6 @@ def hit():
     return hit
 
 
-@pytest.fixture
-def authed_request(access_token, request_factory):
-    request = request_factory.get("/")
-
-    force_authenticate(request, token=access_token.token)
-
-    return APIView().initialize_request(request)
-
-
-@pytest.fixture
-def anon_request(request_factory):
-    return APIView().initialize_request(request_factory.get("/"))
-
-
-@pytest.mark.django_db
-@pytest.mark.parametrize(
-    ("page_size", "authenticated"),
-    (
-        pytest.param(-1, False, marks=pytest.mark.raises(exception=ValidationError)),
-        pytest.param(0, False, marks=pytest.mark.raises(exception=ValidationError)),
-        (1, False),
-        (settings.MAX_ANONYMOUS_PAGE_SIZE, False),
-        pytest.param(
-            settings.MAX_ANONYMOUS_PAGE_SIZE + 1,
-            False,
-            marks=pytest.mark.raises(exception=NotAuthenticated),
-        ),
-        pytest.param(
-            settings.MAX_AUTHED_PAGE_SIZE,
-            False,
-            marks=pytest.mark.raises(exception=NotAuthenticated),
-        ),
-        pytest.param(-1, True, marks=pytest.mark.raises(exception=ValidationError)),
-        pytest.param(0, True, marks=pytest.mark.raises(exception=ValidationError)),
-        (1, True),
-        (settings.MAX_ANONYMOUS_PAGE_SIZE + 1, True),
-        (settings.MAX_AUTHED_PAGE_SIZE, True),
-        pytest.param(
-            settings.MAX_AUTHED_PAGE_SIZE + 1,
-            True,
-            marks=pytest.mark.raises(exception=ValidationError),
-        ),
-    ),
-)
-def test_page_size_validation(page_size, authenticated, anon_request, authed_request):
-    request = authed_request if authenticated else anon_request
-    serializer = MediaSearchRequestSerializer(
-        context={"request": request, "media_type": "image"},
-        data={"page_size": page_size},
-    )
-    assert serializer.is_valid(raise_exception=True)
-
-
 def test_media_serializer_adds_license_url_if_missing(
     anon_request, hit, media_type_config
 ):
@@ -100,15 +32,14 @@ def test_media_serializer_adds_license_url_if_missing(
 
 
 def test_media_serializer_recovers_invalid_or_duplicate_source(
-    media_type_config, request_factory
+    media_type_config, anon_request
 ):
     sources = {
         "image": ("flickr,flickr,invalid", "flickr"),
         "audio": ("freesound,freesound,invalid", "freesound"),
     }
-    request = request_factory.get("/v1/images/")
     serializer_class = media_type_config.search_request_serializer(
-        context={"media_type": media_type_config.media_type, "request": request},
+        context={"media_type": media_type_config.media_type, "request": anon_request},
         data={"source": sources[media_type_config.media_type][0]},
     )
     assert serializer_class.is_valid()
diff --git a/api/test/unit/utils/test_throttle.py b/api/test/unit/utils/test_throttle.py
index 975276c3858..9232b087e13 100644
--- a/api/test/unit/utils/test_throttle.py
+++ b/api/test/unit/utils/test_throttle.py
@@ -1,6 +1,5 @@
 from django.http import HttpResponse
 from rest_framework.settings import api_settings
-from rest_framework.test import force_authenticate
 from rest_framework.views import APIView
 
 import pytest
@@ -8,7 +7,6 @@
 from api.utils import throttle
 from api.views.media_views import MediaViewSet
 from test.factory.models.image import ImageFactory
-from test.factory.models.oauth2 import AccessTokenFactory
 
 
 cache_availability_params = pytest.mark.parametrize(
@@ -67,24 +65,6 @@ def enable_throttles(settings):
     throttle.SimpleRateThrottle.THROTTLE_RATES = api_settings.DEFAULT_THROTTLE_RATES
 
 
-@pytest.fixture
-def access_token():
-    token = AccessTokenFactory.create()
-    token.application.verified = True
-    token.application.client_id = 123
-    token.application.save()
-    return token
-
-
-@pytest.fixture
-def authed_request(access_token, request_factory):
-    request = request_factory.get("/")
-
-    force_authenticate(request, token=access_token.token)
-
-    return request
-
-
 @pytest.fixture
 def view():
     return APIView()