Django优化复杂分组统计：按企业活跃状态统计关联产品

模型定义 #

class Company(models.Model):
    # 其他字段
    cases = models.ManyToMany('Case')
    is_active = models.BooleanField(default=True)

class Case(models.Model):
    # 其他字段
    is_approved = models.BooleanField(default=False)
    product = models.ForeignKey('Product', on_delete=models.CASCADE)

class Product(models.Model):
    # 其他字段

需求说明 #

需要获取所有关联已审批Case的Product，并按Company的活跃状态分组统计，最终得到如下格式的结果：

{
    <Product: 1>: [12, 5], # 12为活跃企业数量，5为非活跃企业数量
    <Product: 3>: [3, 4],
    <Product: 7>: [10, 2]
}

原实现方案（已验证可行） #

companies = Company.objects.prefetch_related('cases').all()
products = {}
for i in companies:
    for c in i.cases.select_related('product').all():
        if c.is_approved == True:
            p = c.product
            if p not in products.keys():
                products[p] = [0, 0]
            if i.is_active == True:
                products[p][0] += 1
            else:
                products[p][1] += 1 # 注：原代码此处存在笔误，应为索引1累加1而非索引0累加2

优化方案 #

原方案在Python层面循环处理数据，数据量较大时效率偏低。推荐直接通过Django ORM在数据库层面完成统计，减少内存占用与循环开销：

方法1：直接通过Product模型聚合统计 #

from django.db.models import Count, Case, When, IntegerField

# 过滤关联已审批Case的Product，同时统计对应活跃/非活跃企业数量
product_stats = Product.objects.filter(
    case__is_approved=True
).annotate(
    active_count=Count(
        Case(
            When(case__company__is_active=True, then=1),
            output_field=IntegerField()
        )
    ),
    inactive_count=Count(
        Case(
            When(case__company__is_active=False, then=1),
            output_field=IntegerField()
        )
    )
).distinct()

# 转换为目标格式字典
result = {product: [product.active_count, product.inactive_count] for product in product_stats}

方法2：先统计ID再关联Product（超大数据量场景优化） #

如果无需完整的Product对象，或数据量极大，可先统计ID再批量关联，进一步降低内存占用：

from django.db.models import Count, Case, When, IntegerField

# 从Case模型出发，按Product分组统计活跃/非活跃企业数
stats = Case.objects.filter(
    is_approved=True
).values('product_id').annotate(
    active_count=Count(Case(When(company__is_active=True, then=1), output_field=IntegerField())),
    inactive_count=Count(Case(When(company__is_active=False, then=1), output_field=IntegerField()))
).distinct()

# 批量获取Product对象，避免N+1查询
product_ids = [stat['product_id'] for stat in stats]
products = Product.objects.in_bulk(product_ids)

# 构建结果字典
result = {products[stat['product_id']]: [stat['active_count'], stat['inactive_count']] for stat in stats}

优化说明 #

• 解决原方案的N+1查询风险：原代码中内层的select_related('product')可能触发重复查询，优化方案通过ORM关联一次性完成统计。
• 数据库聚合效率更高：数据库原生的聚合操作比Python循环处理大量数据更高效，减少内存中数据处理的压力。
• 修正原代码笔误：原代码中非活跃企业统计逻辑错误，已调整为符合需求的索引1累加1。

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