我正在尝试创建一个 celery 任务，用于在将图像存储到 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()

我知道这是一个非常古老的问题，但我一直在努力将文件的内容传递给 celery 任务。当我试图遵循其他人所做的事情时，我会不断出错。所以我写了这篇文章，希望对以后的人有所帮助。

TL;DR

• 将文件内容以base64编码发送到celery任务
• 解码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

The TypeError被抛出io.StringIO。尝试解码数据以消除UnicodeDecodeError没有多大意义。由于数据首先是二进制的，我尝试使用io.BytesIO实例，效果非常好。我唯一需要做的就是使用 Base64 对文件流进行编码，然后我就可以将内容传递给 celery 任务。

代码示例

图片.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)

异步任务

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))

Edit

我还更改了 celery 接受的内容以及它序列化数据的方式。为了避免将 Bytes 实例传递给 celery 任务时出现问题，我必须将以下内容添加到我的配置中：

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