在构建 Web API 时,文件上传是一个绕不开的常见需求——用户头像、PDF 报告、CSV 批量导入等场景都需要服务端接收并处理文件。FastAPI 提供了 UploadFile 和 File 两个核心工具,配合 python-multipart 库即可轻松实现文件上传功能。本文从基础用法出发,逐步深入实际项目中的校验、存储与多文件处理。
一、环境准备
FastAPI 的文件上传依赖 python-multipart,需要先安装:
1
| pip install python-multipart
|
该库负责解析 multipart/form-data 格式的请求体,是 UploadFile 能够正常工作的前提。
二、单文件上传基础
FastAPI 通过 UploadFile 类型来表示上传的文件对象,使用 File() 作为参数声明的标记。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
| from fastapi import FastAPI, File, UploadFile
app = FastAPI()
@app.post("/upload") async def upload_file(file: UploadFile = File(...)): content = await file.read() with open(f"uploads/{file.filename}", "wb") as f: f.write(content) return { "filename": file.filename, "content_type": file.content_type, "size": len(content), }
|
UploadFile 对象的常用属性与方法:
| 属性/方法 |
说明 |
file.filename |
原始文件名(客户端上传时的文件名) |
file.content_type |
MIME 类型,如 image/png |
await file.read() |
读取文件全部字节 |
await file.write(path) |
将文件直接写入磁盘路径 |
file.file |
底层 Python 文件对象,可直接传给其他库 |
注意:UploadFile 的方法都是异步的,调用时必须使用 await。
三、UploadFile 与 bytes 的区别
FastAPI 也支持使用 bytes 接收文件,但两者有本质区别:
1 2 3 4 5 6 7 8 9 10 11
| @app.post("/upload1") async def upload_uploadfile(file: UploadFile = File(...)): data = await file.read() return {"size": len(data)}
@app.post("/upload2") async def upload_bytes(file: bytes = File(...)): return {"size": len(file)}
|
| 对比维度 |
UploadFile |
bytes |
| 内存占用 |
流式处理,小内存占用 |
全部读入内存 |
| 大文件支持 |
✅ 适合大文件 |
❌ 大文件会撑爆内存 |
| 文件属性 |
可获取 filename、content_type |
无法获取元信息 |
| 适用场景 |
通用场景,推荐首选 |
仅小文件(如 JSON 配置) |
结论:除非确认文件极小,否则一律使用 UploadFile。
四、多文件上传
FastAPI 支持同时上传多个文件,只需将参数类型声明为 List[UploadFile]:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18
| from typing import List from fastapi import FastAPI, File, UploadFile
app = FastAPI()
@app.post("/upload-multiple") async def upload_multiple(files: List[UploadFile] = File(...)): results = [] for file in files: file_path = f"uploads/{file.filename}" await file.write(file_path) results.append({ "filename": file.filename, "size": file.size, "content_type": file.content_type, }) return {"uploaded": len(results), "files": results}
|
也可以分别声明多个独立的文件参数,分别处理:
1 2 3 4 5 6 7 8
| @app.post("/upload-dual") async def upload_dual( avatar: UploadFile = File(...), resume: UploadFile = File(...), ): await avatar.write(f"uploads/avatars/{avatar.filename}") await resume.write(f"uploads/resumes/{resume.filename}") return {"avatar": avatar.filename, "resume": resume.filename}
|
五、文件校验
实际项目中需要对上传文件的大小和类型做校验。
文件大小限制
1 2 3 4 5 6 7 8 9 10 11
| MAX_SIZE = 5 * 1024 * 1024
@app.post("/upload-avatar") async def upload_avatar(file: UploadFile = File(...)): content = await file.read() if len(content) > MAX_SIZE: return {"error": "文件大小不能超过 5 MB"} with open(f"uploads/{file.filename}", "wb") as f: f.write(content) return {"filename": file.filename, "size": len(content)}
|
文件类型白名单校验
1 2 3 4 5 6 7 8 9
| ALLOWED_TYPES = {"image/jpeg", "image/png", "image/gif"}
@app.post("/upload-image") async def upload_image(file: UploadFile = File(...)): if file.content_type not in ALLOWED_TYPES: return {"error": f"不支持的文件类型: {file.content_type}"} await file.write(f"uploads/{file.filename}") return {"filename": file.filename}
|
使用 Pydantic 校验 + 依赖注入
更优雅的方式是将校验逻辑封装到依赖函数中,配合 HTTPException 返回标准错误:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
| from fastapi import Depends, HTTPException, status
async def validate_image(file: UploadFile = File(...)): if file.content_type not in ALLOWED_TYPES: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"不支持的文件类型: {file.content_type}", ) return file
@app.post("/upload-safe") async def upload_safe(file: UploadFile = Depends(validate_image)): await file.write(f"uploads/{file.filename}") return {"filename": file.filename}
|
六、与表单字段配合
文件上传通常伴随着额外的表单数据(如用户ID、描述等)。使用 Form() 声明即可:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
| from fastapi import FastAPI, File, Form, UploadFile
app = FastAPI()
@app.post("/upload-with-data") async def upload_with_data( file: UploadFile = File(...), user_id: int = Form(...), description: str = Form(""), ): await file.write(f"uploads/{user_id}/{file.filename}") return { "filename": file.filename, "user_id": user_id, "description": description, }
|
要点:File() 和 Form() 可以混合使用,它们都属于 multipart/form-data 编码。但不能与 Body()(JSON)混用在同一个接口中。
七、文件保存的最佳实践
生产环境中建议遵循以下原则:
1. 避免文件名冲突:使用 uuid 生成唯一文件名:
1 2 3 4 5 6 7 8 9 10
| import uuid from pathlib import Path
@app.post("/upload-unique") async def upload_unique(file: UploadFile = File(...)): ext = Path(file.filename).suffix safe_name = f"{uuid.uuid4()}{ext}" await file.write(f"uploads/{safe_name}") return {"stored_as": safe_name, "original": file.filename}
|
2. 按日期分目录存储:
1 2 3 4 5 6
| from datetime import date
today = date.today().isoformat() dir_path = Path(f"uploads/{today}") dir_path.mkdir(parents=True, exist_ok=True) await file.write(dir_path / safe_name)
|
3. 对接云存储:在实际项目中,文件通常不直接存本地,而是上传到 OSS / S3 等对象存储。核心思路是一致的——读取 UploadFile 的字节流,通过云存储 SDK 上传:
八、总结
FastAPI 的文件上传机制简洁而强大:
UploadFile 是处理文件上传的推荐方式,支持流式读写,内存安全;
List[UploadFile] 轻松搞定多文件上传;
- 文件大小和类型校验可以在路由函数内实现,也可以封装为依赖注入;
- 通过
Form() 可以同时接收文件和表单字段;
- 生产环境中注意文件名安全和存储路径规划,避免直接使用用户上传的原始文件名。
文件上传是 Web 开发的基础能力,掌握 FastAPI 的这些机制后,无论是本地存储还是对接云服务都能游刃有余。