在现代Web开发中,文件上传是常见需求。FastAPI作为高性能、易用性强的Python Web框架,提供了简洁高效的文件上传解决方案。本文将从基础到进阶,详细讲解如何使用FastAPI处理文件上传,并通过实例帮助你快速掌握核心技能。
一、基础准备:安装与配置¶
在开始前,请确保已安装必要的依赖:
pip install fastapi uvicorn
- FastAPI:核心框架,提供文件上传相关API。
- Uvicorn:ASGI服务器,用于运行FastAPI应用。
二、基础文件上传:单个文件¶
2.1 核心概念¶
FastAPI通过UploadFile类处理文件上传,它比传统bytes类型更强大,支持获取文件名、文件类型、文件大小等元数据。
2.2 示例代码¶
from fastapi import FastAPI, UploadFile, File
from fastapi.responses import JSONResponse
app = FastAPI(title="文件上传示例")
@app.post("/upload")
async def upload_single_file(file: UploadFile = File(...)):
"""
上传单个文件(使用UploadFile)
"""
try:
# 读取文件内容(适用于小文件)
file_content = await file.read()
# 保存文件到本地(示例:保存到项目目录下的uploads文件夹)
with open(f"uploads/{file.filename}", "wb") as f:
f.write(file_content)
return {
"status": "success",
"filename": file.filename,
"file_type": file.content_type,
"size": len(file_content)
}
except Exception as e:
return JSONResponse(status_code=500, content={"message": f"上传失败: {str(e)}"})
finally:
await file.close() # 确保文件关闭,释放资源
2.3 代码解释¶
UploadFile = File(...):声明文件参数,...表示必填项(Swagger文档会自动提示)。await file.read():异步读取文件内容到内存(小文件推荐,大文件建议流式处理)。file.filename:获取客户端上传的原始文件名。file.content_type:获取文件MIME类型(如image/png)。file.size:获取文件大小(字节数,FastAPI 0.90+版本支持)。
2.4 测试接口¶
运行应用:
uvicorn main:app --reload # main.py为文件名,app为FastAPI实例
访问Swagger UI(自动生成):http://localhost:8000/docs,在/upload接口中点击”Try it out”,选择本地文件即可测试。
三、进阶文件上传:多文件与复杂场景¶
3.1 多文件上传¶
FastAPI支持同时上传多个文件,只需将参数定义为List[UploadFile]:
@app.post("/upload-multiple")
async def upload_multiple_files(files: list[UploadFile] = File(...)):
"""
上传多个文件
"""
result = []
for file in files:
file_content = await file.read()
with open(f"uploads/{file.filename}", "wb") as f:
f.write(file_content)
result.append({
"filename": file.filename,
"size": len(file_content)
})
return {"files": result, "count": len(files)}
3.2 上传时携带额外表单数据¶
有时需要上传文件的同时传递其他信息(如用户ID、描述),可通过Form参数实现:
from fastapi import Form
@app.post("/upload-with-data")
async def upload_with_data(
user_id: int = Form(...), # 额外表单数据
description: str = Form(""), # 可选字段
file: UploadFile = File(...) # 文件
):
file_content = await file.read()
return {
"user_id": user_id,
"description": description,
"filename": file.filename
}
3.3 文件大小与类型验证¶
FastAPI允许通过条件判断限制文件大小或类型:
@app.post("/validate-file")
async def validate_file(file: UploadFile = File(...)):
# 限制文件大小:1MB(1024*1024字节)
if file.size > 1024 * 1024:
return JSONResponse(status_code=400, content={"message": "文件大小超过1MB"})
# 限制文件类型:仅允许图片
if file.content_type not in ["image/jpeg", "image/png"]:
return JSONResponse(status_code=400, content={"message": "仅支持JPG/PNG格式"})
# 其他处理逻辑...
return {"message": "文件验证通过"}
3.4 大文件流式处理¶
对于大文件(如视频、压缩包),直接读取到内存会导致内存溢出。FastAPI支持流式写入磁盘:
@app.post("/large-file")
async def large_file(file: UploadFile = File(...)):
try:
# 流式写入(异步读取,避免内存占用)
with open(f"uploads/large_{file.filename}", "wb") as f:
async for chunk in file.file: # 逐块读取
f.write(chunk)
return {"message": "大文件上传成功"}
finally:
await file.close()
四、实用技巧与最佳实践¶
4.1 文件路径管理¶
上传文件时建议使用固定目录(如uploads/),并确保目录存在:
import os
def ensure_dir(directory):
if not os.path.exists(directory):
os.makedirs(directory)
ensure_dir("uploads") # 启动时创建目录
4.2 自定义文件重命名¶
为避免文件名冲突(如用户上传相同名称文件),可自定义文件名:
import uuid
def get_unique_filename(original_name):
# 保留扩展名,生成唯一文件名
name, ext = os.path.splitext(original_name)
unique_name = f"{uuid.uuid4()}{ext}"
return unique_name
# 使用示例:
filename = get_unique_filename(file.filename)
with open(f"uploads/{filename}", "wb") as f:
f.write(await file.read())
4.3 错误处理¶
通过try-except捕获异常,提升健壮性:
@app.post("/safe-upload")
async def safe_upload(file: UploadFile = File(...)):
try:
if file.size > 5 * 1024 * 1024: # 5MB限制
raise ValueError("文件过大")
# 处理文件...
return {"status": "success"}
except ValueError as e:
return JSONResponse(status_code=400, content={"error": str(e)})
except Exception as e:
return JSONResponse(status_code=500, content={"error": "服务器错误"})
五、总结¶
通过本文,你已掌握FastAPI文件上传的核心技能:
- 基础:单个文件上传、使用UploadFile获取元数据。
- 进阶:多文件上传、表单数据与文件混合上传、流式处理大文件。
- 实用技巧:文件验证、路径管理、错误处理。
FastAPI的文件上传功能既简单又强大,结合自动生成的API文档(/docs),可快速构建稳定可靠的文件上传服务。
提示:生产环境中建议使用专业存储服务(如MinIO、S3)替代本地存储,避免服务器磁盘压力过大。