You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Flask API接收zip上传解压存储报ImmutableMultiDict错误排查

问题根因

你遇到的'ImmutableMultiDict' object has no attribute 'save'报错,核心原因是request.files是Flask维护的所有上传文件的键值对集合对象,本身不是单个文件对象,不存在save方法,必须通过上传时指定的字段名取出对应的FileStorage文件实例,才能执行保存、读取操作。
除此之外你的原代码还有几个会导致运行失败的问题:

  • 缺少zipfile模块的导入语句
  • 代码中使用了未定义的file_name变量
  • Content-Type校验逻辑和实际上传场景不匹配
  • 异常捕获粒度过粗,所有运行错误都统一返回"Invalid content type",无法定位真实故障点
可直接运行的实现方案

根据客户端上传zip的请求格式不同,分两种常见场景实现,两种方案都加入了zip slip路径遍历漏洞防护,避免恶意压缩包写入系统非预期目录。如果没有留存原始zip包的需求,可以直接从内存读取压缩包内容解压,不需要落盘临时文件,减少磁盘IO开销。

场景1:表单方式上传(multipart/form-data,最通用)

适用于客户端通过表单字段上传文件的场景,对应curl调用示例:curl -X POST -F "file=@your_local_file.zip" http://service-ip:port/invocations,此时请求的Content-Type为multipart/form-data,不是application/zip。

import os
import zipfile
from flask import Flask, request, Response

UPLOAD_FOLDER = os.path.dirname(os.path.realpath(__file__))
app = Flask(__name__)
# 按需调整单文件大小上限,这里设为100MB
app.config['MAX_CONTENT_LENGTH'] = 100 * 1024 * 1024

@app.route("/invocations", methods=["POST"])
def data_extraction():
    # 校验是否存在指定的上传字段
    if "file" not in request.files:
        return Response(response="Missing upload field: file", status=400)
    upload_file = request.files["file"]
    if upload_file.filename == "":
        return Response(response="Empty file uploaded", status=400)
    # 校验文件后缀
    if not upload_file.filename.lower().endswith(".zip"):
        return Response(response="Only .zip format is supported", status=400)
    
    try:
        # 直接从内存读取压缩包解压(推荐,无需存临时zip)
        with zipfile.ZipFile(upload_file, "r") as zip_ref:
            # 防zip slip路径遍历
            for info in zip_ref.infolist():
                if os.path.isabs(info.filename) or ".." in info.filename.split(os.sep):
                    return Response(response="Invalid file path in zip", status=400)
            zip_ref.extractall(UPLOAD_FOLDER)
        
        # 若需要留存原始zip文件,替换上面的解压逻辑为以下代码
        # zip_save_path = os.path.join(UPLOAD_FOLDER, upload_file.filename)
        # upload_file.save(zip_save_path)
        # with zipfile.ZipFile(zip_save_path, "r") as zip_ref:
        #     # 同样加路径校验
        #     for info in zip_ref.infolist():
        #         if os.path.isabs(info.filename) or ".." in info.filename.split(os.sep):
        #             return Response(response="Invalid file path in zip", status=400)
        #     zip_ref.extractall(UPLOAD_FOLDER)

        return f"Unzip success, files saved to {UPLOAD_FOLDER}"
    except zipfile.BadZipFile:
        return Response(response="Uploaded file is not a valid zip archive", status=400)
    except Exception as e:
        return Response(response=f"Process error: {str(e)}", status=500)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

场景2:二进制直传(Content-Type: application/zip)

适用于客户端直接把zip文件的二进制内容放到请求体中传输、不附加表单字段的场景,此时压缩包内容存在于request.data中,不需要从request.files取值。

import os
import io
import zipfile
from flask import Flask, request, Response

UPLOAD_FOLDER = os.path.dirname(os.path.realpath(__file__))
app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 100 * 1024 * 1024

@app.route("/invocations", methods=["POST"])
def data_extraction():
    if request.content_type != "application/zip":
        return Response(response="Content-Type must be application/zip", status=400)
    
    try:
        # 将请求体二进制转为类文件对象供zipfile读取
        zip_buffer = io.BytesIO(request.data)
        with zipfile.ZipFile(zip_buffer, "r") as zip_ref:
            # 防zip slip路径遍历
            for info in zip_ref.infolist():
                if os.path.isabs(info.filename) or ".." in info.filename.split(os.sep):
                    return Response(response="Invalid file path in zip", status=400)
            zip_ref.extractall(UPLOAD_FOLDER)
        return f"Unzip success, files saved to {UPLOAD_FOLDER}"
    except zipfile.BadZipFile:
        return Response(response="Uploaded content is not a valid zip archive", status=400)
    except Exception as e:
        return Response(response=f"Process error: {str(e)}", status=500)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)
补充说明
  • 不需要额外依赖第三方库,BytesIO是Python标准库自带的内存流工具
  • 生产环境建议不要用Flask自带的开发服务器跑服务,用gunicorn等WSGI容器部署
  • 解压前的路径校验必须加,否则存在任意文件写入的安全风险

内容的提问来源于stack exchange,提问作者B2A3R9C9A

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.30 02:54:20