Python关联BackBlaze与MySQL遇TypeError错误，求协助解决

模块说明 #

• app.py：基础Flask服务，负责处理文件上传、MySQL数据存储
• storage.py：BackBlaze存储对接模块，生成文件下载链接

问题场景 #

想要实现：Flask接收上传文件后，将文件内容存入MySQL，同时把文件上传到BackBlaze并将生成的下载链接保存到MySQL，但运行时报错：

TypeError: object of type '_io.BytesIO' has no len()

错误原因 #

1. 文件指针耗尽：在app.py中先调用file.read()把文件内容存到数据库，此时文件对象的指针已经移到末尾，再传给backblazeload函数时，file.read()读不到任何数据
2. 参数类型错误：BackBlaze的upload_bytes方法需要直接传入字节数据，而不是BytesIO对象，这是触发len()错误的直接原因

修复方案 #

步骤1：保存文件内容副本 #

在app.py中读取文件内容后保存一份，避免后续读取空数据

步骤2：调整backblazeload参数 #

让函数直接接收字节数据和文件名，而非文件对象，同时返回生成的下载链接

步骤3：修正文件下载逻辑 #

send_file需要传入类文件对象，所以要把数据库中的二进制数据转为BytesIO

修改后的完整代码 #

app.py #

from io import BytesIO
from flask_migrate import Migrate
from flask import Flask, render_template, request, send_file
from flask_sqlalchemy import SQLAlchemy
from storage import backblazeload

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret key'
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:password@localhost/fof'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
db = SQLAlchemy(app)
migrate = Migrate(app, db)


class Trades(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    filename = db.Column(db.String(50))
    data = db.Column(db.LargeBinary)
    link = db.Column(db.String(200))


@app.route('/', methods=['GET', 'POST'])
def index():
    if request.method == 'POST':
        file = request.files['file']
        # 读取文件内容并保留副本
        file_content = file.read()
        upload = Trades(filename=file.filename, data=file_content)
        db.session.add(upload)
        db.session.commit()

        # 传入字节数据和文件名到BackBlaze上传函数
        download_link = backblazeload(file_content, file.filename)
        upload.link = download_link
        db.session.commit()

        return f'上传成功: {file.filename}'
    return render_template('index.html')


@app.route('/download/<upload_id>')
def download(upload_id):
    upload = Trades.query.filter_by(id=upload_id).first()
    # 将二进制数据转为BytesIO类文件对象
    return send_file(BytesIO(upload.data), attachment_filename=upload.filename, as_attachment=True)


if __name__ == '__main__':
    app.run(debug=True)

storage.py #

import os
from dotenv import load_dotenv
from b2sdk.v2 import InMemoryAccountInfo, B2Api


def backblazeload(file_content, filename):
    load_dotenv()

    info = InMemoryAccountInfo()
    b2_api = B2Api(info)

    # 从.env读取密钥，避免硬编码
    app_key_id = os.getenv("B2_APP_KEY_ID")
    app_key = os.getenv("B2_APP_KEY")
    b2_api.authorize_account("production", app_key_id, app_key)

    bucket = b2_api.get_bucket_by_name("pickabucketname")

    file_info = {'how': 'good-file'}

    # 直接传入字节数据到upload_bytes
    uploaded_info = bucket.upload_bytes(
        file_content,
        file_name=filename,
        file_infos=file_info
    )

    file_id = uploaded_info['fileId']
    download_link = b2_api.get_download_url_for_fileid(file_id)
    print(download_link)
    return download_link

额外优化提示 #

• 务必将BackBlaze的密钥存放在.env文件中，不要硬编码在代码里，防止泄露
• 可以添加异常捕获逻辑，比如处理Bucket不存在、上传失败等情况，让程序更稳定

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