03-API和序列化器(Serializers)

Tomy
41 分钟阅读
61 次浏览
API 开发指南:如何配合服务层使用序列化器,以及在复杂业务下如何保持 API 的简洁性。
DjangoDRFAPI设计

Django Styleguide - API 和序列化器(APIs & Serializers)


🌐 核心基石:RESTful 设计哲学

在深入具体的代码实现之前,我们必须首先理解 API 设计的“灵魂”——REST (Representational State Transfer)

1. 核心思想:资源 (Resource) 与 动词 (Verbs) 的分离

REST 的精髓在于将一切视为 “资源”

  • 资源 (名词):应该是 URL 的一部分,例如 /users/, /courses/

  • 操作 (动词):通过 HTTP 方法来表达。不要在 URL 中写动词(错误示范:/create_user/)。

2. 标准 HTTP 动词映射表

在本指南中,我们严格遵循以下协议约定:

动词动作 (CRUD)含义响应码 (典型)
GETRead获取资源列表或单个资源。200 OK
POSTCreate创建新资源。201 Created
PUTUpdate完整替换一个现有资源。200 OK
PATCHUpdate部分修改一个现有资源(通常推荐这种,更灵活)。200 OK
DELETEDelete物理或逻辑删除一个资源。204 No Content


🏗️ 现代 API 层宣言:架构选择与准则

在本指南中,我们提倡一种“防御性”且“显式”的 API 设计模式。在深入具体代码实现之前,请务必建立以下心理模型:

1. 核心哲学:API 即契约 (API as a Contract)

我们提倡将 API 的输入/输出定义(序列化器/Schema)直接嵌套在 API 视图类内部(Inline Serializers)

  • 对齐现代标准:在 FastAPIDjango-Ninja 中,Schema 往往是为单个 API 定制的。直接在路由/视图中定义它们,能让代码结构极其清晰,且易于维护。

  • 拒绝数据泄露:全局复用的序列化器是数据泄露的温床。嵌套模式确保 API A 的变更绝不意外影响 API B。

  • 极致的上下文局部性:打开一个文件即可看到“参数验证 -> 业务调用 -> 响应返回”的全貌。这不仅降低了心智压力,也让 AI 辅助编程(如 Cursor) 的上下文感知力达到巅峰。

2. 架构选择:内联 (Inline) vs 共享 (Shared)?

很多开发者纠结:如果每个 API 都写自己的 Schema,代码不就重复了吗?我们的观点是:接口契约的“独立演进”优先级高于“DRY (Don't Repeat Yourself)”原则。

  • 90% 的场景:内联定义。如果数据结构只服务于特定 API(如 UserRegisterInput),请直接嵌套。这保证了 API 的独立性,防止“牵一发而动全身”。

  • 10% 的场景:共享提取。只有当结构属于 “全系统公用标准”(如标准分页响应 PagedResponse、全局错误格式、跨核心业务传递的 DTO)时,才提取到单独的 serializers.py

  • 应对方案:如果文件太长,请按功能模块拆分文件/目录(如 apis/auth.py),而不是拆分序列化器。增加文件垂直长度比增加水平耦合安全得多。

3. 创建 API 的通用规则

  • 单一职责:一个 API 只做一个操作(CRUD 分离为 4 个接口)。

  • 继承最简单的 APIView:显式调用逻辑。拒绝 ModelViewSet 这种黑盒魔法。

  • 不要在 API 中编写业务逻辑:API 仅负责协议转换(解析参数、调用 Service、格式化输出)。

  • 性能进阶:追求极致性能或类型安全时,优先考虑 Pydantic

💡 标准 API 模式示例

python
class UserCreateApi(APIView):
    # 1. 输入契约
    class InputSerializer(serializers.Serializer):
        email = serializers.EmailField()
        password = serializers.CharField(write_only=True)

    # 2. 输出契约
    class OutputSerializer(serializers.Serializer):
        id = serializers.IntegerField()
        email = serializers.EmailField()

    def post(self, request):
        # 3. 解析与验证 (Parsing)
        serializer = self.InputSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        # 4. 业务协调 (Orchestration)
        user = user_create_service(**serializer.validated_data)

        # 5. 返回响应 (Responding)
        return Response(self.OutputSerializer(user).data)

🛠️ 在 Django 中如何实现? (从显式到隐式)

在 Django REST Framework (DRF) 中,实现 RESTful 协议有三个阶段的演进,每个阶段的“透明度”和“开发效率”都不同。

第一阶段:基于函数的视图 (Function-based Views - FBV)

这是最原始、最直观的实现方式。你直接操作 HTTP 请求对象。

  • 实现方式:使用 @api_view 装饰器。

  • 代码属性

    python
    @api_view(['GET', 'POST'])
    def course_api(request):
        if request.method == 'GET':
            # 显式处理获取逻辑
            return Response(...)
        elif request.method == 'POST':
            # 显式处理创建逻辑
            return Response(...)
    
  • 评价:非常适合极简的逻辑,但当接口复杂(需要处理 5 个动词)时,代码会嵌套大量的 if/else,变得难以维护。

第二阶段:基于类的视图 (Class-based Views - CBV) —— 本指南推荐

这是将“组织性”引入 API 的关键一步。它利用 Python 类的特性,将不同的动词映射到不同的方法中。

  • 实现方式:继承 APIView

  • 代码属性

    python
    class CourseApi(APIView):
        def get(self, request): ...    # 获取逻辑
        def post(self, request): ...   # 创建逻辑
        def delete(self, request): ... # 删除逻辑
    
  • 评价最平衡的选择。逻辑按动词天然隔离,非常干净。更重要的是,它依然是显式的——每一行逻辑都写在方法里,没有隐藏。

第三阶段:高阶抽象(Generics & ViewSets)—— 即“魔法”

这是 DRF 最引以为傲的特性,我们称之为“魔法”。

  • 实现方式:继承 ListCreateAPIViewModelViewSet

  • 什么是魔法?

    python
    class CourseViewSet(ModelViewSet):
        queryset = Course.objects.all()
        serializer_class = CourseSerializer
        # 消失的方法:你看不到 get(), post(), patch(),它们被藏在了父类的 Mixins 里。
    
  • 含义:它将“代码”转化为了“配置”。你只需要告诉框架数据库源和序列化器,它就自动为你生成全部 CRUD。

  • 魔法的优缺点

    • 优势:极速开发。如果你只需要标准的 CRUD 且不涉及 Service 层,它的效率无与伦比。

    • 劣势(本指南拒绝它的理由):它是隐式的。当你需要插入复杂的 Service 逻辑、进行多表聚合或自定义权限校验时,你必须去 Hook 那些深层的 perform_create 等钩子函数。魔术在后期维护时会变成高昂的心智利息。


🎨 深度博弈:为什么我们选择“显式” APIView?

在 DRF 的世界里,开发者经常会被 ModelViewSet 那种“三行代码搞定 CRUD”的快感所吸引。然而,在本指南中,我们坚持回归最基础的 APIView。这不仅仅是偏好,更是基于以下三个深层维度的平衡:

1. 哲学博弈:显式优于隐式 (Explicit is better than Implicit)

这是 Python 之禅的核心,也是本指南的灵魂。

  • 魔法模式 (ViewSet):将逻辑转化为配置。当你继承 ModelViewSet 时,数百行的逻辑(从 URL 路由到 DB 查询)被隐密地注入。当业务逻辑变得复杂时,你必须去“Hook”那些深层的钩子(如 perform_create),这其实是在与框架暗战。

  • 显式模式 (APIView):每一个 HTTP 动词(GET, POST, PATCH)都有一个明确的方法对应。你可以清楚地看到参数从哪儿进来、调用了哪个 Service、返回了什么数据。

2. 架构选择:组合优于继承 (Composition over Inheritance)

  • 继承地狱ModelViewSet 依赖极其深重的继承链。这种结构极其脆弱——如果你想改变其中一个微小的行为,往往需要覆盖多个 Mixin 的方法,这增加了理解成本和测试难度。

  • 显式注入:使用 APIView,我们提倡像“插件”一样显式注入逻辑。你需要什么功能(异常处理、限流、日志),就通过 Mixins 或辅助函数显式调用。这种基于组合的逻辑不仅更容易测试,由于不依赖父类的复杂状态,其稳定性要高出数倍。

3. 现实考量:Service 层协作与 AI 赋能

  • HackSoft 的洞察:现实项目中的业务几乎从来不是“纯粹的 CRUD”。一旦涉及到多表聚合、第三方 API 调用或复杂的缓存策略,框架提供的“配置”就会变成一种累赘。APIView 提供了最纯粹的容器,让它与 Service 层的配合天衣无缝。

  • AI 时代的样板代码成本:过去,人们选择 ViewSet 是为了避开样板代码(Boilerplate)。但在 AI 辅助编程(如 Cursor) 普及的今天,写 10 行 APIView 的样板代码成本和写 3 行配置的成本几乎持平。AI 能够瞬间生成清晰的结构,我们不再需要为了省那几行代码而去背负“黑盒”带来的心智负担。


📊 最终价值对比

维度显式模式 (APIView)魔法模式 (ViewSet)
可读性“所见即所得”。新人能快速看懂逻辑。“猜猜我在哪”。需要精通框架源码。
Service 集成原生支持。就是几个普通的 Python 调用。摩擦力极大。需要打破框架的封装逻辑。
维护成本长期稳定。逻辑是流动的,不是被封死的。容易随需求复杂度增加而产生“代码坏味道”。
测试便利性极高。可以轻松 mock 掉内部调用。困难。通常被迫进行重型的集成测试。
AI 亲和力完美。AI 对显式逻辑的补全和改错比黑盒更准。一般。AI 可能会写出不符合框架隐式规则的代码。

🧩 深度解析:Mixins 的力量

对我们来说,使用作为 API 视图的核心优势之一就是能够利用 Mixins

1. 什么是 Mixin?

Mixin 是 Python 多重继承的一种形式,它允许你将“插件式”的可重用功能注入到类中。它不是一个独立的视图,而是一组预定义的逻辑片段。

2. 为什么在类中使用它更好?

  • 防止“装饰器地狱”:在函数式视图中,你需要通过堆叠装饰器(Stacking Decorators)来添加功能。如果一个接口需要权限、缓存、异常处理和日志,你可能需要写 5 个装饰器。

  • 清晰的命名空间:类视图为 serializerspermissions 甚至是内联的辅助方法提供了一个自然的容器(命名空间),而函数式视图会将这些东西散落在全局作用域或堆缩在函数体内。

  • 可覆盖性:Mixin 提供的方法可以被子类轻松覆盖(Override),这比修改装饰器的行为要简单得多。

🔍 实战对比:装饰器地狱 vs Mixin 优雅组织

❌ 模式 A:函数式视图的“装饰器地狱”
python
# 很难看出逻辑的重心在哪里,且参数传递(如 user)变得混乱
@api_view(['POST'])
@permission_classes([IsAuthenticated])
@handle_api_errors  # 处理异常
@log_request       # 记录日志
@rate_limit(10)    # 限流
def create_course_api(request):
    # 所有的辅助逻辑都“飘”在函数外面,且很难针对特定接口修改装饰器内部的某个小行为
    ...
✅ 模式 B:类视图 + Mixins 的组织方式
python
# 1. 定义标准化的逻辑片段 (Mixin)
class ApiLoggingMixin:
    def log_action(self, message):
        print(f"Log: {message}") # 集中管理日志逻辑

# 2. 像搭积木一样组装 API
class CourseCreateApi(ApiErrorsMixin, ApiLoggingMixin, APIView):
    permission_classes = [IsAuthenticated]

    def post(self, request):
        self.log_action("Creating course") # 显式调用,逻辑清晰
        ...

    # 🚀 可覆盖性演示:如果这个接口想用特殊的异常处理?
    def handle_exception(self, exc):
        # 轻松重写 Mixin 或基类的方法,而不需要去动全局的“装饰器”逻辑
        if isinstance(exc, SpecialError):
            return Response({"custom": "logic"})
        return super().handle_exception(exc)

3. 常见的 Mixin 示例

在本指南的项目实践中,我们经常使用以下 Mixins:

  • ApiErrorsMixin最核心的 Mixin。它封装了异常处理逻辑,能够自动捕获业务层(Service/Selector)抛出的异常,并将其统一格式化为前端可读的 JSON 错误响应。

  • QueryParamsMixin:用于简化从 request.query_params 中提取并转换数据的逻辑。

  • PermissionMixin:虽然 DRF 有原生的权限系统,但有时我们需要更灵活的、基于方法的动态权限判定,此时自定义 Mixin 非常有用。


命名规范

对于我们的 API,我们使用以下命名约定:<Entity><Action>Api

🔖 命名模式:EntityActionApi

格式:{实体名}{动作}Api

常见动作:

  • List:列表

  • Detail:详情

  • Create:创建

  • Update:更新

  • Delete:删除

  • 自定义动作(如 Activate, Publish 等)

以下是一些示例:UserCreateApiUserSendResetPasswordApiUserDeactivateApi 等。

💡 完整的用户模块 API 示例

python
# users/apis.py

class UserListApi(APIView): pass           # GET /users/
class UserDetailApi(APIView): pass         # GET /users/{id}/
class UserCreateApi(APIView): pass         # POST /users/
class UserUpdateApi(APIView): pass         # PUT /users/{id}/
class UserDeleteApi(APIView): pass         # DELETE /users/{id}/
class UserActivateApi(APIView): pass       # POST /users/{id}/activate/
class UserDeactivateApi(APIView): pass     # POST /users/{id}/deactivate/
class UserResetPasswordApi(APIView): pass  # POST /users/{id}/reset-password/

基于类 vs 基于函数

💡 这主要取决于个人偏好,因为你可以用两种方法实现相同的结果。

我们有以下偏好:

  1. 默认选择基于类的 API / 视图。

  2. 如果其他人都更喜欢并且对函数感到舒适,使用基于函数的 API / 视图。

🔖 类 vs 函数的对比

特性基于类基于函数
继承✅ 容易❌ 需要装饰器
命名空间✅ 类提供❌ 需要模块
配置✅ 类属性❌ 装饰器参数
可读性相对复杂✅ 简单直接
代码量稍多✅ 较少

对我们来说,使用类作为 API / 视图的额外好处如下:

  1. 你可以继承 BaseApi 或添加 mixins。

    • 如果你使用基于函数的 API / 视图,你需要用装饰器做同样的事情。

  2. 类创建了一个命名空间,你可以在其中嵌套东西(属性、方法等)。

    • 额外的 API 配置可以通过类属性完成。

    • 在基于函数的 API / 视图的情况下,你需要堆叠装饰器。

这是一个继承 BaseApi 的类的示例:

python
class SomeApi(BaseApi):
    def get(self, request):
        data = something()

        return Response(data)

📝 BaseApi 示例

python
# common/apis.py
from rest_framework.views import APIView
from rest_framework.permissions import IsAuthenticated

class BaseApi(APIView):
    """所有 API 的基类"""
    permission_classes = [IsAuthenticated]

    def handle_exception(self, exc):
        """统一的异常处理"""
        # 自定义异常处理逻辑
        return super().handle_exception(exc)

这是一个使用 base_api 装饰器的函数示例(实现基于你的需求):

python
@base_api(["GET"])
def some_api(request):
    data = something()
    return Response(data)

💡 函数式 API 的装饰器实现

python
# common/decorators.py
from functools import wraps
from rest_framework.decorators import api_view, permission_classes
from rest_framework.permissions import IsAuthenticated

def base_api(methods):
    """基础 API 装饰器"""
    def decorator(func):
        @api_view(methods)
        @permission_classes([IsAuthenticated])
        @wraps(func)
        def wrapper(request, *args, **kwargs):
            return func(request, *args, **kwargs)
        return wrapper
    return decorator

# 使用
@base_api(['GET'])
def user_list(request):
    users = user_list_selector()
    return Response({'users': users})

查询列表 API(返回列表数据的 API)

简单版本

一个非常简单的列表 API 应该是这样的:

python
from rest_framework.views import APIView
from rest_framework import serializers
from rest_framework.response import Response

from styleguide_example.users.selectors import user_list
from styleguide_example.users.models import BaseUser


class UserListApi(APIView):
    class OutputSerializer(serializers.Serializer):
        id = serializers.CharField()
        email = serializers.CharField()

    def get(self, request):
        users = user_list()

        data = self.OutputSerializer(users, many=True).data

        return Response(data)

📝 代码详解

  1. OutputSerializer

    • 嵌套在 API 类中

    • 只声明需要返回的字段

    • 继承 Serializer(不是 ModelSerializer

  2. get 方法

    • 调用 selector 获取数据

    • 使用 serializer 序列化(many=True 表示列表)

    • 返回 JSON 响应

  3. 职责清晰

    • Selector 负责数据查询

    • Serializer 负责数据格式化

    • API 只是协调者

请记住,默认情况下,此 API 是公开的。身份验证由你决定。

⚠️ 添加身份验证

python
from rest_framework.permissions import IsAuthenticated

class UserListApi(APIView):
    permission_classes = [IsAuthenticated]  # 需要认证

    class OutputSerializer(serializers.Serializer):
        id = serializers.CharField()
        email = serializers.CharField()

    def get(self, request):
        # 只有认证用户可以访问
        users = user_list()
        data = self.OutputSerializer(users, many=True).data
        return Response(data)

过滤器 + 分页

乍一看,这很棘手,因为我们的 API 继承了 DRF 的普通 APIView,而过滤和分页被内置到通用 API 中:

  1. DRF Filtering

  2. DRF Pagination

🔖 DRF 的通用视图提供了什么

python
# DRF 的 ListAPIView 自动处理:
class SomeListView(ListAPIView):
    queryset = Model.objects.all()
    serializer_class = SomeSerializer
    filter_backends = [DjangoFilterBackend]
    pagination_class = PageNumberPagination
    # 自动过滤、分页、序列化

但我们选择不使用它,因为:

  • 隐藏了太多细节

  • 难以自定义

  • 与服务层模式不匹配

这就是为什么我们采取以下方法:

  1. Selectors 负责实际的过滤。

  2. API 负责过滤参数序列化。

  3. 如果你需要 DRF 提供的一些通用分页,API 应该负责。

  4. 如果你需要不同的分页,或者你自己实现它,要么添加一个新层来处理分页,要么让 selector 为你做。

💡 职责分配

查看Mermaid源码
Mermaid
graph TD
    subgraph API_IN[API 层 - 入口阶段]
        A1[1. 接收过滤参数] --> A2[2. 验证 FilterSerializer]
        A2 --> A3[3. 调用 Selector]
    end

    subgraph SELECTOR[Selector 层 - 核心大脑]
        S1[4. 接收验证后的参数] --> S2[5. 执行查询逻辑 / Q对象组装]
        S2 --> S3[6. 返回惰性 QuerySet]
    end

    subgraph API_OUT[API 层 - 出口阶段]
        A4[7. 应用分页器] --> A5[8. 返回格式化响应 / Output]
    end

    %% 跨层流动使用加粗线
    A3 ==> S1
    S3 ==> A4

    %% 样式美化:黑色字体 color:#000
    style SELECTOR fill:#f5f7ff,stroke:#5c7cfa,stroke-width:2px,color:#000
    style API_IN fill:#fff,stroke:#333,color:#000
    style API_OUT fill:#fff,stroke:#333,color:#000

⚠️ 核心注意事项:为什么 Selector 不管分页?

在这个流程中,有一个至关重要的技术细节:

  • Selector 只返回“查询计划”:它返回的是一个 Django 惰性 QuerySet。这意味着当 Selector 执行完毕时,并没有产生真正的数据库查询

  • API 层执行“切片”:分页逻辑(Limit/Offset)是在 API 层拿到 QuerySet 之后才叠加进去的。

  • 延迟触发(Lazy Execution):只有当 API 层应用了分页、并开始序列化数据的那一刻,Django 才会把“过滤条件”和“分页参数”合并成一条最终的 SQL 发送给数据库。

这种模式换来了极致的灵活性:Selector 保持了业务逻辑的纯粹性(不管你要看几页,逻辑都一样),而 API 层保留了对展示规模的精准控制。


💡 设计哲学:显式控制 vs 框架魔法

这种职责分配揭示了本指南的核心思想:“API 层管展示协议,Selector 层管业务边界。”

1. 过滤(Filtering)的本质:定义资源(What)

思想:过滤是业务逻辑的体现。它定义了结果集的本质内容

  • SQL 暗喻:对应 WHERE 子句。它决定了“我到底想要哪些数据”。

  • 业务场景:无论是同步 API、异步 Celery 任务还是管理脚本,“已过期用户”的定义始终是一致的。

  • DRF 魔法的问题:魔法模式通常通过 request 对象自动提取参数。这导致你的业务查询逻辑(过滤)悄悄地绑定在了 HTTP 协议上。如果你没有 request(如在后台任务中),你就无法复用这套逻辑。

2. 分页(Pagination)的本质:控制规模(How many)

思想:分页是展示逻辑的体现。它定义了结果集的呈现窗口

  • SQL 暗喻:对应 LIMIT / OFFSET 子句。它决定了“我一次想看多少数据”。

  • 展示属性:分页取决于消费方的“胃口”。App 屏幕小只要 5 条,网页由于表格大要 50 条。

  • 为什么留在 API?:分页并不改变数据的筛选属性,它只是为了传输的“经济性”。API 层最清楚现在的客户端环境,因此它是处理分页切片的最佳场所。

3. ⚠️ 魔法的代价:逻辑所有权的迷失

魔法 API(如 ModelViewSet)将两者作为并列的“插件”挂载在 View 上。这产生了两大危害:

  1. 隐式 Request 穿透:业务层(过滤)强依赖 HTTP 对象,导致逻辑无法在非 Web 环境复用。

  2. 责任混淆:开发者容易忘记“过滤”和“分页”的区别,将大量本该属于 Selector 的逻辑(甚至是权限控制)写在了 View 的配置或钩子函数里。

4. 本质区别总结

维度过滤 (Filtering)分页 (Pagination)
对应 SQLWHERE ... AND ...LIMIT ... OFFSET ...
决定因素业务规则 (Business Rules)客户端环境 (Consumer Environment)
稳定性极高(规则不常变)低(随设备改变)
逻辑归属Selector 层 (为了最大化复用)API 层 (为了精准控制交付)

🔍 代码实战对比:魔法 vs 显式

模式 A:DRF 传统“魔法”模式 (配置驱动)

python
# ❌ 逻辑被抽空,变成了纯配置。难以在不模拟 HTTP 的情况下复用。
class UserListApi(ListAPIView):
    queryset = User.objects.all()
    filter_backends = [DjangoFilterBackend]   # 过滤逻辑和http强制绑定了,无法复用
    filterset_fields = ['email', 'is_active']
    pagination_class = StandardPagination
    serializer_class = UserSerializer

模式 B:本指南推荐“显式”模式 (逻辑驱动)

1. Selector 层(只管“选苗子”,不管“切段儿”)

python
def user_list_selector(*, email=None, is_active=True):
    # 它只负责组装 QuerySet。注意:此时 SQL 还没发出去!
    qs = User.objects.filter(is_active=is_active) # 过滤逻辑由业务层控制
    if email:
        qs = qs.filter(email__icontains=email)
    return qs  # 返回一个“待执行”的查询计划

2. API 层(负责“切段儿”并包装)

python
class UserListApi(APIView):
    def get(self, request):
        # 显式获取并传入参数
        email = request.query_params.get('email')

        # 调用 Selector 拿到原始 QuerySet (Lazy)
        queryset = user_list_selector(email=email)

        # 核心点:分页器在这里介入。
        # 注意:DRF 的标准分页器通常会在这里执行 list(),从而触发真正的 SQL 查询(带有 LIMIT/OFFSET)。
        paginator = PageNumberPagination()
        page = paginator.paginate_queryset(queryset, request)

        # 序列化阶段:将查询到的 Model 实例转换为字典
        serializer = UserSerializer(page, many=True)
        return paginator.get_paginated_response(serializer.data)

🚀 进阶:使用 Pydantic 进行极速处理

如果你追求性能,这个 get 方法可以这样改写:

python
    def get(self, request):
        queryset = user_list_selector(email=request.query_params.get('email'))

        paginator = PageNumberPagination()
        page = paginator.paginate_queryset(queryset, request)

        # 使用 Pydantic 进行转换 (比 DRF 快 10 倍以上)
        # UserOut 是一个定义了 model_config = {"from_attributes": True} 的 BaseModel
        data = [UserOut.model_validate(user).model_dump() for user in page]

        return paginator.get_paginated_response(data)

💡 深度分析:为什么我们要如此“费劲”地接管过滤与分页?

乍一看,手动调用分页器和编写 FilterSerializer 似乎比直接继承 ListAPIView 麻烦得多。但这种做法换来了三个巨大的架构红利:

  1. 业务逻辑下沉(逻辑不等于 HTTP 请求) 在传统模式下,过滤逻辑绑定在 API 中。如果你有个 Celery 任务或 Management Command 想查询“所有逾期订单”,由于它们没有 HTTP Request 环境,你根本无法复用 API 里的过滤功能。而现在,逻辑被下沉到了 Selector,你可以随时随地在任何 Python 环境下调用它。

  2. 拒绝“黑盒魔法”,拥抱透明度 DRF 的 filter_backends 虽然方便,但其内部逻辑是隐藏的。如果你想实现一个复杂的、多表关联的自定义过滤,你往往需要去查阅框架源码,看如何 Hook 进去。但在我们的模式下,逻辑就在 Selector 里的 if 判断和 .filter() 调用中,白纸白字,清晰可见。

  3. 零性能损失的“惰性架构” 你可能会担心:在 Selector 里查一遍,回 API 再分页,会不会查两次?答案是不会。 感谢 Django 的 QuerySet惰性(Lazy) 执行的。Selector 返回的是一个 SQL 的定义。直到 API 层的分页器给它加上了 LIMITOFFSET,它才会真正去拍数据库。这意味着你在获得架构解耦的同时,性能与 DRF 的自动化方案完全一致。

  4. 对齐现代 API 设计趋势 (FastAPI Alignment) FastAPI 等现代框架之所以受到青睐,很大程度上是因为它们摒弃了“重型类继承”和“隐式黑盒逻辑”,强调显式的参数声明和直观的代码流。通过手动接管分页,你实际上是将 Django 的开发体验提升到了与现代框架同等的透明度水平。

  5. 防御“Request 穿透”污染 DRF 的高级抽象(如 DjangoFilterBackend)通常将整个 request 对象直接喂给过滤逻辑。这是一种严重的架构设计污染:它让本应纯粹的业务查询逻辑(Selector)间接依赖了 HTTP 环境。在本指南推荐的显式模式下,API 层充当了“隔离器”——它从 Request 中提取出纯粹的 Python 数据,再喂给 Selector。这使得 Selector 可以在没有 Request 的环境下(如定时任务、测试脚本)被完美复用。

让我们看一个依赖 DRF 提供的分页的示例:

python
from rest_framework.views import APIView
from rest_framework import serializers

from styleguide_example.api.mixins import ApiErrorsMixin
from styleguide_example.api.pagination import get_paginated_response, LimitOffsetPagination

from styleguide_example.users.selectors import user_list
from styleguide_example.users.models import BaseUser


class UserListApi(ApiErrorsMixin, APIView):
    class Pagination(LimitOffsetPagination):
        default_limit = 1

    class FilterSerializer(serializers.Serializer):
        id = serializers.IntegerField(required=False)
        # 重要:如果我们使用 BooleanField,它会默认为 False
        is_admin = serializers.NullBooleanField(required=False)
        email = serializers.EmailField(required=False)

    class OutputSerializer(serializers.Serializer):
        id = serializers.CharField()
        email = serializers.CharField()
        is_admin = serializers.BooleanField()

    def get(self, request):
        # 确保过滤器有效,如果传递了的话
        filters_serializer = self.FilterSerializer(data=request.query_params)
        filters_serializer.is_valid(raise_exception=True)

        users = user_list(filters=filters_serializer.validated_data)

        return get_paginated_response(
            pagination_class=self.Pagination,
            serializer_class=self.OutputSerializer,
            queryset=users,
            request=request,
            view=self
        )

📝 完整流程解析

  1. FilterSerializer

    python
    # 客户端请求:GET /users/?is_admin=true&email=test@example.com
    # FilterSerializer 验证这些参数
    filters_serializer.is_valid(raise_exception=True)
    # validated_data = {'is_admin': True, 'email': 'test@example.com'}
    
  2. 调用 Selector

    python
    users = user_list(filters=filters_serializer.validated_data)
    # Selector 使用这些过滤条件查询数据库
    
  3. 应用分页

    python
    # get_paginated_response 处理分页逻辑
    # 返回格式:
    # {
    #   "limit": 10,
    #   "offset": 0,
    #   "count": 100,
    #   "next": "...",
    #   "previous": null,
    #   "results": [...]
    # }
    

⚠️ 避坑指南:布尔字段的分层陷阱 (Three-State Logic)

在开发过滤接口时,布尔字段(Boolean)并非只有“真/假”,它其实代表了三个业务意图

  1. True:显式筛选“是”的数据(如:只看管理员)。

  2. False:显式筛选“否”的数据(如:只看普通用户)。

  3. None (不传):不筛选此项,看所有人

❌ 陷阱 A:序列化器误判
python
# 这种定义可能会在客户端不传参时默认补全为 False,或者在某些版本中强制解析为 False
is_admin = serializers.BooleanField(required=False)

结果:客户端原本想“看所有人”,被你由于默认值改成了“只看普通用户”,导致查询范围永远无法覆盖管理员。

✅ 方案 A:显式允许空值
python
# 现代 DRF 推荐做法
is_admin = serializers.BooleanField(required=False, allow_null=True)
# 或者旧版做法
is_admin = serializers.NullBooleanField(required=False)
❌ 陷阱 B:Selector 中的真值测试
python
def user_list(*, filters=None):
    ...
    # 危险!如果 is_admin 为 False,if 语句会判定为假,从而跳过过滤!
    if filters.get('is_admin'):
        queryset = queryset.filter(is_admin=filters['is_admin'])
✅ 方案 B:显式判断 None
python
def user_list(*, filters=None):
    ...
    # 明确区分“传了 False”和“根本没传”这两个状态
    is_admin = filters.get('is_admin')
    if is_admin is not None:
        queryset = queryset.filter(is_admin=is_admin)

当我们看这个 API 时,我们可以识别出几件事:

  1. 有一个 FilterSerializer,它将负责查询参数。如果我们不在这里做这件事,我们就必须在其他地方做,而 DRF 序列化器非常擅长这项工作。

  2. 我们将过滤器传递给 user_list selector。

  3. 我们使用 get_paginated_response 工具来返回...分页响应。

现在,让我们看看 selector:

python
import django_filters

from styleguide_example.users.models import BaseUser


class BaseUserFilter(django_filters.FilterSet):
    class Meta:
        model = BaseUser
        fields = ('id', 'email', 'is_admin')


def user_list(*, filters=None):
    filters = filters or {}

    qs = BaseUser.objects.all()

    return BaseUserFilter(filters, qs).qs

📝 django-filter 详解

django-filter 是一个强大的过滤库:

python
# 基本用法
class UserFilter(django_filters.FilterSet):
    # 自动为字段创建过滤器
    class Meta:
        model = User
        fields = ['email', 'is_admin']

# 高级用法
class UserFilter(django_filters.FilterSet):
    # 自定义过滤器
    email = django_filters.CharFilter(lookup_expr='icontains')
    created_after = django_filters.DateFilter(
        field_name='created_at',
        lookup_expr='gte'
    )
    price_range = django_filters.RangeFilter(field_name='price')

    class Meta:
        model = User
        fields = {
            'email': ['exact', 'icontains'],
            'is_active': ['exact'],
            'created_at': ['gte', 'lte'],
        }

如你所见,我们正在利用强大的 django-filter 库。

🚀 现代化演进:为什么 Pydantic 在过滤场景更香?

虽然 django-filter 在处理简单的字段映射时非常快,但越来越多的现代化团队(尤其是使用 FastAPIDjango-Ninja 的团队)开始转向使用 Pydantic 来定义过滤契约。

对比优势:

特性django-filterPydantic + 显式逻辑
类型提示弱。在 Selector 中处理的是字典。极强。整个参数集是一个类型化的对象。
透明度魔术。SQL 是自动生成的。透明。你可以一眼看出 if 条件是如何拼接 SQL 的。
复杂逻辑难。需要重写 Filter 类的底层方法。。就是普通的 Python 代码,AI 辅助极其精准。

示例:Pydantic 驱动的 Selector

python
class UserFilterSchema(pydantic.BaseModel):
    email: Optional[str] = None
    is_active: bool = True
    created_after: Optional[date] = None

def user_list(*, filters: UserFilterSchema) -> QuerySet[User]:
    qs = User.objects.all()

    # 享受极致的类型安全和代码自动补全
    if filters.email:
        qs = qs.filter(email__icontains=filters.email)

    if filters.created_after:
        qs = qs.filter(created_at__gte=filters.created_after)

    return qs

👀 总结: 如果你的过滤逻辑是标准的“字段 = 值”,django-filter 是捷径;但如果你追求系统的长期可维护性、类型安全以及对 AI 友好,那么 Pydantic + 显式过滤 是现代化的最终答案。

💡 技术内幕:为什么“惰性查询 (Lazy QuerySet)”是架构解耦的关键?

如果你来自其他语言(如 Node.js 或 Go),你可能会担心:在 Selector 里定义了查询,回到 API 再分页,难道不会先查询全量数据吗?

答案是:绝对不会。这就是 Django 架构设计的精妙之处。

  1. 定义不等于执行:当你调用 User.objects.filter() 时,Django 只是创建了一个 QuerySet 对象(类似于查询计划),并没有连接数据库

  2. 链式组合QuerySet 可以像积木一样叠加。Selector 叠加了“过滤逻辑”,API 层随后叠加了“分页切片”。

  3. 最后一刻触发:只有当你真正开始遍历数据(如在序列化器中循环)时,Django 才会将所有的积木(过滤 + 分页)合并成一条最终的 SQL(带有 LIMITOFFSET)发送给数据库。

结论:因为有了惰性特性,我们可以放心地在不同层级编写指令,而不用担心性能损失。这让“逻辑下沉(Selector)”和“规模控制(API)”的完美分离成为可能。

最后,让我们看看 get_paginated_response

python
from rest_framework.response import Response


def get_paginated_response(*, pagination_class, serializer_class, queryset, request, view):
    paginator = pagination_class()

    page = paginator.paginate_queryset(queryset, request, view=view)

    if page is not None:
        serializer = serializer_class(page, many=True)
        return paginator.get_paginated_response(serializer.data)

    serializer = serializer_class(queryset, many=True)

    return Response(data=serializer.data)

📝 分页逻辑说明

  1. 创建分页器实例

  2. 分页 QuerySetpaginate_queryset 返回当前页的数据

  3. 序列化当前页:只序列化当前页的数据

  4. 返回分页响应:包含 next/previous 等元数据

如果没有分页(或 QuerySet 很小),直接返回所有数据。

这基本上是从 DRF 内部提取的代码。

LimitOffsetPagination 也是如此:

python
from collections import OrderedDict

from rest_framework.pagination import LimitOffsetPagination as _LimitOffsetPagination
from rest_framework.response import Response


class LimitOffsetPagination(_LimitOffsetPagination):
    default_limit = 10
    max_limit = 50

    def get_paginated_data(self, data):
        return OrderedDict([
            ('limit', self.limit),
            ('offset', self.offset),
            ('count', self.count),
            ('next', self.get_next_link()),
            ('previous', self.get_previous_link()),
            ('results', data)
        ])

    def get_paginated_response(self, data):
        """
        我们重新定义此方法以返回 `limit` 和 `offset`。
        这被前端用于自己构建分页。
        """
        return Response(OrderedDict([
            ('limit', self.limit),
            ('offset', self.offset),
            ('count', self.count),
            ('next', self.get_next_link()),
            ('previous', self.get_previous_link()),
            ('results', data)
        ]))

💡 自定义分页响应格式

python
# 默认 DRF 分页响应:
{
    "count": 100,
    "next": "http://api.example.org/accounts/?offset=20&limit=10",
    "previous": "http://api.example.org/accounts/?offset=0&limit=10",
    "results": [...]
}

# 自定义后添加 limit 和 offset:
{
    "limit": 10,      # 添加
    "offset": 10,     # 添加
    "count": 100,
    "next": "...",
    "previous": "...",
    "results": [...]
}

这样前端可以更方便地构建自己的分页 UI。

我们基本上做的是逆向工程通用 API。

👀 再次强调,如果你需要其他分页方式,你可以始终实现它并以相同的方式使用它。 在某些情况下,selector 需要负责分页。我们以处理过滤的相同方式处理这些情况。

你可以在 Styleguide Example 项目中找到带有过滤器和分页的示例列表 API 的代码。

完整的过滤和分页的代码示例

python
# selectors.py
import django_filters
from django.db.models import Q

class ProductFilter(django_filters.FilterSet):
    search = django_filters.CharFilter(method='filter_search')
    min_price = django_filters.NumberFilter(field_name='price', lookup_expr='gte')
    max_price = django_filters.NumberFilter(field_name='price', lookup_expr='lte')

    def filter_search(self, queryset, name, value):
        return queryset.filter(
            Q(name__icontains=value) |
            Q(description__icontains=value)
        )

    class Meta:
        model = Product
        fields = {
            'category': ['exact'],
            'is_active': ['exact'],
            'created_at': ['gte', 'lte'],
        }

def product_list(*, filters=None):
    filters = filters or {}
    qs = Product.objects.select_related('category')
    return ProductFilter(filters, qs).qs

# apis.py
class ProductListApi(APIView):
    class Pagination(LimitOffsetPagination):
        default_limit = 20

    class FilterSerializer(serializers.Serializer):
        search = serializers.CharField(required=False)
        category = serializers.IntegerField(required=False)
        min_price = serializers.DecimalField(max_digits=10, decimal_places=2, required=False)
        max_price = serializers.DecimalField(max_digits=10, decimal_places=2, required=False)
        is_active = serializers.BooleanField(required=False)

    class OutputSerializer(serializers.Serializer):
        id = serializers.IntegerField()
        name = serializers.CharField()
        price = serializers.DecimalField(max_digits=10, decimal_places=2)
        category_name = serializers.CharField(source='category.name')

    def get(self, request):
        filters_serializer = self.FilterSerializer(data=request.query_params)
        filters_serializer.is_valid(raise_exception=True)

        products = product_list(filters=filters_serializer.validated_data)

        return get_paginated_response(
            pagination_class=self.Pagination,
            serializer_class=self.OutputSerializer,
            queryset=products,
            request=request,
            view=self
        )

查询单个资源(GET)API

这是一个示例:

python
class CourseDetailApi(SomeAuthenticationMixin, APIView):
    class OutputSerializer(serializers.Serializer):
        id = serializers.CharField()
        name = serializers.CharField()
        start_date = serializers.DateField()
        end_date = serializers.DateField()

    def get(self, request, course_id):
        course = course_get(id=course_id)

        serializer = self.OutputSerializer(course)

        return Response(serializer.data)

📝 查询单个资源 API 代码解析

  1. URL 参数

    python
    def get(self, request, course_id):
        # course_id 来自 URL:/courses/<int:course_id>/
    
  2. 获取对象

    python
    course = course_get(id=course_id)
    # 使用 selector 获取单个对象
    # 如果不存在会抛出异常
    
  3. 序列化

    python
    serializer = self.OutputSerializer(course)
    # 注意:没有 many=True(单个对象)
    

更完整的查询单个资源 API

python
from rest_framework.exceptions import NotFound

class OrderDetailApi(APIView):
    permission_classes = [IsAuthenticated]

    class OutputSerializer(serializers.Serializer):
        id = serializers.IntegerField()
        order_number = serializers.CharField()
        status = serializers.CharField()
        total = serializers.DecimalField(max_digits=10, decimal_places=2)
        created_at = serializers.DateTimeField()

        # 嵌套序列化器
        items = inline_serializer(many=True, fields={
            'id': serializers.IntegerField(),
            'product_name': serializers.CharField(source='product.name'),
            'quantity': serializers.IntegerField(),
            'price': serializers.DecimalField(max_digits=10, decimal_places=2),
        })

        shipping_address = inline_serializer(fields={
            'street': serializers.CharField(),
            'city': serializers.CharField(),
            'country': serializers.CharField(),
        })

    def get(self, request, order_id):
        # 权限检查:只能查看自己的订单
        order = order_get(id=order_id)

        if order.user != request.user:
            raise NotFound("订单不存在")

        serializer = self.OutputSerializer(order)
        return Response(serializer.data)

创建资源(POST)API

这是一个示例:

python
class CourseCreateApi(SomeAuthenticationMixin, APIView):
    # 输入序列化器,客户端需要提供的字段
    class InputSerializer(serializers.Serializer):
        name = serializers.CharField()
        start_date = serializers.DateField()
        end_date = serializers.DateField()

    def post(self, request):
        serializer = self.InputSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        course_create(**serializer.validated_data)

        return Response(status=status.HTTP_201_CREATED)

📝 创建 API 代码解析

  1. InputSerializer

    • 只包含客户端需要提供的字段

    • 不包含自动生成的字段(如 id, created_at)

  2. 验证

    python
    serializer.is_valid(raise_exception=True)
    # 如果验证失败,自动返回 400 错误
    
  3. 调用服务

    python
    course_create(**serializer.validated_data)
    # 使用 ** 解包字典作为关键字参数
    
  4. 返回状态

    python
    return Response(status=status.HTTP_201_CREATED)
    # 201:资源已创建
    

返回创建的对象

python
class CourseCreateApi(APIView):
    # 输入序列化器,客户端需要提供的字段
    class InputSerializer(serializers.Serializer):
        name = serializers.CharField()
        start_date = serializers.DateField()
        end_date = serializers.DateField()

    # 输出序列化器,API 返回的数据格式
    class OutputSerializer(serializers.Serializer):
        id = serializers.IntegerField()
        name = serializers.CharField()
        start_date = serializers.DateField()
        end_date = serializers.DateField()
        created_at = serializers.DateTimeField()
      # POST 请求处理
    def post(self, request):
        serializer = self.InputSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        course = course_create(**serializer.validated_data)

        # 返回创建的对象
        output = self.OutputSerializer(course)
        return Response(output.data, status=status.HTTP_201_CREATED)

💡 完整的创建 API 示例

python
from rest_framework.exceptions import ValidationError

class UserRegisterApi(APIView):
    class InputSerializer(serializers.Serializer):
        email = serializers.EmailField()
        password = serializers.CharField(min_length=8)
        password_confirm = serializers.CharField()
        first_name = serializers.CharField()
        last_name = serializers.CharField()

        def validate(self, data):
            """跨字段验证"""
            if data['password'] != data['password_confirm']:
                raise ValidationError("密码不匹配")
            return data

        def validate_email(self, value):
            """单字段验证"""
            if User.objects.filter(email=value).exists():
                raise ValidationError("邮箱已存在")
            return value

    class OutputSerializer(serializers.Serializer):
        id = serializers.IntegerField()
        email = serializers.EmailField()
        full_name = serializers.CharField()

    def post(self, request):
        serializer = self.InputSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        # 移除不需要传给服务的字段
        validated_data = serializer.validated_data
        validated_data.pop('password_confirm')

        user = user_register(**validated_data)

        output = self.OutputSerializer(user)
        return Response(output.data, status=status.HTTP_201_CREATED)

更新资源(PUT/PATCH)API

这是一个示例:

python
class CourseUpdateApi(SomeAuthenticationMixin, APIView):
    class InputSerializer(serializers.Serializer):
>         name = serializers.CharField(required=False)
        start_date = serializers.DateField(required=False)
        end_date = serializers.DateField(required=False)

    def patch(self, request, course_id):
        serializer = self.InputSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        course_update(course_id=course_id, **serializer.validated_data)

        return Response(status=status.HTTP_200_OK)

📝 更新 API 代码解析

  1. 字段可选

    python
    name = serializers.CharField(required=False)
    # 更新时,客户端可以只提供需要更新的字段
    
  2. 部分更新 vs 完整更新

    • PATCH:部分更新(推荐使用 required=False

    • PUT:完整更新(所有字段都需要)

  3. 动词统一

    • 既然在 InputSerializer 中使用了 required=False,表明这可能是一个局部更新,因此我们使用 patch 方法。

更完整的更新 API 实现

python
class CourseUpdateApi(APIView):
    permission_classes = [IsAuthenticated]

    class InputSerializer(serializers.Serializer):
        name = serializers.CharField(required=False)
        start_date = serializers.DateField(required=False)
        end_date = serializers.DateField(required=False)

        def validate(self, data):
            """确保至少提供一个字段"""
            if not data:
                raise ValidationError("请至少提供一个要更新的字段")
            return data

    class OutputSerializer(serializers.Serializer):
        id = serializers.IntegerField()
        name = serializers.CharField()
        start_date = serializers.DateField()
        end_date = serializers.DateField()
        updated_at = serializers.DateTimeField()

    def patch(self, request, course_id):  # 使用 PATCH
        serializer = self.InputSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        course = course_update(
            course_id=course_id,
            data=serializer.validated_data,
            updated_by=request.user
        )

        output = self.OutputSerializer(course)
        return Response(output.data)

💡 带权限检查的更新 API

python
from rest_framework.exceptions import PermissionDenied

class ArticleUpdateApi(APIView):
    class InputSerializer(serializers.Serializer):
        title = serializers.CharField(required=False)
        content = serializers.CharField(required=False)

    def patch(self, request, article_id):
        # 获取文章
        article = article_get(id=article_id)

        # 权限检查:只有作者可以更新
        if article.author != request.user:
            raise PermissionDenied("只有作者可以更新文章")

        serializer = self.InputSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        article = article_update(
            article=article,
            data=serializer.validated_data
        )

        return Response(status=status.HTTP_200_OK)

🛡️ 深度探讨:ID 到对象的映射——谁该负责获取对象?

当 API 接收到一个 object_id 时,架构设计上最经典的问题就是:“我们在哪一步把这个字符串 ID 变成真正的 Model 实例?”

这不仅是代码习惯问题,它直接影响了系统的性能控制、权限安全和业务复用性

三种模式的深度对比

模式实现位置核心特点适用场景
A. 序列化器驱动PrimaryKeyRelatedField隐式/自动。DRF 在校验阶段自动查询数据库。简单的外键关联。
B. API 视图驱动get_object_or_404显式/透明。对象在进入业务逻辑前就已就绪。对象级权限检查(Object Permission)
C. Service 驱动传递 id 到 Service解耦/通用。业务层自主决定查询逻辑。跨场景复用、高性能查询、异步任务

1. 模式 A:序列化器驱动 (PrimaryKeyRelatedField)

这是 DRF 文档中最推荐的“标准做法”。

  • 优势:极速实现。验证 ID 是否存在与获取实例合二为一。

  • 劣势

    • 隐式副作用:你在 is_valid() 时就触发了 SQL。

    • 复用性差:Service 层被迫接收一个 Instance 对象。如果你想在 Celery 脚本或管理脚本中调用 Service,你还得自己先去查库,非常繁琐。

2. 模式 B:API 视图驱动 (Explicit Fetching)

在 API 方法中利用 get_object_or_404 显式获取。

  • 优势安全第一。如果你需要执行 self.check_object_permissions(request, obj),你必须在 View 层先拿到对象。这是处理复杂动态权限的最佳场所。

  • 劣势:View 会变得稍微繁琐,逻辑不够纯粹。

3. 模式 C:Service 逻辑层驱动 (The Styleguide Way 🚀)

API 层只传 ID,由 Service 内部执行 objects.get()

  • 优势

    • 极致解耦:Service 的接口变为 (user_id: int)。这让它可以在 API、Celery、Scripts 甚至单测中被完美复用。

    • 掌控力:Service 可以自主决定是否加 select_for_update 锁,或者应用复杂的 select_related

  • 劣势:API 层的 404 反馈逻辑需要额外处理(通过特定的异常捕获)。


🎯 架构师的决策建议

在本指南中,我们不推崇“固定标准”,而是推崇 “基于意图的选择”

  1. 原则一:涉及到对象权限吗? 如果需要根据对象内容(如:作者、状态)来决定访问权限,请使用 模式 B (View 获取)。先拿对象,查完权限,再传给 Service。

  2. 原则二:考虑到跨场景复用吗? 如果这个 Service 会被定时任务或管理脚本调用,请使用 模式 C (Service 内部获取)。调用方只需提供 ID,Service 负责一切。

  3. 原则三:尽量避免模式 A。 除非是极为简单的内部记录关联,否则不要在 API 契约中使用 PrimaryKeyRelatedField。它让 API 契约和数据库模型耦合得太紧。

💡 最佳实践代码对比

python
# ❌ 不推荐:模式 A (Serializer 耦合)
class InputSerializer(serializers.Serializer):
    course = serializers.PrimaryKeyRelatedField(queryset=Course.objects.all())

# ✅ 推荐:模式 B 或 C (在API层显式调用)

def patch(self, request, course_id):
    # 模式 B:如果你要检查权限
    course = get_object_or_404(Course, id=course_id)
    self.check_object_permissions(request, course)

    # 模式 C:如果你追求 Service 的极致通用性
    course_update_service(course_id=course_id, **serializer.validated_data)

    service(course=course, ...)

# 方法 3:在 Service 中

def post(self, request, course_id):
service(course_id=course_id, ...)

# 在 service 中

def some_service(\*, course_id: int):
course = Course.objects.get(id=course_id)

完整的对象获取示例

python
# utils.py
from django.http import Http404
from django.shortcuts import get_object_or_404
from rest_framework.exceptions import NotFound

def get_object_or_raise(model_or_queryset, **kwargs):
    """获取对象,不存在则抛出 NotFound"""
    try:
        return get_object_or_404(model_or_queryset, **kwargs)
    except Http404:
        raise NotFound(f"{model_or_queryset.__name__} 不存在")

def get_object_or_none(model_or_queryset, **kwargs):
    """获取对象,不存在则返回 None"""
    try:
        return get_object_or_404(model_or_queryset, **kwargs)
    except Http404:
        return None

# 使用
class CourseDetailApi(APIView):
    def get(self, request, course_id):
        # 方式 1:自动抛出异常
        course = get_object_or_raise(Course, id=course_id)

        # 方式 2:手动处理
        course = get_object_or_none(Course, id=course_id)
        if course is None:
            return Response({"error": "课程不存在"}, status=404)

🧩 嵌套序列化器 (Nested Serializers)

在处理“一对多”或“一对一”的关联关系时,我们往往不希望只返回一个干巴巴的 ID,而是希望返回完整的结构化数据。这就是嵌套序列化器的作用。

1. 为什么需要嵌套? (扁平 vs 嵌套)

  • 扁平模式 (Flat):客户端拿到 category_id=5,还得再发一次请求查分类名。

  • 嵌套模式 (Nested):客户端一次请求拿到 category: {"id": 5, "name": "编程语言"}

我们的核心哲学是:API 应该对前端极度友好。 一次查询能解决的事情,绝不让前端发第二次请求。

2. 局部性原则:使用 inline_serializer

在传统的 Django 开发中,你会为每个嵌套结构定义一个单独的类。这会导致 serializers.py 迅速膨胀,且逻辑散落在各处。

按照本指南的 “上下文局部性 (Context Locality)” 原则,我们提倡使用 inline_serializer 在 API 类内部直接定义嵌套结构。

�️ 工具实现:inline_serializer

这是一个微型但威力无穷的工具,让你能“随手”定义契约,而无需跳出当前视图文件。

python
# 这种工具函数通常放在 common/utils.py 或 api/mixins.py 中
from rest_framework import serializers

def inline_serializer(*, fields, data=None, **kwargs):
    """
    在不显式定义 Class 的情况下,动态创建一个序列化器。
    """
    serializer_class = type(
        'InlineSerializer',
        (serializers.Serializer,),
        fields
    )

    if data is not None:
        return serializer_class(data=data, **kwargs)

    return serializer_class(**kwargs)

3. 如何使用?

OutputSerializer 中像定义普通字段一样定义嵌套结构:

python
class CourseDetailApi(APIView):
    class OutputSerializer(serializers.Serializer):
        id = serializers.IntegerField()
        name = serializers.CharField()

        # 🚀 优雅的内联嵌套
        category = inline_serializer(fields={
            'id': serializers.IntegerField(),
            'name': serializers.CharField(),
            'slug': serializers.SlugField(),
        })

        # 🚀 列表嵌套 (many=True)
        weeks = inline_serializer(many=True, fields={
            'id': serializers.IntegerField(),
            'number': serializers.IntegerField(),
        })

⚠️ 关键性能警示:N+1 问题

嵌套序列化器是 N+1 查询的重灾区。

  • 永远不要在没做优化的情况下嵌套字段。

  • 必须配合:在对应的 Selector 中使用 .select_related() (针对一对一/多对一) 或 .prefetch_related() (针对一对多/多对多)。

  • 测试验证:使用 django-debug-toolbar 观察数据库连接数。如果嵌套了 10 个子项却产生了 11 条 SQL,说明你的优化失败了。

嵌套序列化器的完整示例

python
class OrderDetailApi(APIView):
    class OutputSerializer(serializers.Serializer):
        id = serializers.IntegerField()
        order_number = serializers.CharField()

        # 嵌套用户信息
        user = inline_serializer(fields={
            'id': serializers.IntegerField(),
            'email': serializers.EmailField(),
            'full_name': serializers.CharField(),
        })

        # 嵌套订单项列表
        items = inline_serializer(many=True, fields={
            'id': serializers.IntegerField(),
            'product': inline_serializer(fields={
                'id': serializers.IntegerField(),
                'name': serializers.CharField(),
                'price': serializers.DecimalField(max_digits=10, decimal_places=2),
            }),
            'quantity': serializers.IntegerField(),
            'subtotal': serializers.DecimalField(max_digits=10, decimal_places=2),
        })

        # 嵌套地址信息
        shipping_address = inline_serializer(fields={
            'street': serializers.CharField(),
            'city': serializers.CharField(),
            'postal_code': serializers.CharField(),
            'country': serializers.CharField(),
        })

        total = serializers.DecimalField(max_digits=10, decimal_places=2)
        status = serializers.CharField()
        created_at = serializers.DateTimeField()

💡 何时使用嵌套序列化器 vs 单独定义

使用 inline_serializer:

  • ✅ 简单的嵌套结构

  • ✅ 只在一个地方使用

  • ✅ 字段较少

单独定义序列化器类:

  • ✅ 复杂的嵌套结构

  • ✅ 需要在多处重用

  • ✅ 需要自定义验证方法

  • ✅ 字段较多

python
# 单独定义(可重用)
class AddressSerializer(serializers.Serializer):
    street = serializers.CharField()
    city = serializers.CharField()
    postal_code = serializers.CharField()
    country = serializers.CharField()

class OrderOutputSerializer(serializers.Serializer):
    # 重用定义好的序列化器
    shipping_address = AddressSerializer()
    billing_address = AddressSerializer()

🚀 高级序列化:函数式数据映射 (Functional Data Mapping)

在大多数场景下,OutputSerializer 配合显式的 Selector 已经足够强大。但在处理 超大规模数据、极深嵌套、或是需要跨表聚合的动态响应(如社交媒体 Feed 流、复杂报表或权限驱动的差异化显示) 时,传统的声明式序列化器会显得捉襟见肘。

1. 为什么需要“高级序列化”?

当你的 API 遇到以下瓶颈时,就是该考虑这种模式的时候了:

  • 性能天花板:DRF 序列化器在大数据量(如一次性返回几百条复杂对象)下的库开销非常大。

  • Selector 的逻辑爆炸:为了优化 N+1,你可能在 Selector 里写了三四十行 select_relatedprefetch_related。这种为了“展示方式”而进行的查询优化,其实正在无意间污染业务逻辑的纯粹性。

  • 异构数据组合:API 的结果不是一个单纯的 Model 列表,而是从 5 个不同的数据源(Redis, DB, 第三方 API)拼出来的。

2. 核心架构:逻辑转移

这种模式的核心思想是:将“为了优化展示而进行的二次查询”从业务 Selector 中剥离出来,放入一个专门的序列化函数中。

  • Selector:只负责最基础的筛选(比如:获取当前用户最新 10 条帖子)。它甚至可以只返回一组 ID。

  • 序列化函数 (Serializer Function):它接收 Selector 的结果,负责进行最极致的 SQL 优化(批量抓取)、结果集重组、以及内存性能优化。

3. 实战范例:复杂的社交 Media Feed API

假设我们要实现一个 Feed API。每个 Item 不仅有内容,还有实时点赞数、当前用户是否已点赞、以及前三条热门评论。

python
# API 层
class UserFeedApi(APIView):
    def get(self, request):
        # 1. Selector 只管最简单的“抓取”逻辑,返回基本的 FeedItem 列表
        items = feed_get_latest_for_user(user=request.user)

        # 2. 调用专门的序列化函数进行“深加工”
        data = feed_serialize_with_stats(items, viewer=request.user)

        return Response(data)

# 序列化逻辑 (apis.py 或专门的 mappers.py)
def feed_serialize_with_stats(items: List[FeedItem], viewer: User):
    item_ids = [i.id for i in items]

    # 一次性批量抓取所有需要的详情,彻底干掉任何潜在的 N+1
    optimized_items = FeedItem.objects.select_related('author').filter(id__in=item_ids)

    # 批量计算动态状态(如:Redis 里的点赞数)
    all_likes_counts = redis_client.mget([f"likes:{id}" for id in item_ids])

    # 构建最终响应
    result = []
    for item, like_count in zip(optimized_items, all_likes_counts):
         result.append({
             "id": item.id,
             "content": item.content,
             "author": {"name": item.author.name},
             "stats": {"likes": int(like_count or 0)}
         })
    return result

📝 复杂序列化流程解析

  1. 重新获取数据

    python
    # 使用 select_related 和 prefetch_related 优化
    objects = FeedItem.objects.select_related(...).prefetch_related(...)
    
  2. 构建缓存

    python
    # 一次性获取所有需要的额外数据
    some_cache = get_some_cache(feed_ids)
    
  3. 添加计算字段

    python
    # 为每个对象添加额外的计算字段
    feed_item._calculated_field = some_cache.get(feed_item.id)
    

4. 选择指南

场景推荐方案理由
标准 CRUDOutputSerializer + Selector开发效率高,结构标准化。
轻量关联 (1-2层)inline_serializer + select_related保持 Context Locality。
高性能/高复杂度 Feed函数式映射 (高级序列化)极致的 SQL 优化,完全掌控内存性能,解耦业务逻辑。

💡 架构师总结:这种模式其实是把序列化从“配置”回归到了“代码”。在 AI 辅助编程的今天,直接写 Python 字典构成的函数往往比去研究 DRF 复杂的嵌套规则更直观、更好测试。


📝 本章总结

构建高效、可维护的 API 并非为了追求极致简洁的代码量,而是为了追求极致的逻辑透明度

核心法则

  1. API 即契约 (API as a Contract)

    • API 层是业务系统的“守门员”,负责将不可信的外部输入(Request)翻译为可信的业务指令,并将复杂的内部数据包装为清晰的响应(Response)。

    • 坚持 Context Locality (上下文局部性):90% 的序列化器应内联(Inline)定义在 API 类中,让开发者一眼看透契约全貌。

  2. 显式优于隐式 (Explicit > Implicit)

    • 拒绝魔法:优先使用 APIView 而非 ViewSet。虽然代码略多,但 AI 环境下成本近乎为零,换来的是 100% 的掌控力和 Service 集成度。

    • 组合优于继承:通过显式的 Mixins 或依赖注入来扩展 API 功能,而非背负沉重的框架继承链。

  3. 精准的数据交互

    • 三态布尔陷阱:在过滤接口中,始终使用 allow_null=True 并通过 is not None 来显式区分“是、否、不筛选”三个逻辑意图。

    • ID 映射决策

      • 涉及权限决策?在 API View 获取对象。

      • 追求极致复用?在 Service 内部通过 ID 获取对象。

  4. 性能与扩展

    • N+1 防线:嵌套序列化器只定义结构,N+1 优化的责任永远在 Selector (Select/Prefetch) 层。

    • 性能天花板:当面临 Feed 流或大数据量聚合时,果断跳过声明式 Serializer,回归 函数式映射 (Advanced Serialization) 模式。

命名军规

  • API 类<Entity><Action>Api(如:UserRegisterApi)。

  • 动词匹配:严格遵循 GET (读), POST (写), PATCH (部分改), DELETE (删) 的 REST 语义。


下一步建议:掌握了 API 层的防御性设计后,请进入 04_URLs 与项目配置,学习如何将这些“契约点”优雅地接入系统。