FastAPI上传zip文件未正确保存 仅生成1KB同名文件如何解决

问题根因 #

• 调用zipfile.is_zipfile(file.file)时，方法会读取文件流的头部字节判断格式，执行完成后文件指针已经移动到了文件流末尾，后续执行shutil.copyfileobj时从末尾开始读取，自然没有内容可以写入，最终只能生成空的/体积极小的无效文件。
• 其余潜在问题：
  1. 路径拼接直接用Windows格式的反斜杠\，在Linux/macOS环境下会路径错误
  2. 未判断目标存储目录FILES是否存在，目录不存在时写入会直接报错
  3. 文件格式校验不通过时还是返回了Status: True，前端无法感知错误
  4. 同步接口处理大文件时容易阻塞服务，异步写法性能更优

修复后的可运行代码 #

from fastapi import FastAPI, File, UploadFile, HTTPException
import os
import zipfile
import shutil
from pathlib import Path

app = FastAPI() 

# 提前定义存储目录，不存在则自动创建
BASE_DIR = Path(__file__).parent.parent.resolve()
UPLOAD_DIR = BASE_DIR / "FILES"
UPLOAD_DIR.mkdir(exist_ok=True, parents=True)

@app.post('/FileUpload/')
async def FileUpload(file: UploadFile = File(...)):
    dest_path = UPLOAD_DIR / file.filename
    # 先读取文件流判断格式
    file_content = await file.read()
    # 校验是否为zip
    if not zipfile.is_zipfile(file_content):
        print("Invalid File type")
        raise HTTPException(status_code=400, detail="仅支持上传zip格式文件")
    # 校验通过后写入文件
    try:
        with open(dest_path, "wb+") as f:
            f.write(file_content)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"文件保存失败：{str(e)}")
    finally:
        await file.close()
    return {"Status": True, "msg": "文件上传成功"}

补充说明 #

• 如果你要保留原有同步+shutil的写法，只需要在if zipfile.is_zipfile(file.file):之后加一行file.file.seek(0)重置文件指针到开头即可解决写入不全的问题。
• 如果需要处理100MB以上的超大zip文件，不建议一次性读取全部内容到内存，可以在is_zipfile判断通过后调用await file.seek(0)重置指针，再用异步分块写入的方式存储文件，降低内存占用。

内容的提问来源于stack exchange，提问作者Albert-J.K