Django中annotate结合Subquery获取符合条件全量数据的问题

问题背景 #

我有两个包含多对多（M2M）字段的模型，因仅需读取数据，希望通过单次数据库查询获取所有所需内容。使用prefetch_related搭配Prefetch对象可过滤关联数据并缓存到指定属性，但该方式会产生额外查询；尝试用annotate结合Subquery实现单次查询时，发现annotate生成的字段仅返回单个值（明明存在多个is_special=True的Point实例），而prefetch方式却能正常获取多个特殊点。

模型定义 #

from django.db import models

class Point(models.Model):
    name = models.CharField(max_length=100)
    is_special = models.BooleanField(default=False)

class Route(models.Model):
    name = models.CharField(max_length=100)
    points = models.ManyToManyField(Point, related_name='routes')

视图代码及问题表现 #

1. Prefetch方式（可行但多查询） #

from django.db.models import Prefetch

def route_list(request):
    routes = Route.objects.all().prefetch_related(
        Prefetch(
            'points',
            queryset=Point.objects.filter(is_special=True),
            to_attr='special_points'
        )
    )
    # 每个route.special_points是符合条件的Point实例列表，但会触发额外SELECT查询
    return render(request, 'routes/list.html', {'routes': routes})

• 问题：能拿到正确的列表，但Django会先查询所有Route，再查询关联的符合条件的Point，共两次查询。

2. Annotate+Subquery方式（仅返回单个值） #

from django.db.models import Subquery, OuterRef

def route_list(request):
    special_points_subquery = Point.objects.filter(
        routes=OuterRef('pk'),
        is_special=True
    ).values('name')

    routes = Route.objects.annotate(
        special_points=Subquery(special_points_subquery)
    )
    # 每个route.special_points仅返回单个name值，而非预期的列表
    return render(request, 'routes/list.html', {'routes': routes})

• 问题：Subquery默认只返回子查询的第一条结果，无法获取多个符合条件的Point值。

排查过程 #

• 查看ORM生成的SQL：通过print(routes.query)发现，annotate方式是将子查询作为主查询的单个字段，但子查询未做聚合处理，仅返回单行单列，因此只能拿到单个值。
• 尝试values_list+flat=True，结果依旧为单个值，因为Subquery不支持返回多行结果作为字段值。

可行的原生SQL示例 #

通过数据库聚合函数实现单次查询，不同数据库语法略有差异：

PostgreSQL版本 #

SELECT 
    route.id, 
    route.name, 
    ARRAY_AGG(point.name) AS special_points
FROM route
LEFT JOIN route_points ON route.id = route_points.route_id
LEFT JOIN point ON route_points.point_id = point.id AND point.is_special = true
GROUP BY route.id, route.name;

MySQL版本 #

SELECT 
    route.id, 
    route.name, 
    GROUP_CONCAT(point.name SEPARATOR ',') AS special_points
FROM route
LEFT JOIN route_points ON route.id = route_points.route_id
LEFT JOIN point ON route_points.point_id = point.id AND point.is_special = true
GROUP BY route.id, route.name;

解决方案 #

方案1：annotate结合聚合函数（适配不同数据库） #

利用Django聚合函数，将多值聚合为数组或字符串，实现单次查询：

PostgreSQL环境

from django.contrib.postgres.aggregates import ArrayAgg
from django.db.models import Q

def route_list(request):
    routes = Route.objects.annotate(
        special_points=ArrayAgg(
            'points__name',
            filter=Q(points__is_special=True),
            default=list()
        )
    )
    # route.special_points为符合条件的name列表，仅单次查询
    return render(request, 'routes/list.html', {'routes': routes})

通用数据库（如MySQL）

from django.db.models import Aggregate, CharField

class GroupConcat(Aggregate):
    function = 'GROUP_CONCAT'
    template = '%(function)s(%(expressions)s SEPARATOR ",")'
    output_field = CharField()

def route_list(request):
    routes = Route.objects.annotate(
        special_points=GroupConcat(
            'points__name',
            filter=Q(points__is_special=True),
            default=''
        )
    )
    # 将逗号分隔的字符串转为列表
    for route in routes:
        route.special_points = route.special_points.split(',') if route.special_points else []
    return render(request, 'routes/list.html', {'routes': routes})

方案2：接受Prefetch的两次查询（性能可接受场景） #

如果数据库查询性能不是瓶颈，prefetch_related+Prefetch是Django推荐的关联数据处理方式，两次查询的开销在大多数场景下可忽略，且代码更简洁、易维护。

总结 #

• 若必须严格单次查询，根据数据库类型选择对应聚合函数进行annotate，将多值聚合为数组或字符串后再处理。
• 若可接受两次查询，prefetch_related+Prefetch是更简洁的方案，Django ORM会优化查询逻辑，实际性能表现良好。

内容的提问来源于stack exchange，提问作者mh-firouzjah