如何在Google App Engine中使用Python Flask将文件上传至Google Cloud Storage

问题背景 #

• 运行环境：Google App Engine Python 3.8 + Flask框架
• 需求：将用户上传的文件先存储到App Engine临时目录/tmp，再转存到Google Cloud Storage(GCS)
• 配置信息：
  app.yaml配置：

runtime: python38

requirements.txt依赖：

Flask==2.0.0;
google-cloud
google-cloud-storage

报错信息 #

OSError: [Errno 30] Read-only file system: 'file.xlsx'

尝试多种上传方案均复现相同报错。

根因分析 #

App Engine的文件系统除/tmp目录为临时可写外，其余路径全部为只读权限。本次报错是因为临时目录配置时遗漏了尾部斜杠，错误使用/tmp而非/tmp/，拼接文件路径时生成的路径为/tmpfile.xlsx，落到了只读的根目录，触发权限报错。

可正常运行的完整代码 #

from flask import Flask, request
from werkzeug.utils import secure_filename
from google.cloud import storage
import os

app = Flask(__name__)
# 注意末尾必须加斜杠
app.config['UPLOAD_FOLDER'] = "/tmp/"

@app.route('/')
def index():
    page = '''
    <form action="/upload" method="post" enctype="multipart/form-data">
        <input name="file" class="custom-file-input" type="file">
        <button type="submit" class="btn btn-success">提交</button>
    </form>
    '''
    return page

@app.route('/upload',methods=["POST"])
def upload():
    try:
        if request.method == 'POST':
            f = request.files['file']
            f.save(os.path.join(app.config['UPLOAD_FOLDER'],secure_filename(f.filename)))
            
            # 核心业务逻辑开始
            TMP_FILE_NAME = os.path.join(app.config['UPLOAD_FOLDER'],secure_filename(f.filename))
            # 替换为你自己的GCS存储桶名称
            BUCKET_NAME = 'fsample123'
            result = upload_blob(BUCKET_NAME, TMP_FILE_NAME, f"tmp/{secure_filename(f.filename)}")
            # 核心业务逻辑结束
            return result
            
    except Exception as e:
        return str(e)

def upload_blob(bucket_name, source_file_name, destination_blob_name):
    """上传文件到GCS存储桶"""
    # 参数说明
    # bucket_name: GCS存储桶ID
    # source_file_name: 本地待上传文件路径
    # destination_blob_name: GCS上的对象存储路径
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)
    blob.upload_from_filename(source_file_name)
    return "文件 {} 已上传到 {}。".format(source_file_name, destination_blob_name)

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

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