如何在Django中按字段分组并直接注解关联模型对象？

现有代码如下：

class City(models.Model):
    postal_code = models.IntegerField()

class CountryViewset(GenericViewSet, ListModelMixin, RetrieveModelMixin):
    queryset = City.objects.values("country_code").annotate(cities_ids=ArrayAgg("id"))
    serializer_class = CountrySerializer

class CountrySerializer(serializers.Serializer):
    country_code = serializers.IntegerField()
    cities_ids = serializers.ListField(child=serializers.IntegerField())

当前接口返回结果：

{
    "country_code": 1,
    "cities_ids": [1, 2, 3]
}

需求：能否直接在注解中返回模型对象？希望查询集返回形式如下：

<Queryset [
  {
    "country_code": 1,
    "cities_ids": <QuerySet [<City: 1>, <City: 2>, <City: 3>]>,
  },
  ...
]>

原因：未来要添加大量Django过滤器，为了性能不想每次重新获取/过滤City对象；且无法创建Country模型，不允许用manage命令定期同步该模型。

不能直接用ArrayAgg注解返回模型对象，因为Django的聚合函数是数据库层面的操作，只能返回字段值，无法直接映射为ORM模型实例。不过可以通过以下两种方式实现需求，同时保证性能：

1. 内存分组+预取数据 #

放弃values+annotate的数据库分组方式，一次性查询所有City数据后在内存中按country_code分组，全程仅需一次数据库请求：

class CountryViewset(GenericViewSet, ListModelMixin, RetrieveModelMixin):
    def get_queryset(self):
        # 可提前在此处添加City相关过滤器，减少后续内存处理数据量
        cities_queryset = City.objects.all()
        grouped_data = {}
        
        for city in cities_queryset:
            code = city.country_code
            if code not in grouped_data:
                grouped_data[code] = {
                    'country_code': code,
                    'cities_ids': []
                }
            grouped_data[code]['cities_ids'].append(city)
        
        # 转换为列表格式，若需要QuerySet特性可自定义类包装
        return list(grouped_data.values())

    serializer_class = CountrySerializer

同步修改序列化器，支持直接序列化City模型实例：

class CitySerializer(serializers.ModelSerializer):
    class Meta:
        model = City
        fields = ['id', 'postal_code']

class CountrySerializer(serializers.Serializer):
    country_code = serializers.IntegerField()
    cities_ids = serializers.ListField(child=CitySerializer())

2. 聚合ID+批量查询实例 #

先通过数据库聚合获取分组后的ID列表，再批量查询所有City实例并映射到分组结果中，仅需两次数据库请求：

from django.db.models import ArrayAgg

class CountryViewset(GenericViewSet, ListModelMixin, RetrieveModelMixin):
    def get_queryset(self):
        # 第一步：按country_code分组获取对应City ID列表
        grouped_id_data = City.objects.values("country_code").annotate(cities_ids=ArrayAgg("id"))
        # 提取所有需要的City ID，批量查询
        all_city_ids = [city_id for group in grouped_id_data for city_id in group['cities_ids']]
        city_instance_map = {city.id: city for city in City.objects.filter(id__in=all_city_ids)}
        # 将ID替换为模型实例
        for group in grouped_id_data:
            group['cities_ids'] = [city_instance_map[city_id] for city_id in group['cities_ids']]
        
        return grouped_id_data

    serializer_class = CountrySerializer

关键提示 #

• 数据库层面无法直接返回ORM模型实例，所有模型实例都是查询后在内存中构建的；
• 内存分组方式适合数据量较小的场景，操作更灵活；批量查询方式适合数据量大的场景，内存占用更低；
• 两种方式都能避免后续重复查询City对象，满足性能优化需求。

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