如何在Django中对预取的QuerySet使用Count进行统计？

简化模型代码 #

class Student(models.Model):
    name = models.CharField(max_length=100)

    def __str__(self):
        return self.name


class Routine(models.Model):
    owner =  models.OneToOneField(Student, on_delete=models.CASCADE, related_name='routine')

    def __str__(self):
        return f"Routine | {self.owner}"


class Activity(models.Model):
    title = models.CharField(max_length=100)
    is_active = models.BooleanField(default=True)
    routine = models.ForeignKey(Routine, on_delete=models.CASCADE, related_name='activities')

    def __str__(self):
        return f"{self.title} | {'Active' if self.is_active else 'Expired'} | {self.routine.owner}"

初始实现与问题 #

最初的QuerySet可以预取所有Routine及其关联的is_active=True的Activity，但会包含没有有效Activity的Routine：

routines = Routine.objects.all().prefetch_related(
    Prefetch('activities', queryset=Activity.objects.filter(is_active=True))
)

尝试用Count标注并过滤无效Routine，但因为Count("activities")统计的是所有关联的Activity而非预取过滤后的结果，所以该方法无效：

routines = Routine.objects.all().prefetch_related(
    Prefetch('activities', queryset=Activity.objects.filter(is_active=True))
).annotate(active_activity_count=Count("activities")).exclude(active_activity_count=0)

给Count传入过滤条件可以实现需求，但实际场景中过滤条件复杂，重复编写会增加维护成本：

routines = Routine.objects.all().prefetch_related(
    Prefetch('activities', queryset=Activity.objects.filter(is_active=True))
).annotate(active_activity_count=Count("activities", filter=Q(activities__is_active=True))).exclude(active_activity_count=0)

解决方案：复用过滤条件 #

方法1：定义可复用的条件变量

将过滤条件拆分为针对Activity的基础条件和Routine关联查询的条件，统一引用避免重复：

from django.db.models import Q

# 针对Activity模型的基础过滤条件
activity_base_filter = Q(is_active=True)
# 转换为Routine关联查询的条件
routine_activity_filter = Q(activities__is_active=True)

# 如果是复杂条件，只需修改这里即可
# activity_base_filter = Q(is_active=True) & Q(duration__gt=30)
# routine_activity_filter = Q(activities__is_active=True) & Q(activities__duration__gt=30)

routines = Routine.objects.all().prefetch_related(
    Prefetch('activities', queryset=Activity.objects.filter(activity_base_filter))
).annotate(
    active_activity_count=Count("activities", filter=routine_activity_filter)
).exclude(active_activity_count=0)

方法2：使用Exists实现更简洁的复用

直接定义一次Activity的过滤查询，既用于过滤Routine，又用于预取数据，完全避免重复编写条件：

from django.db.models import Exists, OuterRef

# 定义一次过滤条件的查询，可添加任意复杂条件
active_activities_query = Activity.objects.filter(
    routine=OuterRef('pk'), 
    is_active=True
    # 示例复杂条件：
    # start_time__gte=timezone.now(),
    # category__in=['study', 'exercise']
)

# 先过滤出存在有效Activity的Routine，再预取对应的Activity
routines = Routine.objects.filter(
    Exists(active_activities_query)
).prefetch_related(
    Prefetch('activities', queryset=active_activities_query)
)

这种方式的优势：

• 过滤条件仅编写一次，后续修改只需调整active_activities_query
• 无需额外的annotate和exclude，查询逻辑更简洁高效
• 预取的Activity与过滤条件完全匹配，数据一致性有保障

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