如何在FastAPI中用UploadFile验证大ZIP文件且避免完整处理多部分上传

要验证大ZIP文件且不完整接收整个文件，核心是只读取文件开头的固定标识字节——ZIP文件的前4字节是十六进制的50 4B 03 04，只要验证这部分就能判断是否为ZIP文件，不用等整个大文件上传完成。

实现步骤： #

1. 从UploadFile异步读取前4字节获取文件头
2. 对比文件头是否符合ZIP格式的固定标识
3. 验证通过则继续保存完整文件，否则直接返回错误中断上传

完整代码示例： #

from fastapi import FastAPI, File, UploadFile, HTTPException
import os

app = FastAPI()

UPLOAD_FOLDER = 'uploads/'

# 确保上传目录存在
if not os.path.exists(UPLOAD_FOLDER):
    os.makedirs(UPLOAD_FOLDER)

# ZIP文件的固定文件头（十六进制转字节）
ZIP_FILE_HEADER = b'\x50\x4b\x03\x04'

async def is_zip_file(file: UploadFile) -> bool:
    # 只读取前4字节验证文件头
    header = await file.read(4)
    # 读取后重置文件指针到开头，避免后续保存丢失内容
    await file.seek(0)
    return header == ZIP_FILE_HEADER

@app.post("/upload-zip/")
async def upload_zip_file(file: UploadFile = File(...)):
    # 验证是否为ZIP文件
    if not await is_zip_file(file):
        raise HTTPException(status_code=400, detail="文件不是ZIP文件")
    
    # 分块保存大文件，避免内存占用过高
    file_location = os.path.join(UPLOAD_FOLDER, file.filename)
    with open(file_location, "wb") as buffer:
        while chunk := await file.read(1024*1024):  # 每次读取1MB
            buffer.write(chunk)

    return {"info": f"文件 '{file.filename}' 上传成功"}

关键细节说明： #

• 避免完整上传：仅读取前4字节，哪怕是GB级大文件，这一步也能瞬间完成，不会占用过多带宽和内存
• 重置文件指针：读取文件头后必须用await file.seek(0)把指针移回开头，否则后续保存会从第5字节开始写入，导致文件损坏
• 分块保存：用循环分块读取大文件，避免一次性将整个文件加载到内存中

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