优化Django多对多关联（含中间模型）的查询次数

问题背景 #

需要减少skills_percentage方法的重复查询次数，现有Django模型定义如下：

class Skill(models.Model):
    name = models.TextField()

class Employee(models.Model):
    firstname = models.TextField()
    skills = models.ManyToManyField(Skill, through='SkillStatus')

    def skills_percentage(self):
        completed = 0
        total = 0

        for skill in self.skills.all().prefetch_related("skillstatus_set__employee"):
            for item in skill.skillstatus_set.all():
                if item.employee.firstname == self.firstname:
                    total += 1
                    if item.status:
                        completed += 1
        try:
            percentage = round((completed / total * 100), 2)
        except ZeroDivisionError:
            percentage = 0.0
        return f"{percentage} %"

class SkillStatus(models.Model):
    employee = models.ForeignKey(Employee, on_delete=models.CASCADE)
    skill = models.ForeignKey(Skill, on_delete=models.CASCADE)
    status = models.BooleanField(default=False)

当前核心问题是skills_percentage方法计算时产生过多查询，已尝试prefetch_related初步优化，但Django Debug Toolbar仍显示额外查询，试过select_related与prefetch_related的不同组合及其他计算方式，仍存在查询过多问题，以下是优化方案：

优化方案 #

1. 直接从中间表SkillStatus做聚合查询 #

完全绕开遍历技能的逻辑，直接针对当前员工的SkillStatus记录做聚合统计，仅需1次查询：

from django.db.models import Count, Sum

def skills_percentage(self):
    # 聚合当前员工的技能状态记录
    stats = self.skillstatus_set.aggregate(
        total=Count('id'),
        completed=Sum('status')  # BooleanField在Sum中True会被当作1，False当作0
    )
    total = stats['total'] or 0
    completed = stats['completed'] or 0
    
    if total == 0:
        return "0.0 %"
    percentage = round((completed / total * 100), 2)
    return f"{percentage} %"

优势：

• 彻底消除循环带来的N+1查询问题，仅执行1次SQL聚合查询
• 逻辑简洁，避免遍历和冗余条件判断

2. 精准预取当前员工的SkillStatus记录 #

如果必须保留遍历逻辑，可调整prefetch_related的方式，只预取当前员工的关联记录，避免加载其他员工的数据：

from django.db.models import Prefetch

def skills_percentage(self):
    # 仅预取当前员工的SkillStatus记录，并映射到自定义属性
    skills = self.skills.prefetch_related(
        Prefetch(
            'skillstatus_set',
            queryset=SkillStatus.objects.filter(employee=self),
            to_attr='employee_skill_status'
        )
    )
    completed = 0
    total = 0
    for skill in skills:
        # 直接使用预取的当前员工专属状态记录
        for item in skill.employee_skill_status:
            total += 1
            if item.status:
                completed += 1
    try:
        percentage = round((completed / total * 100), 2)
    except ZeroDivisionError:
        percentage = 0.0
    return f"{percentage} %"

优势：

• 预取数据仅包含当前员工的SkillStatus，减少内存占用
• 避免原代码中遍历所有skillstatus_set再判断员工的冗余逻辑
• 仅产生2次查询（一次查询技能，一次查询当前员工的技能状态）

3. 缓存计算结果（可选） #

如果技能完成率不会频繁变化，可给方法添加缓存，避免重复计算：

from django.core.cache import cache

def skills_percentage(self):
    cache_key = f"employee_skill_percent_{self.pk}"
    percentage = cache.get(cache_key)
    if percentage is not None:
        return percentage
    
    # 复用第一种聚合查询逻辑
    stats = self.skillstatus_set.aggregate(
        total=Count('id'),
        completed=Sum('status')
    )
    total = stats['total'] or 0
    completed = stats['completed'] or 0
    
    percentage = "0.0 %" if total == 0 else f"{round((completed / total * 100), 2)} %"
    
    # 设置缓存过期时间（示例为1小时）
    cache.set(cache_key, percentage, 3600)
    return percentage

注意：当员工的技能状态发生变化时，需要手动清除对应缓存键，保证数据一致性。

内容的提问来源于stack exchange，提问作者Антон Ласько