Flask与Celery文件上传报错：无法序列化FileStorage/BufferedRandom对象

直接把Flask的FileStorage对象传给Celery任务会触发序列化错误，核心原因：

• JSON序列化器无法识别FileStorage自定义类型
• 即便改用pickle，FileStorage内部包含的_io.BufferedRandom文件流是操作系统级别的资源，同样无法被序列化

以下是两种可行的解决方案：

方案一：先存临时文件，传递路径给任务 #

在视图层先把上传文件保存到临时目录，再将临时路径和文件名传给Celery任务处理。

修改后的代码：

import os
from flask import request, jsonify
from celery import shared_task
from werkzeug.utils import secure_filename

@flask_app.post('/upload')
def upload_post():
    uploaded_file = request.files['file']
    if uploaded_file.filename != '':
        filename = secure_filename(uploaded_file.filename)
        # 先在本地保存临时文件
        temp_path = os.path.join('tmp/', filename)
        uploaded_file.save(temp_path)
        
        # 传递临时路径和文件名给任务
        result = upload_file_task.delay(temp_path, filename)
        return {"result_id": result.id}
    else:
        return jsonify({'status':'No file'})

@shared_task(bind=True, ignore_result=False)
def upload_file_task(self, temp_path, filename):
    try:
        # 示例：将临时文件移到最终存储目录
        final_path = os.path.join('uploads/', filename)
        os.rename(temp_path, final_path)
        return 'File uploaded successfully'
    except Exception as exc:
        # 出错时清理临时文件，避免磁盘占用
        if os.path.exists(temp_path):
            os.remove(temp_path)
        raise self.retry(exc=exc)

方案二：读取文件字节流，传递内容给任务 #

如果不想生成临时文件，可以直接读取文件内容为字节流，传递给任务后再写入磁盘。

修改后的代码：

import os
from flask import request, jsonify
from celery import shared_task
from werkzeug.utils import secure_filename

@flask_app.post('/upload')
def upload_post():
    uploaded_file = request.files['file']
    if uploaded_file.filename != '':
        filename = secure_filename(uploaded_file.filename)
        # 读取文件内容为字节流
        file_content = uploaded_file.read()
        
        # 传递文件名和字节内容给任务
        result = upload_file_task.delay(filename, file_content)
        return {"result_id": result.id}
    else:
        return jsonify({'status':'No file'})

@shared_task(bind=True, ignore_result=False)
def upload_file_task(self, filename, file_content):
    try:
        save_path = os.path.join('uploads/', filename)
        with open(save_path, 'wb') as f:
            f.write(file_content)
        return 'File uploaded successfully'
    except Exception as exc:
        raise self.retry(exc=exc)

注意事项 #

• 大文件推荐用方案一，避免方案二的内存占用过高问题
• 确保Celery Worker进程有权限读写目标文件目录
• 方案一中的临时文件要做好异常清理，防止磁盘冗余

内容的提问来源于stack exchange，提问作者jwesonga