Django ORM子查询文本聚合问题：仅返回单个Puzzle ID而非拼接结果

我有两个Django模型，尝试设计自定义管理器/查询集，为每个Series标注符合特定条件的关联Puzzle的ID，格式为'2,13,26'这样的拼接字符串。

简化模型如下：

class Series(models.Model):
    puzzles = models.ManyToManyField(
        Puzzle, through='SeriesElement', related_name='series')
    is_public = models.BooleanField(null=False, blank=False, default=False)

class Puzzle(models.Model):
    pass

自定义聚合器代码：

from django.db.models.aggregates import Aggregate
from django.db.models.functions import Coalesce
from django.db.models.fields import CharField
from django.db.models.expressions import Value

class GroupConcat(Aggregate):
    """
    兼容MySQL和SQLite的GROUP_CONCAT聚合器
    """
    function = 'GROUP_CONCAT'

    def __init__(self, expression, distinct=False, ordering=None, **extra):
        super(GroupConcat, self).__init__(expression,
                                          distinct='DISTINCT ' if distinct else '',
                                          ordering=' ORDER BY %s' % ordering if ordering is not None else '',
                                          output_field=CharField(),
                                          **extra)


    def as_sqlite(self, compiler, connection, **extra):
        return super().as_sql(compiler,
                              connection, 
                              template='%(function)s(%(distinct)s%(expressions)s%(ordering)s)',
                              **extra)

尝试实现的代码：

pzl_sub = apps.get_model('puzzles', 'Puzzle').objects.filter(series__id= OuterRef('id'))
pzl_sub = pzl_sub.filter(series_elements__isnull=False).add_nb_public_series().filter(nb_public_series=5)
pzl_ids= pzl_sub.order_by().values('id')
qs = Series.objects.annotate(id_str_pzl = GroupConcat(pzl_ids))

但结果仅返回一个符合条件的Puzzle ID，而非所有符合条件的ID拼接结果，问题出在哪里？

核心错误是直接将子查询的values('id') QuerySet传给了GroupConcat，Django聚合函数需要接收字段引用（如F()表达式）或字段名，而非QuerySet对象。以下是两种修正方案：

方案一：通过关联关系直接聚合（推荐） #

利用Series与Puzzle的多对多关联，直接过滤符合条件的Puzzle并聚合ID，逻辑更简洁高效：

from django.db.models import F

qs = Series.objects.annotate(
    id_str_pzl=GroupConcat(
        F('puzzles__id'),
        ordering='puzzles__id'  # 可选，指定ID排序规则
    )
).filter(
    puzzles__series_elements__isnull=False,
    puzzles__nb_public_series=5
).distinct()

• 用F('puzzles__id')指定要聚合的字段，GroupConcat会自动拼接所有符合条件的Puzzle ID。
• 添加distinct()避免多对多关联产生的笛卡尔积导致Series重复。

方案二：子查询内部完成聚合 #

如果业务逻辑必须依赖子查询过滤Puzzle，需在子查询内部先完成GroupConcat，再通过Subquery注入主查询：

from django.db.models import Subquery, OuterRef

# 子查询：对当前Series关联的符合条件的Puzzle ID进行聚合
pzl_sub = apps.get_model('puzzles', 'Puzzle').objects.filter(
    series__id=OuterRef('id'),
    series_elements__isnull=False
).add_nb_public_series().filter(
    nb_public_series=5
).order_by().annotate(
    concat_ids=GroupConcat('id')
).values('concat_ids')

# 主查询：将子查询的聚合结果赋值给Series的id_str_pzl字段
qs = Series.objects.annotate(
    id_str_pzl=Subquery(pzl_sub[:1])  # [:1]确保子查询仅返回单个聚合结果
)

• 子查询内部先完成ID拼接，再通过Subquery把结果传递给主查询的annotate。
• [:1]是必要的，因为Subquery要求子查询只能返回单条结果。

错误点总结 #

1. GroupConcat的参数应为字段表达式或字段名，不能直接传入values('id')生成的QuerySet。
2. 子查询方式需在子查询内部完成聚合操作，而非在主查询的聚合函数中嵌套子查询。

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