Flask中的Celery任务,用于上传和调整图像大小并将其存储到Amazon S3

Question

我正在尝试创建一个芹菜任务,用于在将图像存储到Amazon S3之前上传和调整图像大小.但它没有按预期工作.没有任务,一切都运转正常.这是到目前为止的代码:

堆栈跟踪

Traceback (most recent call last):
  File "../myVE/lib/python2.7/site-packages/kombu/messaging.py", line 579, in _receive_callback
    decoded = None if on_m else message.decode()
  File "../myVE/lib/python2.7/site-packages/kombu/transport/base.py", line 147, in decode
    self.content_encoding, accept=self.accept)
  File "../myVE/lib/python2.7/site-packages/kombu/serialization.py", line 187, in decode
    return decode(data)
  File "../myVE/lib/python2.7/site-packages/kombu/serialization.py", line 74, in pickle_loads
    return load(BytesIO(s))
  File "../myVE/lib/python2.7/site-packages/werkzeug/datastructures.py", line 2595, in __getattr__
    return getattr(self.stream, name)
  File "../myVE/lib/python2.7/site-packages/werkzeug/datastructures.py", line 2595, in __getattr__
    return getattr(self.stream, name)
    ...
RuntimeError: maximum recursion depth exceeded while calling a Python object

views.py

from PIL import Image

from flask import Blueprint, redirect, render_template, request, url_for

from myapplication.forms import UploadForm
from myapplication.tasks import upload_task


main = Blueprint('main', __name__)

@main.route('/upload', methods=['GET', 'POST'])
def upload():
    form = UploadForm()
    if form.validate_on_submit():
        upload_task.delay(form.title.data, form.description.data,
                          Image.open(request.files['image']))
        return redirect(url_for('main.index'))
    return render_template('upload.html', form=form)

tasks.py

from StringIO import StringIO

from flask import current_app

from myapplication.extensions import celery, db
from myapplication.helpers import resize, s3_upload
from myapplication.models import MyObject


@celery.task(name='tasks.upload_task')
def upload_task(title, description, source):
    stream = StringIO()
    target = resize(source, current_app.config['SIZE'])
    target.save(stream, 'JPEG', quality=95)
    stream.seek(0)
    obj = MyObject(title=title, description=description, url=s3_upload(stream))
    db.session.add(obj)
    db.session.commit()

Answer 1

我知道这是一个非常古老的问题,但我正在努力将文件的内容传递给芹菜任务.我会不断尝试跟随其他人做的错误.所以我写了这篇文章,希望将来可以帮助其他人.

TL; DR

• 使用base64编码将文件内容发送到celery任务
• 解码芹菜任务中的数据并io.BytesIO用于流

答案很长

我没有兴趣将图像保存到磁盘并再次读取它,所以我想传递所需的数据以在后台重建文件.

试图按照别人的建议,我不断遇到编码错误.一些错误是:

• UnicodeDecodeError: 'utf-8' codec can't decode byte 0xff in position 0: invalid start byte
• TypeError: initial_value must be str or None, not bytes

在TypeError被抛出io.StringIO.试图解码数据以摆脱它UnicodeDecodeError没有多大意义.由于数据首先是二进制的,所以我尝试使用一个io.BytesIO实例,这非常有效.我唯一需要做的就是使用base64对文件的流进行编码,然后我就能将内容传递给celery任务.

代码示例

images.py

import base64

file_.stream.seek(0) # start from beginning of file
# some of the data may not be defined
data = {
  'stream': base64.b64encode(file_.read()),
  'name': file_.name,
  'filename': file_.filename,
  'content_type': file_.content_type,
  'content_length': file_.content_length,
  'headers': {header[0]: header[1] for header in file_.headers}
}

###
# add logic to sanitize required fields
###

# define the params for the upload (here I am using AWS S3)
bucket, s3_image_path = AWS_S3_BUCKET, AWS_S3_IMAGE_PATH
# import and call the background task
from async_tasks import upload_async_photo 
upload_async_photo.delay(
  data=data,
  image_path=s3_image_path,
  bucket=bucket)

async_tasks

import base64, io
from werkzeug.datastructures import FileStorage

@celery.task
def upload_async_photo(data, image_path, bucket):
    bucket = get_s3_bucket(bucket) # get bucket instance
    try:
        # decode the stream
        data['stream'] = base64.b64decode(data['stream'])
        # create a BytesIO instance
        # https://docs.python.org/3/library/io.html#binary-i-o
        data['stream'] = io.BytesIO(data['stream'])
        # create the file structure
        file_ = FileStorage(**data)
        # upload image
        bucket.put_object(
                Body=file_,
                Key=image_path,
                ContentType=data['content_type'])
    except Exception as e:
        print(str(e))

编辑

我还改变了芹菜接受的内容以及它如何序列化数据.为了避免在将Bytes实例传递给celery任务时遇到问题,我不得不将以下内容添加到我的配置中:

CELERY_ACCEPT_CONTENT = ['pickle']
CELERY_TASK_SERIALIZER = 'pickle'
CELERY_RESULT_SERIALIZER = 'pickle'

Answer 2

您似乎正在尝试将整个上传的文件作为Celery消息的一部分传递。我想这会给您带来麻烦。我建议您查看是否可以将文件作为视图的一部分保存到Web服务器，然后让消息（“延迟”参数）包含文件名而不是整个文件的数据。然后，任务可以从硬盘驱动器中读取文件，上传到s3，然后在本地将其删除。