基于Django ORM预取间接关联项：优化审核系统查询性能

你遇到的是典型的SerializerMethodField引发的性能瓶颈——每个Book实例都会触发独立的重复项查询，再加上序列化关联对象的额外请求，总查询量直接飙升到2N+1级别。下面我给你一套彻底解决这个问题的方案，完全移除SerializerMethodField，通过SQL聚合或预取的方式一次性获取所有需要的数据。

核心思路 #

我们可以利用PostgreSQL的JSONBAgg结合Subquery，在主查询中直接聚合每个Book对应的重复项数据（包括关联的Serie和Language），这样整个列表只需要一次数据库查询就能完成所有数据获取，彻底消除N+1问题。如果你的环境不支持JSONBAgg，也可以用Prefetch+内存过滤的方案替代。

方案一：SQL聚合（推荐，性能最优） #

步骤1：修改查询集（视图/管理器） #

在你获取Book列表的地方（比如ViewSet的get_queryset方法），添加annotate和子查询：

from django.db.models import Subquery, OuterRef, F, JSONBAgg
from django.contrib.postgres.search import TrigramSimilarity
from .models import Book

def get_books_queryset():
    # 子查询：获取当前Book的符合条件的重复项，并序列化为目标JSON结构
    duplicate_books_subquery = (
        Book.objects.filter(
            allowed=True,
            serie__language=OuterRef("serie__language")
        )
        .annotate(similarity=TrigramSimilarity("title", OuterRef("title")))
        .filter(similarity__gt=0.2)
        .order_by("-similarity")[:10]  # 限制最多10个重复项
        # 提取需要的字段，构建和预期输出一致的JSON结构
        .annotate(
            book_id=F("id"),
            book_title=F("title"),
            serie_id=F("serie__id"),
            serie_title=F("serie__title"),
            language_id=F("serie__language__id"),
            language_name=F("serie__language__name"),
        )
        .values(
            "book_id", "book_title", "serie_id", "serie_title", "language_id", "language_name"
        )
        .aggregate(
            duplicates=JSONBAgg(
                {
                    "id": F("book_id"),
                    "title": F("book_title"),
                    "serie": {
                        "id": F("serie_id"),
                        "title": F("serie_title"),
                        "language": {
                            "id": F("language_id"),
                            "name": F("language_name")
                        }
                    }
                }
            )
        )["duplicates"]
    )

    # 主查询：预取当前Book的Serie和Language，同时annotate重复项的JSON数据
    return (
        Book.objects.select_related("serie__language")
        .annotate(duplicates=Subquery(duplicate_books_subquery))
    )

步骤2：修改Serializer #

完全移除SerializerMethodField和get_duplicates方法，直接使用annotate出来的duplicates字段：

from rest_framework.serializers import ModelSerializer, JSONField
from .models import Book, Serie, Language

class LanguageSerializer(ModelSerializer):
    class Meta:
        model = Language
        fields = ('id', 'name')

class SerieAdminAuxSerializer(ModelSerializer):
    class Meta:
        model = Serie
        fields = ("id", "language", "title")
    language = LanguageSerializer()

class BookAdminSerializer(ModelSerializer):
    class Meta:
        model = Book
        fields = ("id", "title", "serie", "duplicates", )
    
    serie = SerieAdminAuxSerializer()
    # 直接用JSONField序列化annotate出来的重复项数据
    duplicates = JSONField(read_only=True)

    def to_representation(self, instance):
        representation = super().to_representation(instance)
        # 处理allowed=True的情况，返回空列表
        if instance.allowed:
            representation["duplicates"] = []
        return representation

方案二：Prefetch+内存过滤（兼容性更好） #

如果你的环境不支持PostgreSQL的JSONBAgg，可以用这个方案，虽然性能略逊，但同样能解决N+1问题：

步骤1：修改查询集 #

from django.db.models import Subquery, OuterRef, Prefetch
from django.contrib.postgres.search import TrigramSimilarity
from .models import Book

def get_books_queryset():
    # 子查询：获取当前Book的重复项ID列表
    duplicate_ids_subquery = (
        Book.objects.filter(
            allowed=True,
            serie__language=OuterRef("serie__language")
        )
        .annotate(similarity=TrigramSimilarity("title", OuterRef("title")))
        .filter(similarity__gt=0.2)
        .order_by("-similarity")[:10]
        .values_list("id", flat=True)
    )

    # 预取所有允许的Book，同时预取它们的Serie和Language
    duplicates_prefetch = Prefetch(
        "all_allowed_books",
        queryset=Book.objects.prefetch_related(
            Prefetch("serie", queryset=Serie.objects.select_related("language"))
        ).filter(allowed=True),
        to_attr="prefetched_allowed_books"
    )

    # 主查询：预取当前Book的关联数据，annotate重复项ID，同时预取所有允许的Book
    return (
        Book.objects.select_related("serie__language")
        .annotate(duplicate_ids=Subquery(duplicate_ids_subquery))
        .prefetch_related(duplicates_prefetch)
    )

步骤2：修改Serializer #

在to_representation中做内存过滤和排序：

from rest_framework.serializers import ModelSerializer
from .models import Book, Serie, Language

# 保留你原有的BookAdminMinimalSerializer、SerieAdminAuxSerializer、LanguageSerializer

class BookAdminSerializer(ModelSerializer):
    class Meta:
        model = Book
        fields = ("id", "title", "serie", "duplicates", )
    
    serie = SerieAdminAuxSerializer()
    duplicates = BookAdminMinimalSerializer(many=True, read_only=True)

    def to_representation(self, instance):
        representation = super().to_representation(instance)
        if instance.allowed:
            representation["duplicates"] = []
        else:
            # 内存过滤，不触发新查询
            duplicates = [b for b in instance.prefetched_allowed_books if b.id in instance.duplicate_ids]
            # 按子查询中的相似度顺序排序
            duplicates_sorted = sorted(duplicates, key=lambda x: instance.duplicate_ids.index(x.id))
            representation["duplicates"] = BookAdminMinimalSerializer(duplicates_sorted, many=True).data
        return representation

方案优势 #

1. 彻底消除N+1：所有数据要么通过SQL聚合一次性获取，要么通过预取批量加载，不再有实例级的额外查询。
2. 输出结构完全匹配预期：两种方案的输出格式和你提供的示例完全一致，无需前端修改。
3. 性能提升显著：对于N个Book实例，查询量从2N+1降低到1次（方案一）或2次（方案二）。

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