Python 基础体系 · 第 80/112 篇。示例统一以 Python 3.14 为语言基线;第三方库使用与其兼容的现代稳定版本,版本敏感行为会单独说明。

Django REST Framework:Serializer、ViewSet、权限、分页和限流

Django REST Framework(DRF)是建立在 Django 请求、模型、认证和 URL 路由之上的 Web API 工具包。它解决的不是“把 Django View 改成返回 JSON”这么简单的问题,而是把一次 API 请求拆成若干具有明确职责的阶段:

  1. 解析 HTTP 请求;
  2. 认证请求者;
  3. 检查权限;
  4. 检查限流;
  5. 查询或修改模型;
  6. 用 Serializer 验证输入或生成输出;
  7. 分页查询结果;
  8. 将 Python 原生数据渲染为 JSON 等响应格式;
  9. 将异常转换为统一的 HTTP 错误响应。

本文以 Python 3.14 和 Django 5.2 为范围。当前 DRF 官方文档列出的支持范围包括 Python 3.10 至 3.14,以及 Django 5.2、6.0 和 6.1;实际项目仍应固定经过验证的 Django、DRF 和数据库驱动版本,而不是只依赖“大版本兼容”。(docs.djangoproject.com)


一、先建立整体模型:一次请求如何穿过 DRF

下面的关系比单独记忆某个类名更重要:

sequenceDiagram
    participant C as Client
    participant U as URL Router
    participant V as ViewSet
    participant A as Authentication
    participant P as Permission
    participant T as Throttle
    participant Q as QuerySet/DB
    participant S as Serializer
    participant R as Renderer

    C->>U: HTTP 请求
    U->>V: 绑定 method -> action
    V->>A: 认证
    A-->>V: request.user / request.auth
    V->>P: 权限检查
    P-->>V: 允许或拒绝
    V->>T: 限流检查
    T-->>V: 允许或拒绝
    V->>Q: 查询对象或集合
    Q-->>V: Model / QuerySet
    V->>S: 输入验证或输出序列化
    S-->>V: validated_data / primitive data
    V->>R: 内容协商与渲染
    R-->>C: JSON / HTML 等响应

在这个流程中,认证 Authentication回答“你是谁”,权限 Permission回答“你能不能访问”,限流 Throttling回答“你现在是否暂时超出请求额度”。三者不能互相替代。

权限检查在 View 主体代码之前执行;只要权限列表中任意一个权限检查失败,View 主体就不会运行。限流也在 View 主体之前执行,但限流表示一种临时状态,而不是永久授权关系。(django-rest-framework.org)


二、Serializer:API 契约、验证和对象转换

2.1 Serializer 到底解决什么问题

Serializer 是 API 数据边界上的转换器。它至少承担两种相反方向的工作:

请求 JSON
    ↓ Parser
Python 原生数据
    ↓ Serializer.is_valid()
validated_data
    ↓ create() / update()
Django Model

Django Model / QuerySet
    ↓ Serializer(instance)
Python 原生数据
    ↓ Renderer
响应 JSON

DRF 官方对 Serializer 的定义包含两个关键点:

  • 将 Model、QuerySet 等复杂对象转换为可渲染的 Python 原生数据类型;
  • 将已经解析的请求数据转换为经过验证的结构化数据。(django-rest-framework.org)

因此,Serializer 不是简单的“JSON 格式化工具”,也不是 Django Model 的自动镜像。它是 API 的输入和输出契约

例如,数据库模型可能包含:

class Article(models.Model):
    title = models.CharField(max_length=200)
    body = models.TextField()
    owner = models.ForeignKey(User, on_delete=models.CASCADE)
    is_published = models.BooleanField(default=False)
    internal_note = models.TextField(blank=True)
    created_at = models.DateTimeField(auto_now_add=True)

但 API 不一定应该暴露 internal_note,也不应该允许客户端提交任意 owner。所以 API Serializer 可以只暴露:

from rest_framework import serializers

class ArticleSerializer(serializers.ModelSerializer):
    owner = serializers.ReadOnlyField(source="owner.username")

    class Meta:
        model = Article
        fields = [
            "id",
            "title",
            "body",
            "owner",
            "is_published",
            "created_at",
        ]
        read_only_fields = ["id", "owner", "created_at"]

ModelSerializer 会根据模型字段自动生成很多 Serializer 字段,但 fields 仍然应该明确指定。fields = "__all__" 在原型阶段方便,却会让未来新增的模型字段自动进入 API 边界,增加数据泄漏和兼容性风险。


2.2 序列化输出:Model 到 JSON 之间还有一层

下面的代码:

article = Article.objects.first()
serializer = ArticleSerializer(article)

print(serializer.data)

得到的是类似这样的 Python 数据:

{
    "id": 1,
    "title": "DRF 入门",
    "body": "正文",
    "owner": "alice",
    "is_published": False,
    "created_at": "2026-09-01T10:00:00Z",
}

这里的 serializer.data 还不是最终的 JSON 字节。它是由字符串、数字、布尔值、列表、字典等组成的可渲染数据。Response 依赖 Renderer 将这些数据转换成 JSON、Browsable API HTML 等具体表示;Renderer 不能直接处理 Django Model 实例,所以应先经过 Serializer。(django-rest-framework.org)

from rest_framework.response import Response

def list(self, request):
    queryset = Article.objects.all()
    serializer = ArticleSerializer(queryset, many=True)
    return Response(serializer.data)

这里必须使用 many=True,因为传入的是 QuerySet。若把 QuerySet 当成单个对象序列化,Serializer 会按照单对象路径处理,通常会产生类型错误或错误的数据结构。


2.3 反序列化:请求数据不会自动变成 Model

客户端发送:

POST /api/articles/
Content-Type: application/json

{
  "title": "新文章",
  "body": "正文",
  "is_published": true
}

DRF 首先通过 Parser 将请求体解析为 Python 数据。访问 request.data 时,如果 JSON 格式错误,会产生 ParseError 并默认返回 400;如果 Content-Type 没有可用 Parser,则会返回 415 Unsupported Media Type。(django-rest-framework.org)

随后,View 要显式创建 Serializer:

serializer = ArticleSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
article = serializer.save(owner=request.user)

这四步分别代表:

  1. data=request.data:告诉 Serializer 这是输入数据;
  2. is_valid():执行字段级、对象级和模型相关验证;
  3. raise_exception=True:验证失败时抛出 ValidationError
  4. save(owner=request.user):将额外的服务端数据注入保存过程。

验证成功后,数据会进入:

serializer.validated_data

例如:

{
    "title": "新文章",
    "body": "正文",
    "is_published": True,
    "owner": request.user,
}

调用 .save() 前必须先调用 .is_valid()。否则不能安全地访问 validated_data,也不能保存对象。验证错误通常被转换成 400 响应,字段错误以字段名为键;不属于某一个字段的错误通常放在 non_field_errors 下。(django-rest-framework.org)


2.4 三层验证:字段、对象和数据库

字段级验证

字段级验证只依赖一个字段:

class ArticleSerializer(serializers.ModelSerializer):
    class Meta:
        model = Article
        fields = ["id", "title", "body", "owner", "is_published", "created_at"]
        read_only_fields = ["id", "owner", "created_at"]

    def validate_title(self, value):
        value = value.strip()
        if len(value) < 3:
            raise serializers.ValidationError("标题至少需要 3 个字符。")
        return value

方法名必须是 validate_<field_name>。它接收已经经过字段基础转换的值,例如 IntegerField 的输入可能已经被转换成 Python int

对象级验证

对象级验证需要同时查看多个字段:

class ArticleSerializer(serializers.ModelSerializer):
    class Meta:
        model = Article
        fields = ["id", "title", "body", "owner", "is_published", "created_at"]
        read_only_fields = ["id", "owner", "created_at"]

    def validate(self, attrs):
        title = attrs.get("title", getattr(self.instance, "title", ""))
        body = attrs.get("body", getattr(self.instance, "body", ""))

        if title.strip().lower() == body.strip().lower():
            raise serializers.ValidationError(
                "标题不能与正文完全相同。"
            )
        return attrs

这里使用 getattr(self.instance, ...) 是因为 PATCH 是部分更新。部分更新时,attrs 只包含客户端实际提交的字段;如果验证逻辑直接访问 attrs["title"],当客户端只修改 body 时就会触发 KeyError。DRF 文档明确指出,部分更新的对象级验证必须考虑缺失字段,并从 self.instance 读取旧值。(django-rest-framework.org)

数据库约束

Serializer 验证不能替代数据库约束。例如“同一用户不能拥有两个相同标题的文章”属于跨请求并发下的唯一性约束,应该同时使用模型约束:

class Article(models.Model):
    # ...

    class Meta:
        constraints = [
            models.UniqueConstraint(
                fields=["owner", "title"],
                name="uniq_article_owner_title",
            )
        ]

原因是两个并发请求可能同时通过 Serializer 的“查询后判断”,随后一起插入。只有数据库的唯一约束才能在最终写入点阻止竞争条件。Serializer 负责友好的输入错误,数据库负责不可绕过的完整性约束。


2.5 .create().update()perform_create()

ModelSerializer 默认可以处理常规 Model 创建和更新,但服务端字段不应从客户端输入中获取:

class ArticleViewSet(viewsets.ModelViewSet):
    queryset = Article.objects.all()
    serializer_class = ArticleSerializer

    def perform_create(self, serializer):
        serializer.save(owner=self.request.user)

serializer.save(owner=...) 传入的关键字参数会并入 validated_data,然后交给 Serializer 的 create()update()。(django-rest-framework.org)

在这个例子中:

POST /api/articles/
{
  "title": "文章",
  "body": "正文",
  "owner": 999
}

即使客户端提交了 owner,由于 Serializer 将其声明为只读字段,服务端仍然应该以 request.user 为准。真正的所有者关系由服务器的认证上下文决定,而不是由客户端声明决定。

如果需要复杂的写入逻辑,可以覆盖:

class ArticleSerializer(serializers.ModelSerializer):
    class Meta:
        model = Article
        fields = ["id", "title", "body", "owner", "is_published", "created_at"]
        read_only_fields = ["id", "owner", "created_at"]

    def create(self, validated_data):
        return Article.objects.create(**validated_data)

    def update(self, instance, validated_data):
        instance.title = validated_data.get("title", instance.title)
        instance.body = validated_data.get("body", instance.body)
        instance.is_published = validated_data.get(
            "is_published",
            instance.is_published,
        )
        instance.save(update_fields=["title", "body", "is_published"])
        return instance

但不要为了“看起来完整”而覆盖默认行为。自定义 create()update() 的价值在于处理默认映射无法表达的逻辑,例如嵌套对象、跨表事务或领域规则。


2.6 读写 Serializer 不一定应该相同

创建和读取通常不是同一份契约。一个常见做法是为写入和读取分别定义 Serializer:

class ArticleWriteSerializer(serializers.ModelSerializer):
    class Meta:
        model = Article
        fields = ["title", "body", "is_published"]


class ArticleReadSerializer(serializers.ModelSerializer):
    owner = serializers.CharField(source="owner.username", read_only=True)

    class Meta:
        model = Article
        fields = [
            "id",
            "title",
            "body",
            "owner",
            "is_published",
            "created_at",
        ]
        read_only_fields = fields

在 ViewSet 中根据 action 选择:

def get_serializer_class(self):
    if self.action in {"create", "update", "partial_update"}:
        return ArticleWriteSerializer
    return ArticleReadSerializer

这不是强制规范,而是对 API 契约的显式建模:

  • 写入字段少,避免客户端提交只读或内部字段;
  • 读取字段可以包含计算值、关联信息和审计字段;
  • 未来改变返回结构时,不必破坏输入接口。

三、ViewSet:把资源操作组合到一个控制器中

3.1 ViewSet 与普通 View 的差异

普通 Django View 通常编写:

def get(request):
    ...


def post(request):
    ...

DRF 的 ViewSet 不直接定义 .get().post(),而是定义资源动作:

list()
create()
retrieve()
update()
partial_update()
destroy()

在 URL 最终绑定时,HTTP 方法才映射到这些 action。Router 可以根据 ViewSet 自动生成 URL。(django-rest-framework.org)

默认的 ModelViewSet 由以下行为组合而成:

CreateModelMixin       -> create
RetrieveModelMixin     -> retrieve
UpdateModelMixin       -> update / partial_update
DestroyModelMixin      -> destroy
ListModelMixin         -> list
GenericViewSet         -> 通用查询、Serializer、权限、分页等能力

典型代码:

from rest_framework import viewsets

class ArticleViewSet(viewsets.ModelViewSet):
    queryset = Article.objects.all()
    serializer_class = ArticleSerializer

这会提供常见资源操作:

HTTP 方法 URL action
GET /api/articles/ list
POST /api/articles/ create
GET /api/articles/{pk}/ retrieve
PUT /api/articles/{pk}/ update
PATCH /api/articles/{pk}/ partial_update
DELETE /api/articles/{pk}/ destroy

ViewSet 的收益是集中复用 queryset、Serializer 和策略配置;代价是 URL 和行为不如手写 Django View 那样显式。对需要严格控制每条路由、每种方法或非资源型操作的 API,APIView、Generic Views 或普通 Django View 可能更合适。


3.2 Router 如何生成 URL

from django.urls import include, path
from rest_framework.routers import DefaultRouter

router = DefaultRouter()
router.register("articles", ArticleViewSet, basename="article")

urlpatterns = [
    path("api/", include(router.urls)),
]

注册前缀不要写尾部斜杠:

router.register("articles", ArticleViewSet)   # 正确
router.register("articles/", ArticleViewSet)  # 不推荐

Router 会根据自身配置决定是否追加尾部斜杠。(django-rest-framework.org)

如果 ViewSet 没有 queryset 属性,而是只通过 get_queryset() 提供查询集,Router 可能无法推断 basename,这时应显式传入:

router.register(
    "articles",
    ArticleViewSet,
    basename="article",
)

3.3 self.action 的使用时机

ViewSet 在分发过程中会设置:

self.action
self.detail
self.basename
self.suffix

例如:

def get_permissions(self):
    if self.action == "list":
        permission_classes = [IsAuthenticated]
    elif self.action == "destroy":
        permission_classes = [IsAdminUser]
    else:
        permission_classes = [IsAuthenticated]

    return [permission() for permission in permission_classes]

但是,self.action 并不是所有生命周期方法中都可用。DRF 文档特别指出,在 get_parsers()get_authenticators()get_content_negotiator() 中访问 self.action 可能导致 AttributeError,因为这些方法执行时 action 尚未设置。(django-rest-framework.org)


3.4 自定义 action:资源动作之外的操作

对于“发布文章”这样的非标准 CRUD 操作,可以使用 @action

from rest_framework.decorators import action
from rest_framework.response import Response

class ArticleViewSet(viewsets.ModelViewSet):
    queryset = Article.objects.all()
    serializer_class = ArticleSerializer

    @action(
        detail=True,
        methods=["post"],
        url_path="publish",
    )
    def publish(self, request, pk=None):
        article = self.get_object()
        article.is_published = True
        article.save(update_fields=["is_published"])
        return Response({"status": "published"})

Router 会生成:

POST /api/articles/1/publish/

detail=True 表示该操作针对单个对象,因此 URL 中包含 pkdetail=False 表示针对整个集合,例如:

@action(detail=False, methods=["get"])
def statistics(self, request):
    ...

对应:

GET /api/articles/statistics/

@action 还可以覆盖自己的权限、限流和 Serializer:

@action(
    detail=True,
    methods=["post"],
    permission_classes=[IsAdminUser],
    throttle_classes=[UserRateThrottle],
)
def publish(self, request, pk=None):
    ...

DRF 支持通过 @action 为额外动作配置路由、HTTP 方法以及策略类。不要使用绕过 Router 的 .as_view() 来绑定带 @action 的 ViewSet,因为这样可能跳过 action 的相关配置。(django-rest-framework.org)


四、权限:认证成功不等于允许操作

4.1 认证、权限和对象所有权

假设请求头中有合法 Token:

Authorization: Token abc123

认证器可能得到:

request.user = alice
request.auth = token_object

这只能说明请求者被识别为 Alice。是否可以读取某篇文章,还要经过权限判断。

DRF 内置权限包括:

  • AllowAny:任何请求都允许;
  • IsAuthenticated:必须认证;
  • IsAdminUser:要求 user.is_staff is True
  • IsAuthenticatedOrReadOnly:匿名用户只能使用安全方法;
  • DjangoModelPermissions:使用 Django 模型权限;
  • DjangoObjectPermissions:支持对象级权限。(django-rest-framework.org)

安全方法通常指 GETHEADOPTIONS。例如:

from rest_framework.permissions import IsAuthenticatedOrReadOnly

class ArticleViewSet(viewsets.ModelViewSet):
    permission_classes = [IsAuthenticatedOrReadOnly]

这意味着匿名用户可以读取,但不能创建、修改或删除。

如果没有配置全局权限,DRF 默认权限是 AllowAny。生产 API 不应该因为“忘记配置”而意外开放,因此通常会在 settings 中明确设置:

REST_FRAMEWORK = {
    "DEFAULT_PERMISSION_CLASSES": [
        "rest_framework.permissions.IsAuthenticated",
    ],
}

单个 ViewSet 上设置的 permission_classes 会替换默认列表,而不是与默认列表自动合并。(django-rest-framework.org)


4.2 权限列表的逻辑关系

默认权限列表的语义是:

permission_classes = [A, B, C]

当所有权限都通过时,请求才允许继续,因此它相当于:

A AND B AND C

需要 OR 关系时,可以使用权限类组合:

from rest_framework.permissions import (
    BasePermission,
    IsAuthenticated,
    SAFE_METHODS,
)


class ReadOnly(BasePermission):
    def has_permission(self, request, view):
        return request.method in SAFE_METHODS


class ArticleViewSet(viewsets.ModelViewSet):
    permission_classes = [IsAuthenticated | ReadOnly]

权限类可以使用 &|~ 组合,并遵循类似 Python 逻辑运算符的优先级;复杂表达式应使用括号明确意图。(django-rest-framework.org)


4.3 自定义对象级权限

“只有文章作者可以修改自己的文章”是对象级权限:

from rest_framework.permissions import BasePermission, SAFE_METHODS


class IsOwnerOrReadOnly(BasePermission):
    def has_object_permission(self, request, view, obj):
        if request.method in SAFE_METHODS:
            return True
        return obj.owner_id == request.user.id

ViewSet:

class ArticleViewSet(viewsets.ModelViewSet):
    serializer_class = ArticleSerializer
    permission_classes = [IsAuthenticated, IsOwnerOrReadOnly]

    def get_queryset(self):
        return Article.objects.select_related("owner").all()

对于 Generic Views 和 ModelViewSet,调用 get_object() 时会执行对象级权限检查。如果手动覆盖 get_object(),必须显式调用:

self.check_object_permissions(self.request, obj)

否则你可能写出了“有权限类但实际没有检查”的代码。(django-rest-framework.org)


4.4 列表接口不会逐个执行对象级权限

这是一个非常容易产生安全漏洞的边界。

对:

GET /api/articles/

DRF 不会为了每个返回对象逐一调用 has_object_permission()。原因是逐个检查会增加查询和计算成本。列表权限必须通过 QuerySet 本身过滤:

class ArticleViewSet(viewsets.ModelViewSet):
    permission_classes = [IsAuthenticated, IsOwnerOrReadOnly]
    serializer_class = ArticleSerializer

    def get_queryset(self):
        user = self.request.user
        return Article.objects.filter(owner=user)

如果业务规则是“公开文章所有人可见,私有文章仅作者可见”:

from django.db.models import Q

def get_queryset(self):
    user = self.request.user
    return Article.objects.filter(
        Q(is_published=True) | Q(owner=user)
    )

对象级权限解决单个对象的操作判断;QuerySet 过滤解决列表中“哪些对象根本应该出现”。创建对象时也不会调用对象级权限,因为对象尚未通过 get_object() 得到;创建限制应该放在 Serializer、perform_create() 或视图级权限中。(django-rest-framework.org)


4.5 401 和 403 的区别

权限失败可能返回 401 或 403,不是简单地由“有没有登录”决定:

  • 已认证但没有权限:403;
  • 未认证,且最高优先级认证类不使用 WWW-Authenticate:403;
  • 未认证,且最高优先级认证类使用 WWW-Authenticate:401,并附带该响应头。

因此,看到 403 不一定说明用户已经登录;看到 401 也不意味着所有认证方式都统一如此。最终状态码取决于认证类和权限失败类型。(django-rest-framework.org)


五、分页:限制响应规模,还要保持结果稳定

5.1 为什么不能直接返回整个 QuerySet

下面的代码在数据量小的时候可用:

return Response(ArticleSerializer(
    Article.objects.all(),
    many=True,
).data)

数据量增加后,它可能同时造成:

  • 数据库读取大量行;
  • Python 创建大量 Model 实例;
  • Serializer 生成大量字典;
  • JSON 响应变大;
  • 网络传输时间增长;
  • 客户端内存和解析时间增加。

分页将集合拆成有限大小的窗口。DRF 内置分页只会自动应用于 Generic Views 和 ViewSets;如果使用普通 APIView,必须手动调用分页 API。全局分页默认关闭,需要同时配置分页类和页大小。(django-rest-framework.org)


5.2 PageNumberPagination

页码分页适合后台管理、用户可以跳转到第 N 页的场景:

from rest_framework.pagination import PageNumberPagination


class ArticlePagination(PageNumberPagination):
    page_size = 20
    page_size_query_param = "page_size"
    max_page_size = 100

ViewSet:

class ArticleViewSet(viewsets.ModelViewSet):
    queryset = Article.objects.all().order_by("-created_at", "-id")
    serializer_class = ArticleSerializer
    pagination_class = ArticlePagination

请求:

GET /api/articles/?page=2&page_size=20

典型响应:

{
  "count": 135,
  "next": "https://example.com/api/articles/?page=3&page_size=20",
  "previous": "https://example.com/api/articles/?page=1&page_size=20",
  "results": [
    {
      "id": 21,
      "title": "..."
    }
  ]
}

page_size_query_param 允许客户端请求不同页大小,但必须配合 max_page_size,否则客户端可以请求极大的页面。DRF 的 PageNumberPagination 默认使用 page 参数,也可以通过子类修改参数名。(django-rest-framework.org)


5.3 LimitOffsetPagination

Limit/offset 分页更接近数据库查询:

GET /api/articles/?limit=20&offset=40

其含义是:

limit  = 最多返回多少条
offset = 跳过前多少条

例如总数为 135、limit=20offset=40 时,返回第 41 至 60 条记录。这个方案适合客户端自己维护偏移量,或者需要与已有数据查询接口保持一致的场景。

但 offset 越大,数据库通常越需要跳过更多记录。对于高增长数据集,深分页可能变慢,不能仅靠 DRF 配置解决。


5.4 CursorPagination:用位置代替页码

游标分页返回一个不透明 cursor:

GET /api/articles/?cursor=cD0yMDI2...

客户端不应该解析 cursor 的内部内容,而应该把它原样交给下一次请求。游标分页只支持向前或向后翻页,不支持任意跳到第 100 页,但它适合时间线、消息流和不断增长的数据集。

from rest_framework.pagination import CursorPagination


class ArticleCursorPagination(CursorPagination):
    page_size = 20
    ordering = "-created_at"

游标分页要求排序依据稳定。排序字段应满足:

  1. 创建后不再变化;
  2. 唯一或接近唯一;
  3. 不为空;
  4. 不是浮点数;
  5. 有数据库索引。

DRF 文档还特别提醒,使用非唯一字段或可变字段作为游标排序依据可能导致重复、遗漏或不稳定结果。(django-rest-framework.org)

一个更稳妥的排序是:

queryset = Article.objects.order_by("-created_at", "-id")

直觉上,分页依赖一个稳定的全序关系。若两条记录的 created_at 相同,id 作为第二排序键可以打破平局。若排序字段会被更新,用户翻页期间记录可能从当前页移动到另一页,从而产生重复或遗漏。


5.5 分页与并发写入

分页不是数据库快照。考虑如下时序:

t1: 客户端读取第 1 页,得到 A、B、C
t2: 新记录 X 插入,并排在 A 之前
t3: 客户端读取第 2 页

如果使用 offset 分页,第 2 页的 offset 仍然从旧位置计算,可能再次读到 C,或者跳过某条记录。

CursorPagination 通过保存排序位置降低了这种风险,但它仍然不是跨多个请求的事务快照。游标只能保证在稳定排序和合理查询条件下更适合连续遍历,不能保证整个 API 遍历期间数据绝对不变。


六、限流:暂时拒绝,不是安全边界

6.1 限流与权限的区别

权限通常表达稳定关系:

Alice 是否可以删除 Article 1?

限流表达时间窗口内的临时状态:

Alice 在当前一分钟内是否已经请求了太多次?

DRF 会在 View 主体运行前检查每一个 throttle;只要一个检查失败,就抛出 Throttled,业务方法不会执行。(django-rest-framework.org)

常用配置:

REST_FRAMEWORK = {
    "DEFAULT_THROTTLE_CLASSES": [
        "rest_framework.throttling.AnonRateThrottle",
        "rest_framework.throttling.UserRateThrottle",
    ],
    "DEFAULT_THROTTLE_RATES": {
        "anon": "30/min",
        "user": "300/min",
    },
}

限流速率可以使用秒、分钟、小时或天,例如:

10/second
60/min
1000/day

多个限流器可以叠加:

class BurstRateThrottle(UserRateThrottle):
    scope = "burst"


class SustainedRateThrottle(UserRateThrottle):
    scope = "sustained"
REST_FRAMEWORK = {
    "DEFAULT_THROTTLE_CLASSES": [
        "myapp.throttles.BurstRateThrottle",
        "myapp.throttles.SustainedRateThrottle",
    ],
    "DEFAULT_THROTTLE_RATES": {
        "burst": "60/min",
        "sustained": "1000/day",
    },
}

这样一个用户同时受到短期突发限制和长期累计限制。UserRateThrottle 对已认证用户通常按用户 ID 生成 key;匿名请求则回退到 IP 地址。(django-rest-framework.org)


6.2 ScopedRateThrottle:按接口类型限流

不同接口的资源成本通常不同:

from rest_framework.throttling import ScopedRateThrottle


class ArticleViewSet(viewsets.ModelViewSet):
    throttle_classes = [ScopedRateThrottle]
    throttle_scope = "articles"
class ReportViewSet(viewsets.ViewSet):
    throttle_classes = [ScopedRateThrottle]
    throttle_scope = "reports"
REST_FRAMEWORK = {
    "DEFAULT_THROTTLE_CLASSES": [
        "rest_framework.throttling.ScopedRateThrottle",
    ],
    "DEFAULT_THROTTLE_RATES": {
        "articles": "300/min",
        "reports": "10/min",
    },
}

同一 scope 下的接口共享额度;昂贵的报表接口可以单独降低速率。@action 也可以覆盖 ViewSet 的 throttle 配置。(django-rest-framework.org)


6.3 限流依赖缓存,默认实现不是严格计数器

DRF 内置限流使用 Django cache。开发环境中的 LocMemCache 可以工作,但多进程、多实例部署时,每个进程的内存缓存彼此独立,不能形成全局限流状态。生产环境通常需要共享缓存后端。

此外,DRF 文档明确说明,内置实现使用非原子操作判断请求速率,因此并发下可能出现一定程度的计数模糊。它适合业务配额和基本过载保护,不应作为抵御暴力破解或 DDoS 的唯一安全措施。恶意客户端可以伪造来源 IP,应用层限流也可能在请求已经抵达应用后才生效。(django-rest-framework.org)

因此,生产系统通常需要分层保护:

CDN / WAF
    ↓
反向代理或 API Gateway 限流
    ↓
应用层 DRF throttle
    ↓
数据库和业务级配额

应用层限流解决业务语义,例如“普通用户每天最多创建 100 篇文章”;网关层限流解决连接、IP、请求体大小和异常流量。


6.4 代理环境中的 IP 识别

DRF 会参考 X-Forwarded-For 和 WSGI 的 REMOTE_ADDR 判断客户端 IP。若部署在反向代理后面,必须正确配置可信代理数量,否则可能出现:

  • 所有请求被识别为代理服务器;
  • 客户端伪造 X-Forwarded-For 绕过限流;
  • 多个用户经过 NAT 后被当成一个客户端。

NUM_PROXIES 用来说明应用前面有多少个可信代理。配置错误时,匿名用户限流结果会不稳定;该设置不能脱离真实的网络拓扑。(django-rest-framework.org)


七、一个完整的 Article API

下面组合 Serializer、ViewSet、权限、分页和限流。

7.1 模型

# articles/models.py
from django.conf import settings
from django.db import models


class Article(models.Model):
    owner = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        on_delete=models.CASCADE,
        related_name="articles",
    )
    title = models.CharField(max_length=200)
    body = models.TextField()
    is_published = models.BooleanField(default=False)
    created_at = models.DateTimeField(auto_now_add=True, db_index=True)

    class Meta:
        constraints = [
            models.UniqueConstraint(
                fields=["owner", "title"],
                name="uniq_article_owner_title",
            )
        ]
        ordering = ["-created_at", "-id"]

    def __str__(self):
        return self.title

执行:

python manage.py makemigrations
python manage.py migrate

created_at 建立索引并参与排序,是为了让列表分页具有稳定且可查询的顺序。模型的 Meta.ordering 可以提供默认排序,但复杂 API 仍建议在 get_queryset() 中明确写出关键排序。


7.2 Serializer

# articles/serializers.py
from rest_framework import serializers

from .models import Article


class ArticleReadSerializer(serializers.ModelSerializer):
    owner = serializers.CharField(
        source="owner.username",
        read_only=True,
    )

    class Meta:
        model = Article
        fields = [
            "id",
            "title",
            "body",
            "owner",
            "is_published",
            "created_at",
        ]
        read_only_fields = fields


class ArticleWriteSerializer(serializers.ModelSerializer):
    class Meta:
        model = Article
        fields = ["title", "body", "is_published"]

    def validate_title(self, value):
        value = value.strip()
        if len(value) < 3:
            raise serializers.ValidationError(
                "标题至少需要 3 个字符。"
            )
        return value

    def validate(self, attrs):
        title = attrs.get(
            "title",
            getattr(self.instance, "title", ""),
        )
        body = attrs.get(
            "body",
            getattr(self.instance, "body", ""),
        )

        if title.strip().lower() == body.strip().lower():
            raise serializers.ValidationError(
                "标题不能与正文完全相同。"
            )
        return attrs

读取和写入使用不同 Serializer,可以避免把 owner 当成客户端可修改字段,同时保证读取时能输出用户名。


7.3 权限

# articles/permissions.py
from rest_framework.permissions import BasePermission, SAFE_METHODS


class IsOwnerOrReadOnly(BasePermission):
    def has_object_permission(self, request, view, obj):
        if request.method in SAFE_METHODS:
            return True
        return obj.owner_id == request.user.id

这里的权限含义是:

读取:只要已经通过 View 级权限即可
写入:必须是对象 owner

但列表仍然需要 QuerySet 过滤,否则其他用户的文章可能出现在列表中。


7.4 分页

# articles/pagination.py
from rest_framework.pagination import PageNumberPagination


class ArticlePagination(PageNumberPagination):
    page_size = 20
    page_size_query_param = "page_size"
    max_page_size = 100

7.5 限流

# articles/throttles.py
from rest_framework.throttling import UserRateThrottle


class ArticleBurstThrottle(UserRateThrottle):
    scope = "article_burst"

7.6 ViewSet

# articles/views.py
from django.db.models import Q
from rest_framework import status, viewsets
from rest_framework.decorators import action
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response

from .models import Article
from .pagination import ArticlePagination
from .permissions import IsOwnerOrReadOnly
from .serializers import (
    ArticleReadSerializer,
    ArticleWriteSerializer,
)
from .throttles import ArticleBurstThrottle


class ArticleViewSet(viewsets.ModelViewSet):
    pagination_class = ArticlePagination
    throttle_classes = [ArticleBurstThrottle]
    permission_classes = [IsAuthenticated, IsOwnerOrReadOnly]

    def get_queryset(self):
        user = self.request.user

        return (
            Article.objects
            .select_related("owner")
            .filter(Q(is_published=True) | Q(owner=user))
            .order_by("-created_at", "-id")
        )

    def get_serializer_class(self):
        if self.action in {
            "create",
            "update",
            "partial_update",
        }:
            return ArticleWriteSerializer
        return ArticleReadSerializer

    def perform_create(self, serializer):
        serializer.save(owner=self.request.user)

    @action(detail=True, methods=["post"])
    def publish(self, request, pk=None):
        article = self.get_object()

        article.is_published = True
        article.save(update_fields=["is_published"])

        return Response(
            {"status": "published"},
            status=status.HTTP_200_OK,
        )

这段代码的关键因果关系如下:

  • get_queryset() 先过滤可见数据,所以列表不会返回无权查看的文章;
  • get_object() 获取详情时,会进一步执行 IsOwnerOrReadOnly 的对象级权限;
  • perform_create()request.user 设置 owner;
  • get_serializer_class() 让读取和写入使用不同契约;
  • pagination_classlist 自动分页;
  • throttle_classeslistcreate 和自定义 action 进入方法前进行限流;
  • publish 使用 self.get_object(),所以同样受查询集和对象级权限保护。

7.7 URL 路由

# project/urls.py
from django.contrib import admin
from django.urls import include, path
from rest_framework.routers import DefaultRouter

from articles.views import ArticleViewSet

router = DefaultRouter()
router.register("articles", ArticleViewSet, basename="article")

urlpatterns = [
    path("admin/", admin.site.urls),
    path("api/", include(router.urls)),
]

启动:

python manage.py runserver

创建文章:

curl -X POST http://127.0.0.1:8000/api/articles/ \
  -H "Content-Type: application/json" \
  -H "Authorization: Token <token>" \
  -d '{
    "title": "Django REST Framework",
    "body": "学习 Serializer 和 ViewSet",
    "is_published": false
  }'

成功时可能得到:

HTTP/1.1 201 Created
{
  "id": 1,
  "title": "Django REST Framework",
  "body": "学习 Serializer 和 ViewSet",
  "owner": "alice",
  "is_published": false,
  "created_at": "2026-09-01T10:00:00Z"
}

注意,写入 Serializer 没有 owner 字段,但读取 Serializer 有 owner 字段。这正是读写契约分离的效果。

验证失败:

curl -X POST http://127.0.0.1:8000/api/articles/ \
  -H "Content-Type: application/json" \
  -H "Authorization: Token <token>" \
  -d '{
    "title": "a",
    "body": "内容",
    "is_published": false
  }'

响应:

HTTP/1.1 400 Bad Request
{
  "title": [
    "标题至少需要 3 个字符。"
  ]
}

未认证请求:

HTTP/1.1 401 Unauthorized

或者:

HTTP/1.1 403 Forbidden

具体返回哪一个,取决于认证类是否提供 WWW-Authenticate。(django-rest-framework.org)

限流失败:

HTTP/1.1 429 Too Many Requests

如果 throttle 实现能够计算等待时间,响应还可能包含 Retry-After。(django-rest-framework.org)

分页请求:

curl \
  -H "Authorization: Token <token>" \
  "http://127.0.0.1:8000/api/articles/?page=2&page_size=20"

预期结构:

{
  "count": 37,
  "next": null,
  "previous": "http://127.0.0.1:8000/api/articles/?page=1&page_size=20",
  "results": [
    {
      "id": 17,
      "title": "..."
    }
  ]
}

八、统一错误处理与异常边界

DRF 默认会将许多异常转换为带状态码和内容类型的响应:

异常 常见状态码
ValidationError 400
ParseError 400
NotAuthenticated 401 或 403
AuthenticationFailed 401 或 403
PermissionDenied 403
NotFound 404
MethodNotAllowed 405
UnsupportedMediaType 415
Throttled 429

普通错误通常包含 detail 字段,Serializer 验证错误则按字段组织。(django-rest-framework.org)

可以在保留 DRF 默认行为的基础上添加统一字段:

# project/api_exceptions.py
from rest_framework.views import exception_handler


def custom_exception_handler(exc, context):
    response = exception_handler(exc, context)

    if response is not None:
        response.data = {
            "status": response.status_code,
            "error": response.data,
        }

    return response

配置:

REST_FRAMEWORK = {
    "EXCEPTION_HANDLER": (
        "project.api_exceptions.custom_exception_handler"
    ),
}

但这个处理器只处理“以异常形式抛出”的错误。若 View 直接返回:

return Response(
    {"title": ["标题无效"]},
    status=400,
)

它不会经过异常处理器。若项目要求所有错误都使用相同结构,就应统一采用异常抛出方式,或者在代码审查中明确哪些响应属于业务主动返回。(django-rest-framework.org)


九、常见失败模式与诊断顺序

9.1 Serializer 没有调用 is_valid()

错误表现:

serializer.validated_data

触发异常,或者:

serializer.save()

失败。

诊断原则是检查请求处理顺序:

serializer = Serializer(data=request.data)
serializer.is_valid(raise_exception=True)
serializer.save()

不能把 serializer.data 当作输入验证结果。serializer.data 面向输出,serializer.validated_data 面向验证成功后的输入。


9.2 PATCH 的验证逻辑假定所有字段都存在

错误代码:

def validate(self, attrs):
    if attrs["start"] >= attrs["finish"]:
        ...

当 PATCH 只提交 finish 时,start 不在 attrs 中。正确做法是把新值和旧实例值合并后再验证。


9.3 只写对象权限,不过滤列表 QuerySet

错误代码:

class ArticleViewSet(viewsets.ModelViewSet):
    permission_classes = [IsOwnerOrReadOnly]
    queryset = Article.objects.all()

这可能让用户在列表中看到其他人的私有文章,因为列表不会逐对象调用 has_object_permission()。必须在 get_queryset() 中按可见性过滤。


9.4 用可变字段做 CursorPagination 排序

错误代码:

class BadCursorPagination(CursorPagination):
    ordering = "title"

如果标题可以修改,记录在用户翻页过程中会改变位置。排序键应优先使用创建时间、不可变 slug 或稳定 ID,并为平局提供第二排序键。


9.5 把 DRF 限流当成 DDoS 防护

DRF 限流运行在应用层,并且依赖缓存和非原子计数。它可以控制普通客户端和业务套餐额度,却不能替代 WAF、网关、连接级限制、请求体大小限制和密码登录保护。官方文档也明确将它定位为基本过度使用保护和业务策略,而非安全防护。(django-rest-framework.org)


十、如何选择这些组件

可以用以下判断来减少无意义的抽象:

  • Serializer:需要定义 API 输入、输出、字段验证或跨字段验证时使用;
  • ModelSerializer:数据主要来自 Django Model,且默认字段映射足够时使用;
  • APIView:单个接口的行为明显不是标准资源 CRUD,或需要完全显式控制生命周期时使用;
  • ViewSet:一组相关资源动作可以统一组织,并希望使用 Router 时使用;
  • 权限类:表达“谁可以访问哪个接口或对象”;
  • QuerySet 过滤:表达“列表中哪些对象可以被看见”;
  • 分页:所有可能增长的集合接口都应明确设计;
  • PageNumberPagination:需要页码和总数,适合后台和管理界面;
  • LimitOffsetPagination:客户端需要偏移查询语义;
  • CursorPagination:连续遍历时间线,并且能提供稳定排序;
  • DRF Throttle:实现应用层业务额度和基本过载控制;
  • 网关/WAF 限流:处理恶意流量、IP 级别、连接级别和大规模突发流量。

这几个组件共同构成了 DRF API 的核心边界:Serializer 控制数据形状,ViewSet 组织资源动作,权限控制授权关系,分页控制集合规模,限流控制时间窗口内的使用频率。真正可靠的 API 不是把它们全部配置一遍,而是让每个组件承担它能够保证的那一部分职责,并把并发、列表可见性、数据库约束和异常路径一起纳入设计。


系列导航与关联阅读

官方资料

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理;正文、示例与工程清单由 WR BLOG 编写。