如何在Django REST Framework中实现关联与聚合查询

1. 优化模型关联（可选但推荐） #

给NoiseData的外键添加related_name，让反向关联查询更直观：

class NoiseData(models.Model):
    location = models.CharField(max_length=100)
    # 添加related_name，方便通过City模型直接关联NoiseData
    city_id = models.ForeignKey(City, on_delete=models.CASCADE, related_name='noise_records')
    noise_type = models.CharField(max_length=30)

2. 用ORM实现聚合查询 #

通过Django ORM一次性完成所有统计计算，减少数据库请求：

适用于PostgreSQL（支持JSON聚合函数） #

from django.db.models import Count, F, Case, When, Value, Subquery, OuterRef
from django.db.models.functions import Func
from django.db import models

# 子查询：按城市分组统计各噪音类型的数量
noise_type_subquery = NoiseData.objects.filter(
    city_id=OuterRef('pk')
).values('noise_type').annotate(
    type_count=Count('id')
).values_list('noise_type', 'type_count')

# 主查询：聚合城市的所有统计数据
city_stats = City.objects.annotate(
    # 统计当前城市的噪音数据总数
    count=Count('noise_records'),
    # 计算数量/城市面积的归一化值，处理面积为0的情况
    count_area_normalized=Case(
        When(area=0, then=Value(0)),
        default=F('count') / F('area'),
        output_field=models.FloatField()
    ),
    # 生成噪音类型统计的JSON对象
    noise_type_metadata=Func(
        Subquery(noise_type_subquery.values('noise_type')),
        Subquery(noise_type_subquery.values('type_count')),
        function='JSON_OBJECT_AGG',
        output_field=models.JSONField()
    )
).values('id', 'name', 'count', 'count_area_normalized', 'noise_type_metadata')

适用于SQLite（无JSON聚合函数） #

from django.db.models import Count, F, Case, When, Value, Subquery, OuterRef

# 子查询返回噪音类型统计列表
noise_type_subquery = NoiseData.objects.filter(
    city_id=OuterRef('pk')
).values('noise_type').annotate(
    count=Count('id')
).values('noise_type', 'count')

city_stats = City.objects.annotate(
    count=Count('noise_records'),
    count_area_normalized=Case(
        When(area=0, then=Value(0)),
        default=F('count') / F('area'),
        output_field=models.FloatField()
    ),
    noise_type_metadata=Subquery(noise_type_subquery)
).values('id', 'name', 'count', 'count_area_normalized', 'noise_type_metadata')

3. 序列化器处理数据格式 #

用Django REST Framework序列化器统一输出格式，处理SQLite下的列表转字典逻辑：

from rest_framework import serializers

class CityNoiseStatsSerializer(serializers.Serializer):
    id = serializers.IntegerField()
    name = serializers.CharField()
    count = serializers.IntegerField()
    count_area_normalized = serializers.FloatField()
    noise_type_metadata = serializers.JSONField()

    def to_representation(self, instance):
        data = super().to_representation(instance)
        # 把列表格式的噪音统计转成键值对JSON
        if isinstance(data['noise_type_metadata'], list):
            data['noise_type_metadata'] = {item['noise_type']: item['count'] for item in data['noise_type_metadata']}
        return data

4. 编写API视图 #

创建视图返回聚合后的JSON数据：

from rest_framework.views import APIView
from rest_framework.response import Response

class CityNoiseAggregateAPI(APIView):
    def get(self, request):
        # 执行对应数据库类型的ORM查询（示例用PostgreSQL版本）
        noise_type_subquery = NoiseData.objects.filter(
            city_id=OuterRef('pk')
        ).values('noise_type').annotate(
            type_count=Count('id')
        ).values_list('noise_type', 'type_count')

        city_stats = City.objects.annotate(
            count=Count('noise_records'),
            count_area_normalized=Case(
                When(area=0, then=Value(0)),
                default=F('count') / F('area'),
                output_field=models.FloatField()
            ),
            noise_type_metadata=Func(
                Subquery(noise_type_subquery.values('noise_type')),
                Subquery(noise_type_subquery.values('type_count')),
                function='JSON_OBJECT_AGG',
                output_field=models.JSONField()
            )
        ).values('id', 'name', 'count', 'count_area_normalized', 'noise_type_metadata')

        serializer = CityNoiseStatsSerializer(city_stats, many=True)
        return Response(serializer.data)

配置URL #

在项目urls.py中添加API路由：

from django.urls import path
from .views import CityNoiseAggregateAPI

urlpatterns = [
    path('city-noise-stats/', CityNoiseAggregateAPI.as_view(), name='city-noise-stats'),
]

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