前提需要安装python-multipart,是处理表单数据(如文件上传)所必需的。

单文件上传

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    # 可选:检查文件类型、大小等
    print(f"文件名: {file.filename}")
    print(f"内容类型: {file.content_type}")

    # 保存文件到本地
    os.makedirs("uploads", exist_ok=True)
    file_path = f"uploads/{file.filename}"
    with open(file_path, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)

    return {"filename": file.filename, "message": "文件上传成功"}

多文件上传

@app.post("/upload-multiple/")
async def upload_multiple_files(files: list[UploadFile] = File(...)):
    saved_files = []
    os.makedirs("uploads", exist_ok=True)
    for file in files:
        file_path = f"uploads/{file.filename}"
        with open(file_path, "wb") as buffer:
            shutil.copyfileobj(file.file, buffer)
        saved_files.append(file.filename)
    return {"filenames": saved_files, "message": "多个文件上传成功"}

上传安全

MAX_FILE_SIZE = 5 * 1024 * 1024  # 5MB
ALLOWED_EXTENSIONS = {".jpg", ".png", ".pdf"}

@app.post("/upload-safe/")
async def upload_safe(file: UploadFile = File(...)):
    # 检查文件扩展名
    ext = os.path.splitext(file.filename)[1].lower()
    if ext not in ALLOWED_EXTENSIONS:
        raise HTTPException(status_code=400, detail="不支持的文件类型")

    # 检查文件大小
    file.file.seek(0, 2)  # 移动到文件末尾
    size = file.file.tell()  # 获取大小
    file.file.seek(0)  # 重置指针
    if size > MAX_FILE_SIZE:
        raise HTTPException(status_code=400, detail="文件太大")

    # 安全保存
    safe_filename = f"{uuid.uuid4()}{ext}"
    file_path = f"uploads/{safe_filename}"
    with open(file_path, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)

    return {"filename": safe_filename}

大文件上传

@app.post("/upload-large/")
async def upload_large_file(file: UploadFile = File(...)):
    os.makedirs("uploads", exist_ok=True)
    # 分块写入,避免内存爆炸
    with open(f"uploads/{file.filename}", "wb") as f:
        while chunk := await file.read(1024 * 1024):  # 每次读 1MB
            f.write(chunk)
    return {"filename": file.filename, "status": "uploaded"}