如何在Django中查询四层嵌套外键下用户拥有的对象及父级？

首先，你遇到的重复Level1对象问题，是因为Django ORM在处理跨层级关联查询时，会为每个匹配的子对象生成一条关联记录。咱们先解决去重问题，再一步步优化查询性能和结构组装。

一、优化查询：避免重复并高效获取相关对象 #

1. 用Exists子查询替代Q对象（更高效） #

原查询用Q对象会产生大量重复记录，改用Exists子查询可以只检查是否存在匹配的子对象，既避免重复，又提升查询性能：

from django.db.models import Exists, OuterRef

query_id = # 指定用户的ID

# 定义各层级的存在性子查询
level2_has_user = Level2.objects.filter(parent=OuterRef('pk'), owner=query_id)
level3_has_user = Level3.objects.filter(parent__parent=OuterRef('pk'), owner=query_id)
level4_has_user = Level4.objects.filter(parent__parent__parent=OuterRef('pk'), owner=query_id)

# 获取所有关联指定用户的Level1（含自身属于用户或子层级有用户对象的）
level1_queryset = Level1.objects.filter(
    Q(owner=query_id) |
    Exists(level2_has_user) |
    Exists(level3_has_user) |
    Exists(level4_has_user)
).distinct()

2. 嵌套预取关联对象（避免N+1查询） #

为了一次性获取所有需要的子对象，用Prefetch对象嵌套预取，只加载和指定用户相关的层级（或用户对象的父节点）：

from django.db.models import Prefetch

# 预取属于用户的Level4
level4_prefetch = Prefetch(
    'level3__level4',
    queryset=Level4.objects.filter(owner=query_id),
    to_attr='user_level4'
)

# 预取属于用户的Level3，或包含用户Level4的Level3
level3_prefetch = Prefetch(
    'level2__level3',
    queryset=Level3.objects.filter(
        Q(owner=query_id) | Q(level4__owner=query_id)
    ).distinct().prefetch_related(level4_prefetch),
    to_attr='user_level3'
)

# 预取属于用户的Level2，或包含用户Level3/Level4的Level2
level2_prefetch = Prefetch(
    'level2',
    queryset=Level2.objects.filter(
        Q(owner=query_id) | Q(level3__owner=query_id) | Q(level3__level4__owner=query_id)
    ).distinct().prefetch_related(level3_prefetch),
    to_attr='user_level2'
)

# 最终查询集，预取所有必要子对象
level1_queryset = Level1.objects.filter(
    Q(owner=query_id) |
    Exists(level2_has_user) |
    Exists(level3_has_user) |
    Exists(level4_has_user)
).distinct().prefetch_related(level2_prefetch)

这样每个Level1对象会有user_level2属性，里面包含符合条件的Level2对象；每个Level2有user_level3，以此类推，结构清晰且查询次数最少。

二、序列化器组装结构（以DRF为例） #

如果用Django REST Framework，可以编写嵌套序列化器，直接基于预取的属性序列化：

from rest_framework import serializers

class Level4Serializer(serializers.ModelSerializer):
    class Meta:
        model = Level4
        fields = ['id', 'owner']

class Level3Serializer(serializers.ModelSerializer):
    user_level4 = Level4Serializer(many=True)

    class Meta:
        model = Level3
        fields = ['id', 'owner', 'user_level4']

class Level2Serializer(serializers.ModelSerializer):
    user_level3 = Level3Serializer(many=True)

    class Meta:
        model = Level2
        fields = ['id', 'owner', 'user_level3']

class Level1Serializer(serializers.ModelSerializer):
    user_level2 = Level2Serializer(many=True)

    class Meta:
        model = Level1
        fields = ['id', 'owner', 'user_level2']

# 使用示例
serializer = Level1Serializer(level1_queryset, many=True)
response_data = serializer.data

三、备选方案：批量查询+手动构建树结构 #

如果追求极致性能，可以先批量获取所有需要的对象，再手动构建树结构，避免ORM的额外开销：

# 1. 获取所有属于指定用户的对象ID
user_level1 = set(Level1.objects.filter(owner=query_id).values_list('pk', flat=True))
user_level2 = set(Level2.objects.filter(owner=query_id).values_list('pk', flat=True))
user_level3 = set(Level3.objects.filter(owner=query_id).values_list('pk', flat=True))
user_level4 = set(Level4.objects.filter(owner=query_id).values_list('pk', flat=True))

# 2. 收集所有需要的父节点ID
level2_parents = set(Level2.objects.filter(pk__in=user_level2).values_list('parent_id', flat=True))
level3_parents = set(Level3.objects.filter(pk__in=user_level3).values_list('parent_id', flat=True))
level3_grandparents = set(Level2.objects.filter(pk__in=level3_parents).values_list('parent_id', flat=True))
level4_parents = set(Level4.objects.filter(pk__in=user_level4).values_list('parent_id', flat=True))
level4_grandparents = set(Level3.objects.filter(pk__in=level4_parents).values_list('parent_id', flat=True))
level4_greatgrandparents = set(Level2.objects.filter(pk__in=level4_grandparents).values_list('parent_id', flat=True))

# 3. 合并所有需要的对象ID
all_level1_ids = user_level1.union(level2_parents, level3_grandparents, level4_greatgrandparents)
all_level2_ids = user_level2.union(level3_parents, level4_grandparents)
all_level3_ids = user_level3.union(level4_parents)
all_level4_ids = user_level4

# 4. 批量获取所有对象（用字典存储方便查找）
level1_map = {obj.pk: obj for obj in Level1.objects.filter(pk__in=all_level1_ids)}
level2_map = {obj.pk: obj for obj in Level2.objects.filter(pk__in=all_level2_ids)}
level3_map = {obj.pk: obj for obj in Level3.objects.filter(pk__in=all_level3_ids)}
level4_map = {obj.pk: obj for obj in Level4.objects.filter(pk__in=all_level4_ids)}

# 5. 构建树结构
for pk in all_level4_ids:
    level4 = level4_map[pk]
    level3 = level3_map[level4.parent_id]
    level3.children = getattr(level3, 'children', []) + [level4]

for pk in all_level3_ids:
    level3 = level3_map[pk]
    level2 = level2_map[level3.parent_id]
    level2.children = getattr(level2, 'children', []) + [level3]

for pk in all_level2_ids:
    level2 = level2_map[pk]
    level1 = level1_map[level2.parent_id]
    level1.children = getattr(level1, 'children', []) + [level2]

# 最终结果就是所有Level1对象的列表
result = list(level1_map.values())

这种方式所有查询都是批量的，没有冗余数据，性能最优，适合查询频率极高的场景。

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