Django嵌套序列化器批量创建时预取关联模型性能优化问题

问题场景 #

批量创建1000+条Site数据时，序列化器会为每个Site的country_code发起单独查询，引发严重的N+1性能问题。涉及模型关系：Project包含多个Site，Site关联Country，一个Country对应多个Site。Silky性能检测显示，2784个站点的请求存在大量重复查询，小批量数据无此问题。

现有代码 #

models.py #

class Project(models.Model):
    project_name = models.CharField(max_length=255)
    status = models.CharField(max_length=255)
    owner = models.ForeignKey(User, null=True, on_delete=models.CASCADE)
    user_id = models.PositiveIntegerField()

    def __str__(self) -> str:
        return self.project_name
    
class Country(models.Model):
    SUPPORTED = "supported"
    NOT_SUPPORTED = "not_supported"
    SUSPENDED = "suspended"
    STATUS_CHOICES = [
        (SUPPORTED, "Supported"),
        (NOT_SUPPORTED, "Not supported"),
        (SUSPENDED, "Suspended"),
    ]
    
    name = models.CharField(max_length=255)
    iso_code2 = models.CharField(max_length=2, primary_key=True)
    iso_code3 = models.CharField(max_length=3, unique=True)
    status = models.CharField(max_length=255, choices=STATUS_CHOICES, default=NOT_SUPPORTED)

    class Meta:
        verbose_name_plural = "Countries"
        ordering = ["name"]

    def __str__(self) -> str:
        return self.name

class Site(models.Model):
    project = models.ForeignKey(Project, on_delete=models.CASCADE, null=True, related_name="sites")
    site_id1 = models.CharField(max_length=255)
    site_id2 = models.CharField(max_length=255, blank=True)
    country_code = models.ForeignKey(Country, on_delete=models.CASCADE)
    postal_code = models.CharField(max_length=255)
    city = models.CharField(max_length=255)
    address = models.CharField(max_length=512)

serializers.py #

class SiteSerializer(serializers.ModelSerializer):
    def validate(self, attrs):
        error_details = {}
        country = attrs.get("country_code")

        if country.status != "supported":
            error_details.setdefault("country", []).append(f"The given country_code: '{country.iso_code2}' ({country.name}) is currently not supported. Current status: {country.status}")
        
        if error_details:
            raise serializers.ValidationError(detail=error_details, code="config_error")
        return super().validate(attrs)
    
    class Meta:
        model = Site
        fields = [
            "site_id1",
            "site_id2",
            "country_code",
            "city",
            "postal_code",
            "address",
        ]

class FileProjectSerializer(serializers.ModelSerializer):
    sites = SiteSerializer(many=True, min_length=1, max_length=3000, required=True)

    class Meta:
        model = Project
        fields = [
            "project_name",
            "status",
            "sites",
            "owner",
            "user_id",
        ]
        read_only_fields = [
            "owner",
            "status",
        ]
    
    def create(self, validated_data):
        sites = validated_data.pop("sites",[])
        project = Project.objects.create(**validated_data)
        for site in sites:
            Site.objects.create(project=project, **site)
        return project

views.py #

class FileUploadView(generics.CreateAPIView):
    parser_classes = [FormParser, MultiPartParser, FileUploadParser]
    permission_classes = [permissions.IsAuthenticated]
    serializer_class = FileProjectSerializer
        
    def perform_create(self, serializer):
        return serializer.save()
    
    def create(self, request, *args, **kwargs):
        file_obj = request.FILES.get("file", None)
        if file_obj is None:
            raise ParseError(detail=f"No file found.")
        
        # 文件转换逻辑（省略，仅用pandas转为字典）
        sites = self.get_data_from_file_as_dict(file_obj)
        sites = self.translate_file_keys_to_model(sites)

        project_name = file_obj.name.replace(".xlsx", "").replace(".csv","")
        project_name = project_name.encode("ascii", "replace").decode()
        # 准备序列化数据（user_id为示例值）
        project = {
            "project_name": project_name,
            "sites": sites,
            "user_id": 2345678543
        }
        # 序列化验证
        serializer = self.get_serializer(data=project)
        serializer.is_valid(raise_exception=True)
        
        instance = self.perform_create(serializer)
        
        response_data = serializer.data
        response = {
            "project_name": response_data["project_name"],
            "sites": response_data["sites"]
        }

        # 删除实例（仅用于返回验证后的JSON，此逻辑可优化）
        instance.delete()
        
        return Response(data=response, status=status.HTTP_200_OK)

解决方案 #

1. 批量预取Country，避免重复查询 #

在视图层先批量获取所有站点涉及的Country，替换原始数据中的编码为对象，同时完成状态验证，避免序列化器逐个查询。修改views.py的create方法，在构造project数据前添加：

# 提取所有站点的country_code值
country_codes = {site["country_code"] for site in sites}
# 批量查询所有涉及的Country，生成编码到对象的映射
country_map = {country.iso_code2: country for country in Country.objects.filter(iso_code2__in=country_codes)}

# 替换站点数据中的编码为Country对象，并验证状态
invalid_messages = []
for idx, site in enumerate(sites):
    code = site["country_code"]
    country = country_map.get(code)
    if not country:
        invalid_messages.append(f"第{idx+1}条站点：无效国家编码{code}")
    elif country.status != Country.SUPPORTED:
        invalid_messages.append(f"第{idx+1}条站点：国家{code}({country.name})当前状态为{country.status}，不支持")
    else:
        site["country_code"] = country

# 存在无效数据直接抛出错误
if invalid_messages:
    raise ParseError(detail="\n".join(invalid_messages))

2. 简化序列化器验证逻辑 #

由于已在视图层完成Country状态验证，移除SiteSerializer中的重复验证：

class SiteSerializer(serializers.ModelSerializer):
    class Meta:
        model = Site
        fields = [
            "site_id1",
            "site_id2",
            "country_code",
            "city",
            "postal_code",
            "address",
        ]

3. 批量创建Site，减少数据库交互 #

将FileProjectSerializer中的循环创建改为批量创建，大幅降低SQL执行次数：

def create(self, validated_data):
    sites = validated_data.pop("sites", [])
    project = Project.objects.create(**validated_data)
    # 批量构造Site对象
    site_objects = [Site(project=project, **site) for site in sites]
    # 批量插入数据库
    Site.objects.bulk_create(site_objects)
    return project

优化效果 #

• Country查询从N+1次减少为1次批量查询
• Site创建从N次单条插入变为1次批量插入，数据库交互次数大幅降低
• 大数量站点（如2784条）的请求性能将显著提升

内容的提问来源于stack exchange，提问作者Árpád Adorjáni