Django序列化器文件数据丢失问题排查及解决方案

开发过程中遇到问题：Django序列化器的validate和create函数存在数据丢失情况，导致创建包含图片的Post时，关联的PostsImage对象无法生成，测试用例执行失败。

相关代码 #

模型定义 #

class PostsImage(models.Model):
    image = models.ImageField(upload_to="posts/images")
    post = models.ForeignKey(
        "socialmedia.Post", on_delete=models.CASCADE, related_name="posts_images"
    )

    class Meta:
        verbose_name = "PostsImage"
        verbose_name_plural = "PostsImages"

    def __str__(self):
        return self.image


class PostsAttachedFile(models.Model):
    attached = models.FileField(upload_to="posts/attached")
    post = models.ForeignKey(
        "socialmedia.Post",
        on_delete=models.CASCADE,
        related_name="posts_attached_files",
    )

    class Meta:
        verbose_name = "PostsAttachedFile"
        verbose_name_plural = "PostsAttachedFiles"

    def __str__(self):
        return self.attached


class Post(models.Model):
    title = models.CharField(max_length=255)
    content = models.TextField()
    business_sector = models.CharField(max_length=50, choices=BUSINESS_SECTOR_CHOICES)
    author = models.ForeignKey(
        "users.User", on_delete=models.CASCADE, related_name="posts"
    )
    created_at = models.DateTimeField(auto_now_add=True)
    valid_until = models.DateTimeField()
    location_type = models.CharField(max_length=20, choices=LOCATION_CHOICES)
    location = models.CharField(max_length=255)
    location_radius = models.PositiveIntegerField(null=True, blank=True)

    class Meta:
        verbose_name = "Post"
        verbose_name_plural = "Posts"

    def __str__(self):
        return self.title

序列化器代码 #

class PostSerializer(serializers.ModelSerializer):
    images = PostsImageSerializer(many=True, source="posts_images", required=False)
    attached_files = PostsAttachedFileSerializer(
        many=True, source="posts_attached_files", required=False
    )

    class Meta:
        model = Post
        fields = "__all__"

    def validate(self, attrs):
        print(attrs)
        author = attrs.get("author")
        location_type = attrs.get("location_type")
        location_radius = attrs.get("location_radius")
        valid_until = attrs.get("valid_until")
        user = self.context["request"].user

        if author and author != user:
            raise serializers.ValidationError(
                {"author": "You can't create post for another user"}
            )

        if location_type == "local" and not location_radius:
            raise serializers.ValidationError(
                {
                    "location_radius": "You must provide location radius if location type is local",
                }
            )

        if valid_until and valid_until.timestamp() < datetime.now().timestamp():
            raise serializers.ValidationError(
                {"valid_until": "Valid until must be in the future"}
            )
        print("#### validation ok")
        return attrs

    def create(self, validated_data):
        print(validated_data)
        images = None
        attached_files = None
        if "posts_images" in validated_data.keys():
            images = validated_data.pop("posts_images")
        if "posts_attached_files" in validated_data.keys():
            attached_files = validated_data.pop("posts_attached_files")
        post = Post.objects.create(**validated_data)

        images_from_files = self.context["request"].FILES.getlist("posts_images")
        print(images_from_files)
        print(self.context["request"].data)
        if images_from_files:
            for image in images_from_files:
                PostsImage.objects.create(post=post, image=image)
        if attached_files:
            for attached_file in attached_files:
                PostsAttachedFile.objects.create(post=post, attached_file=attached_file)
        return post

    def update(self, instance, validated_data):
        images = None
        attached_files = None
        if "posts_images" in validated_data.keys():
            images = validated_data.pop("posts_images")
        if "posts_attached_files" in validated_data.keys():
            attached_files = validated_data.pop("posts_attached_files")
        instance = super().update(instance, validated_data)
        if images and type(images) == list:
            instance.posts_images.all().delete()
            for image in images:
                PostsImage.objects.create(post=instance, **image)
        if attached_files and type(attached_files) == list:
            instance.posts_attached_files.all().delete()
            for attached_file in attached_files:
                PostsAttachedFile.objects.create(post=instance, **attached_file)
        return instance

视图集代码 #

class PostModelViewSet(viewsets.ModelViewSet):
    def get_queryset(self):
        return Post.objects.filter(author=self.request.user)

    serializer_class = PostSerializer
    permission_classes = [permissions.IsAuthenticated]

测试代码 #

class PostModelViewSetTestCase(TestCase):
    def setUp(self):
        self.user = User.objects.create(
            email="testuser@test.com", password="testpassword"
        )
        self.client = APIClient()
        self.client.force_authenticate(user=self.user)

    def test_create_post(self):
        tmp_image = SimpleUploadedFile(
            "file.jpg", b"file_content", content_type="image/jpg"
        )
        img_path = f"{CURRENT_DIR}/socialmedia/tests/default.png"
        image = SimpleUploadedFile(
            name="test_image.jpg",
            content=open(img_path, "rb").read(),
            content_type="image/jpeg",
        )
        data = {
            "author": self.user.id,
            "location_type": "local",
            "location": "nord",
            "location_radius": 5,
            "valid_until": datetime.now() + timedelta(days=1),
            "content": "content",
            "title": "title",
            "business_sector": "IT",
            "images": [
                {"image": image},
                {"image": image},
                {"image": image},
                {"image": image},
            ],
        }

        response = self.client.post("/socialmedia/posts/", data, format="multipart")
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
        self.assertEqual(Post.objects.count(), 1)
        print(PostsImage.objects.all())
        self.assertEqual(PostsImage.objects.count(), 4)
        post = Post.objects.first()
        self.assertEqual(post.author, self.user)
        self.assertEqual(post.posts_images.count(), 4)

问题排查过程 #

• 测试创建带图片的Post时，Post对象成功生成，但PostsImage未创建，测试失败。检查发现序列化器validate方法的attrs中缺少images或posts_images字段。
• 改用JSON格式请求时能看到posts_images字段，但无法处理文件类型；尝试从request.FILES获取图片，发现FILES为空。
• 进一步排查发现，multipart格式下图片数据被序列化为字符串，无法被序列化器正确解析识别。

解决方案 #

创建两个序列化器，分别适配不同请求场景：

class PostListSerializer(serializers.ModelSerializer):
    images = PostsImageSerializer(many=True, source="posts_images", required=False)
    attached_files = PostsAttachedFileSerializer(
        many=True, source="posts_attached_files", required=False
    )

    class Meta:
        model = Post
        fields = "__all__"


class PostSerializer(serializers.ModelSerializer):
    images = serializers.ListField(
        required=False, child=serializers.ImageField(use_url=True)
    )
# 此处省略原PostSerializer中validate、create、update等方法的实现

• PostListSerializer用于列表查询（list）和详情查询（retrieve）请求；
• PostSerializer用于创建（create）、更新（update）等涉及文件上传的请求。

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