Django中如何使用Celery上传文件

Question

我想知道如何使用 Celery 工作人员来处理文件上传。 所以我尝试在一个简单的类上实现它。 我覆盖了我的 ModelViewSet 中的创建类。 但显然 Django 的默认 json 编码器不会序列化 ImageFields (Lame)。 如果你们能告诉我如何解决这个问题，我将不胜感激。 这是我想出的：

序列化程序.py：

class ProductImageSerializer(serializers.ModelSerializer):
    class Meta:
        model = ProductImage
        fields = ['id', 'image']

任务.py：

from time import sleep
from celery import shared_task
from .models import ProductImage

@shared_task:
def upload_image(product_id, image):
    print('Uploading image...')
    sleep(10)
    product = ProductImage(product_id=product_id, image=image)
    product.save()

视图.py：

class ProductImageViewSet(ModelViewSet):
    serializer_class = ProductImageSerializer

    def get_queryset(self):
        return ProductImage.objects.filter(product_id=self.kwargs['product_pk'])

    def create(self, request, *args, **kwargs):
        product_id = self.kwargs['product_pk']
        image = self.request.FILES['image']
        image.open()
        image_data = Image.open(image)
        upload_image.delay(product_id, image_data)

        return Response('Thanks')

这是包含我的 ImageField 的我的模型：

class ProductImage(models.Model):
    product = models.ForeignKey(Product, on_delete=models.CASCADE, related_name='images')
    image = models.ImageField(upload_to='store/images', validators=[validate_image_size])

Answer 1

所以我想出了一种方法来做到这一点。 这是我的解决方案：

问题是 celery 的默认 json 编码器无法序列化 Images、InMemoryUploadedFile、ModelObjects 和... 所以我们需要传递一个可以 json 序列化的值。 在这种情况下，我们想要序列化一个图像。 所以我们可以做的是将我们的 Image 转换为字节，然后将字节对象转换为字符串，这样我们就可以将它发送到我们的 celery 任务。 在我们的任务中收到字符串后，我们可以将其转换回图像并使用 celery 上传。互联网上的许多人建议使用此解决方案，但没有人提供任何代码。 所以这里是上面例子的代码，如果你想看看它的实际效果：

在我的 views.py 中，我使用了 ModelViewSet 并覆盖了 create 方法：

def create(self, request, *args, **kwargs):

        image = self.request.FILES['image'].read()

        byte = base64.b64encode(image)
        
        data = {
            'product_id': self.kwargs['product_pk'],
            'image': byte.decode('utf-8'),
            "name": self.request.FILES['image'].name
        }

        upload_image.delay(data=data)

        return Response('Uploading...')

这是我的tasks.py：

from time import sleep
from celery import shared_task
from .models import ProductImage
import PIL.Image as Image
import io
import base64
import os
from django.core.files import File

@shared_task
def upload_image(data):
    
    print('Uploading image...')
    
    sleep(10)
    
    product_id = data['product_id']

    byte_data = data['image'].encode(encoding='utf-8')
    b = base64.b64decode(byte_data)
    img = Image.open(io.BytesIO(b))
    img.save(data['name'], format=img.format)
    
    with open(data['name'], 'rb') as file:
        picture = File(file)

        instance = ProductImage(product_id=product_id, image=picture)
        instance.save()
    
    os.remove(data['name'])

    print('Uploaded!')

我希望有人觉得这很有帮助。 任何人有任何建议请在 cmets 中告诉我。 祝你今天过得愉快;）

Answer 2

大家好，我之前发布了这个问题的解决方案，即使该解决方案运行正常，我也找到了更好的解决方案。 使用 base64 编码和解码二进制文件会使它们变大，这不是我们想要的。所以更好的解决办法是把上传的文件临时保存在磁盘上，把路径传给我们的celery worker上传，然后在我们的数据库中创建一个ProductImage实例，然后删除我们保存在磁盘上的文件。

以下是如何实现它：

任务.py：

from time import sleep
from celery import shared_task
from .models import ProductImage
from django.core.files import File
from django.core.files.storage import FileSystemStorage
from pathlib import Path

@shared_task
def upload(product_id, path, file_name):

    print('Uploading image...')

    sleep(10)
    
    storage = FileSystemStorage()

    path_object = Path(path)

    with path_object.open(mode='rb') as file:
        
        picture = File(file, name=path_object.name)

        instance = ProductImage(product_id=product_id, image=picture)

        instance.save()


    storage.delete(file_name)

    print('Uploaded!')

在 serializers.py 中，您应该像这样覆盖 ProductImage 序列化程序的 create 方法：

    def create(self, validated_data):
        product_id = self.context['product_id']
        image_file = self.context['image_file']
        storage = FileSystemStorage()
        
        storage.save(image_file.name, File(image_file))

        return upload.delay(product_id=product_id, path=storage.path(image_file.name), file_name=image_file.name)

您还应该覆盖 ProductImage 的 ViewSet 中的 create 方法，为您的序列化程序的上下文提供图像文件：

    def create(self, request, *args, **kwargs):
        product_id = self.kwargs['product_pk']
        image_file = self.request.FILES['image']
        serializer = ProductImageSerializer(
            data=request.data,
            context={
                'product_id': product_id,
                'image_file': image_file
            }
        )
        serializer.is_valid(raise_exception=True)
        serializer.save()
        return Response('Upload Started...')