如何正确合并Django查询集？合并查询集与单个对象报错

问题场景 #

我在Django API端点中实现了以下逻辑：根据URL参数获取指定buildNumber范围的笔记，同时强制返回最新的笔记（即使指定范围无数据），但运行时报错：

AttributeError at /api/v2/notes/Test/Test/Test/1/2/
'Notes' object has no attribute '_known_related_objects'

原代码：

def get(self, request, branchName, stack, resultType, buildNumberMIN, buildNumberMAX, format=None):
    try:
        # use ONE query to pull all data
        relevant_notes = Notes.objects.filter(
            branchName=branchName, stack=stack, resultType=resultType)

        # filter on the endpoint parameters (range of numbers)
        requested_range = relevant_notes.filter(
            buildNumber__gte=buildNumberMIN, buildNumber__lte=buildNumberMAX)
        
        # also pull latest note -- regardless of requested range
        latest_note = relevant_notes.latest('buildNumber')

        # join the notes
        return_set = requested_range | latest_note
        #serialize the data
        serializer = serializers.NoteSerializerWithJiraData(
            return_set, many=True)

        return Response(serializer.data)

    except Notes.DoesNotExist:
        return Response({"message": f"No notes found"}, status=404)

错误原因 #

relevant_notes.latest('buildNumber')返回的是单个Notes实例，而requested_range是Django的QuerySet对象。|运算符仅支持两个QuerySet之间的合并，直接将单个对象与QuerySet合并会触发属性错误，因为单个模型实例没有QuerySet特有的_known_related_objects属性。

解决方案 #

方案1：将单个对象转为QuerySet后合并

把最新笔记转换成QuerySet，再和范围查询集合并，同时用distinct()避免重复数据（如果最新笔记已经在指定范围内）：

def get(self, request, branchName, stack, resultType, buildNumberMIN, buildNumberMAX, format=None):
    try:
        relevant_notes = Notes.objects.filter(
            branchName=branchName, stack=stack, resultType=resultType)

        requested_range = relevant_notes.filter(
            buildNumber__gte=buildNumberMIN, buildNumber__lte=buildNumberMAX)
        
        latest_note = relevant_notes.latest('buildNumber')
        # 将单个对象转为QuerySet
        latest_note_qs = Notes.objects.filter(pk=latest_note.pk)

        # 合并QuerySet并去重
        return_set = (requested_range | latest_note_qs).distinct()
        
        serializer = serializers.NoteSerializerWithJiraData(return_set, many=True)
        return Response(serializer.data)

    except Notes.DoesNotExist:
        return Response({"message": f"No notes found"}, status=404)

方案2：序列化后手动合并数据

先序列化范围查询的结果，再检查最新笔记是否已在结果中，若不在则手动追加：

def get(self, request, branchName, stack, resultType, buildNumberMIN, buildNumberMAX, format=None):
    try:
        relevant_notes = Notes.objects.filter(
            branchName=branchName, stack=stack, resultType=resultType)

        requested_range = relevant_notes.filter(
            buildNumber__gte=buildNumberMIN, buildNumber__lte=buildNumberMAX)
        
        latest_note = relevant_notes.latest('buildNumber')

        # 序列化范围数据
        range_serializer = serializers.NoteSerializerWithJiraData(requested_range, many=True)
        data_list = list(range_serializer.data)

        # 序列化最新笔记并检查是否已存在
        latest_serializer = serializers.NoteSerializerWithJiraData(latest_note)
        latest_data = latest_serializer.data
        # 以id作为唯一标识判断是否重复
        if not any(item['id'] == latest_data['id'] for item in data_list):
            data_list.append(latest_data)

        return Response(data_list)

    except Notes.DoesNotExist:
        return Response({"message": f"No notes found"}, status=404)

方案对比 #

• 方案1：在数据库层面完成合并和去重，性能更优，适合数据量较大的场景。
• 方案2：在内存中处理数据，逻辑更直观，适合数据量较小的场景。

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