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
相关产品推荐
相关产品推荐

