前提需要安装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"}
评论已关闭