DRF中如何通过Filter后端实现序列化后自定义字段过滤

你不需要持久化动态计算的字段值，两种可落地方案按需选择即可：

方案1：数据库层自定义过滤（生产环境推荐，性能最优） #

django-filter 原生支持自定义过滤方法，你只需要把序列化方法里的分支判断，反向翻译成对应的ORM查询条件，直接在数据库层完成筛选，不需要拉取全量数据到内存，适配大数据量场景。
实现代码示例：

import django_filters
from django.db.models import Q, OuterRef, Subquery
from django.utils import timezone
from .models import Document, DocumentParticipant
from .constants import (
    DocumentSigningStatus,
    DocumentParticipationType,
    DocumentSignerSigningStatus
)

class DocumentFilter(django_filters.FilterSet):
    signer_signing_status = django_filters.CharFilter(method="filter_by_signing_status")

    class Meta:
        model = Document
        fields = ["signer_signing_status"]

    def filter_by_signing_status(self, queryset, name, value):
        user = self.request.user
        now = timezone.now()

        # 已签署分支
        if value == DocumentSigningStatus.SIGNED:
            return queryset.filter(document_signing_status=DocumentSigningStatus.SIGNED)
        
        # 已过期分支：未签署 + 签署截止时间早于当前时间
        if value == "EXPIRED":
            return queryset.filter(
                ~Q(document_signing_status=DocumentSigningStatus.SIGNED),
                document_signing_date__end_date__lt=now
            )
        
        # 基础过滤：排除已签署、已过期的记录
        base_qs = queryset.filter(
            ~Q(document_signing_status=DocumentSigningStatus.SIGNED)
        ).filter(
            Q(document_signing_date__isnull=True) | Q(document_signing_date__end_date__gte=now)
        )

        # 预取当前用户在所有文档里的参与记录，避免重复查询
        user_participants = DocumentParticipant.objects.filter(
            owner=user,
            document__in=base_qs
        ).values("document_id", "order_of_signing", "participation_type", "signer_signing_status")
        user_doc_map = {p["document_id"]: p for p in user_participants}

        # 待我签署分支
        if value == "WAITING_FOR_ME":
            wait_me_order_ids = []
            wait_me_no_order_ids = []
            for doc_id, p in user_doc_map.items():
                doc = Document.objects.get(id=doc_id)
                if doc.is_with_signers_order:
                    if p["order_of_signing"] == doc.current_signing_order_number:
                        wait_me_order_ids.append(doc_id)
                else:
                    if p["participation_type"] == DocumentParticipationType.SIGNER \
                        and p["signer_signing_status"] != DocumentSignerSigningStatus.SIGNED:
                        wait_me_no_order_ids.append(doc_id)
            return base_qs.filter(id__in=wait_me_order_ids + wait_me_no_order_ids)
        
        # 待他人签署（非顺序签场景）
        if value == "WAITING_FOR_OTHERS":
            wait_other_ids = []
            for doc_id, p in user_doc_map.items():
                doc = Document.objects.get(id=doc_id)
                if not doc.is_with_signers_order:
                    if p["participation_type"] == DocumentParticipationType.VIEWER \
                        or (p["participation_type"] == DocumentParticipationType.SIGNER \
                            and p["signer_signing_status"] == DocumentSignerSigningStatus.SIGNED):
                        wait_other_ids.append(doc_id)
            return base_qs.filter(id__in=wait_other_ids)
        
        # 顺序签待他人签署分支
        if value == "WAITING_FOR_OTHERS_IN_ORDER":
            wait_other_order_ids = []
            for doc_id, p in user_doc_map.items():
                doc = Document.objects.get(id=doc_id)
                if doc.is_with_signers_order and p["order_of_signing"] != doc.current_signing_order_number:
                    wait_other_order_ids.append(doc_id)
            return base_qs.filter(id__in=wait_other_order_ids)
        
        # 传入非法过滤值时返回原查询集
        return queryset

• 优化提示：给document_signing_date.end_date、DocumentParticipant.owner、DocumentParticipant.document字段加数据库索引，可以进一步提升查询速度。

方案2：内存级过滤（小数据量场景适用） #

如果你的业务中Document单用户关联的数据量常年低于1000条，可以直接遍历查询集计算动态值后过滤，实现逻辑更简单，不需要维护复杂的ORM查询条件。
实现代码示例：

# 在对应视图类的get_queryset方法中添加过滤逻辑
def get_queryset(self):
    qs = super().get_queryset()
    filter_status = self.request.query_params.get("signer_signing_status")
    if not filter_status:
        return qs
    
    # 必须预取关联数据，避免N+1查询
    qs = qs.select_related("document_signing_date").prefetch_related("participants")
    serializer = self.get_serializer(context={"request": self.request})
    
    match_doc_ids = []
    for doc in qs:
        current_status = serializer.get_signer_signing_status(doc)
        if current_status == filter_status:
            match_doc_ids.append(doc.id)
    
    return qs.filter(id__in=match_doc_ids)

注意：该方案会把所有符合基础条件的Document记录拉到服务端内存计算，数据量超过1000条后接口响应速度会明显下降，高并发、大数据量场景禁止使用。

内容的提问来源于stack exchange，提问作者Viktor Vostrikov