django - 使用 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')

实际 celery 任务.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")

直接上传在没有 celery 的情况下工作，但使用 celery 我收到此错误:

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

Answer 1

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

在您的项目设置中配置的任务序列化程序集 CELERY_TASK_SERIALIZER = json .
request.FILES['<input>']是 django.core.files.uploaded.files.InMemoryUploadedFile 的一个实例并且不能直接用 json 编码序列化程序 ( List of supported types )。
虽然有多种方法可以将文件序列化为二进制数据，但如果您的用户上传大文件，您的应用程序就有可能耗尽大量内存

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

To enforce this, configure FILE_UPLOAD_MAX_MEMORY_SIZE = 0 in your project settings. Caveat: This deactivates the MemoryFileUploadHandler for your entire project.

随后在任务定义中，您可以将文件读入内存以保存新的 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")