如何用带过滤关联查询集注解的QuerySet替换指定函数？

模型定义 #

class A(models.Model):
    name = models.CharField()


class B(models.Model):
    name = models.CharField()


class C(models.Model):
    name = models.CharField()


class D(models.Model):
    a = models.ForeignKey(A, related_name='set_of_a')
    b = models.ForeignKey(B, related_name='set_of_d')
    set_of_c = models.ManyToManyField(C, related_name='set_of_d')

需求说明 #

需要用QuerySet替代以下含伪代码的函数功能：

def get_queryset(name, list_of_c):
    result = list(A.objects.filter(name__icontains=name))
    for a in result:
        a.map_b_to_list_of_d = {}  # annotation
        for b in B.objects.filter(set_of_d__set_of_c__intersects_with=list_of_d):
            a.map_b_to_list_of_d.setdefault(c, []) += list(b.set_of_d.filter(a__name__icontains=name, set_of_c__intersects_with=list_of_c))
    return result

核心功能：

• 筛选name包含指定值的A实例
• 为每个A实例添加map_b_to_list_of_d注解：这是一个字典，键为符合条件的B实例，值为对应D实例列表（D需关联当前A，且关联的C集合与list_of_c有交集）

解决方案 #

利用Django的ORM特性（Subquery、ArrayAgg）实现无循环的QuerySet查询，提升性能：

from django.db.models import Subquery, OuterRef, ArrayAgg, Q

def get_queryset(name, list_of_c):
    # 子查询：筛选关联当前A且C集合与list_of_c有交集的D实例主键
    valid_d_pks = Subquery(
        D.objects.filter(
            a=OuterRef('pk'),
            set_of_c__in=list_of_c
        ).distinct().values('pk')
    )

    # 为B实例注解符合条件的D主键列表，过滤掉无对应D的B
    b_with_valid_ds = B.objects.annotate(
        valid_d_list=ArrayAgg('set_of_d__pk', filter=Q(set_of_d__pk__in=valid_d_pks))
    ).exclude(valid_d_list__isnull=True)

    # 为A实例注解B到D的映射结构
    a_qs = A.objects.filter(name__icontains=name).annotate(
        map_b_to_list_of_d=Subquery(
            b_with_valid_ds.values(
                'pk',
                'name',
                'valid_d_list'
            ).order_by('pk')
        )
    )

    # 将注解结构转换为B实例到D QuerySet的字典，和原函数输出结构对齐
    for a in a_qs:
        a.map_b_to_list_of_d = {
            B.objects.get(pk=item['pk']): D.objects.filter(pk__in=item['valid_d_list'])
            for item in a.map_b_to_list_of_d
        }

    return a_qs

优化说明（Django 3.2+） #

如果使用Django 3.2及以上版本，可通过JSONAgg直接生成字典格式，省去后续循环转换步骤：

from django.db.models import JSONAgg

# 修改A实例的注解部分
a_qs = A.objects.filter(name__icontains=name).annotate(
    map_b_to_list_of_d=JSONAgg(
        b_with_valid_ds.values(
            'pk',
            'name',
            'valid_d_list'
        ).order_by('pk')
    )
)

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