使用celery任务上传文件到s3

Question

我正在尝试将视频文件上传到 s3，但在使用 celery 放入任务队列后。当视频上传时，用户可以做其他事情。

我的views.py调用celery任务

def upload_blob(request, iterator, interview_id, candidate_id, question_id):
    try:
        interview_obj = Interview.objects.get(id=interview_id)
    except ObjectDoesNotExist:
        interview_obj = None
    current_interview = interview_obj
    if request.method == 'POST':
        print("inside POST")
        # newdoc1 = Document(upload=request.FILES['uploaded_video'], name="videos/interview_"+interview_id+"_candidate_"+candidate_id+"_question_"+question_id)
        # newdoc1.save()
        save_document_model.delay(request.FILES['uploaded_video'],"videos/interview_"+interview_id+"_candidate_"+candidate_id+"_question_"+question_id)
        # newdoc2 = Document(upload=request.FILES['uploaded_audio'], name="audios/interview_"+interview_id+"_candidate_"+candidate_id+"_question_"+question_id)
        # newdoc2.save()
        save_document_model.delay(request.FILES['uploaded_audio'],"audios/interview_"+interview_id+"_candidate_"+candidate_id+"_question_"+question_id)
        iterator = str(int(iterator) + 1)

        return HttpResponseRedirect(reverse('candidate:show_question', kwargs={'iterator': iterator,'interview_id':current_interview.id,'question_id':question_id}))
    else:

        return render(request, 'candidate/record_answer.html')

实际的芹菜任务.py

@task(name="save_document_model")
def save_document_model(uploaded_file, file_name):

    newdoc = Document(upload=uploaded_file, name=file_name)
    newdoc.save()

    logger.info("document saved successfully")
    return HttpResponse("document saved successfully")

文档模型

def upload_function(instance, filename):
    getname = instance.name
    customlocation = os.path.join(settings.AWS_S3_CUSTOM_DOMAIN, settings.MEDIAFILES_LOCATION, getname)
    # Add other filename logic here
    return getname # Return the end filename where you want it saved.

class Document(models.Model):
    name = models.CharField(max_length=25)
    uploaded_at = models.DateTimeField(auto_now_add=True)
    upload = models.FileField(upload_to=upload_function)

设置.py

AWS_ACCESS_KEY_ID = '**********************'
AWS_SECRET_ACCESS_KEY = '**************************'
AWS_STORAGE_BUCKET_NAME = '*********'
AWS_S3_CUSTOM_DOMAIN = '%s.s3.amazonaws.com' % AWS_STORAGE_BUCKET_NAME
AWS_S3_OBJECT_PARAMETERS = {
    'CacheControl': 'max-age=86400',
}
AWS_LOCATION = 'static'
AWS_DEFAULT_ACL = None

MEDIAFILES_LOCATION = 'uploads/'
DEFAULT_FILE_STORAGE = 'watsonproj.storage_backends.MediaStorage'

# CELERY STUFF
BROKER_URL = 'redis://localhost:6379'
CELERY_RESULT_BACKEND = 'redis://localhost:6379'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Africa/Nairobi'
CELERY_IMPORTS=("candidate.tasks")

直接上传可以在没有芹菜的情况下工作，但是使用芹菜我收到此错误：

“InMemoryUploadedFile”类型的对象不可 JSON 序列化

Answer 1

Celery 提供了配置任务负载序列化方式的选项。

在项目设置集中配置的任务序列化程序CELERY_TASK_SERIALIZER = json。

request.FILES['<input>']是 的实例django.core.files.uploaded.files.InMemoryUploadedFile，不能直接使用json序列化器进行编码（支持类型列表）。
虽然有多种方法可以将文件序列化为二进制数据，但如果您的用户上传大型文件，您的应用程序有可能会耗尽大量内存

您可以考虑django.core.files.uploadedfile.TemporaryFileUploadHandler在任何情况下使用并转发临时文件路径 ( request.FILES['<input>'] .temporary_file_path()) 而不是request.FILES['<input>']在任务有效负载中。

要强制执行此操作，请FILE_UPLOAD_MAX_MEMORY_SIZE = 0在项目设置中进行配置。 警告：这会停用MemoryFileUploadHandler整个项目的 。

随后在任务定义中，您可以将文件读入内存以保存新的Document.

from django.core.files import File
from django.conf import DEFAULT_FILE_STORAGE as storage

@task(name="save_document_model")
def save_document_model(file_path, file_name):

    with open(file_path, 'r') as f:
        file = File(f)

        newdoc = Document(upload=file, name=file_name)
        newdoc.save()

        logger.info("document saved successfully")

        storage.delete(file_path) # cleanup temp file

    return HttpResponse("document saved successfully")