Django非托管模型反向外键失效及RawSQL多查询优化咨询

问题背景 #

定义了两个非托管Django模型：

class A(models.Model):
    # some fields
    class Meta:
        app_label = "app"
        db_table = "table_a"
        managed = False

class B(models.Model):
    # some fields

    a_fk = models.ForeignKey(A, on_delete=models.RESTRICT, related_name="b_objects")

    class Meta:
        app_label = "app"
        db_table = "table_b"
        managed = False

为B的外键设置了related_name="b_objects"，期望从A对象关联查询B对象，统计每个A对应的B数量时，使用以下代码报错：

class InventoryLocationsCollectionView(ListAPIView):
    queryset = A.objects.prefetch_related("b_objects").annotate(total=Count("b_objects"))

错误提示A模型不存在b_objects字段。尝试raw查询时，因RawQuerySet不支持.all()及后续过滤、分页需求而放弃。

后续用annotate + RawSQL实现了统计：

queryset = A.objects.annotate(
    total=RawSQL("select count(1) from table_b where a_id = table_a.id", params=[]),
    london=RawSQL("select count(1) from table_b where a_id = table_a.id and city = 'London'", params=[]),
    # 另外两个城市的统计RawSQL
)

但多个子查询导致性能大幅下降，希望优化为单次查询。

解决方案 #

1. 修复反向外键不生效问题 #

非托管模型本身支持反向外键关联，报错通常是以下原因：

• 模型字段与数据库结构不匹配：检查A的主键是否与table_a的主键一致，B的a_fk对应的数据库列（如a_id）是否正确关联到table_a的主键，且字段类型匹配。
• 拼写错误：确认related_name的拼写在annotate中是否正确（比如是否写成b_object而非b_objects）。
• Django元数据缓存：尝试重启服务或清理migrations缓存（删除__pycache__文件夹）。

2. 多条件统计优化：用Case When合并查询 #

使用Django的条件表达式Case+When，可以将多个统计字段合并为一次关联查询，避免多次子查询的性能损耗：

方式一：用Count+Case实现

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

queryset = A.objects.annotate(
    # 统计总数
    total=Count("b_objects"),
    # 统计伦敦的数量
    london=Count(
        Case(
            When(b_objects__city="London", then=1),
            output_field=IntegerField()
        )
    ),
    # 统计巴黎的数量
    paris=Count(
        Case(
            When(b_objects__city="Paris", then=1),
            output_field=IntegerField()
        )
    ),
    # 统计纽约的数量
    new_york=Count(
        Case(
            When(b_objects__city="New York", then=1),
            output_field=IntegerField()
        )
    )
)

方式二：用Sum+Case实现（更直观）

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

queryset = A.objects.annotate(
    total=Count("b_objects"),
    london=Sum(
        Case(
            When(b_objects__city="London", then=1),
            default=0,
            output_field=IntegerField()
        )
    ),
    paris=Sum(
        Case(
            When(b_objects__city="Paris", then=1),
            default=0,
            output_field=IntegerField()
        )
    ),
    new_york=Sum(
        Case(
            When(b_objects__city="New York", then=1),
            default=0,
            output_field=IntegerField()
        )
    )
)

原理说明 #

以上两种方式都会生成一次LEFT JOIN table_b的SQL查询，在SELECT子句中通过条件判断计算各城市的统计数，相比多个RawSQL子查询，减少了数据库的IO开销，性能提升明显。同时，该查询支持后续的过滤、分页等ORM操作，满足业务需求。

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