首页 > 基础资料 博客日记
FastAPI 文件上传避坑全指南:分块存盘、类型校验与安全兜底
2026-04-27 12:30:02基础资料围观1次
你有没有遇到过这种情况:
用 FastAPI 写个文件上传接口,本地跑得好好的,一发到线上就各种 422、413,要不就是大文件一传内存直接爆炸?
今天咱们就来把这些坑一个个填平,聊聊 FastAPI 里表单和文件上传那些事儿,让你上线的时候心里有底。
📦 这篇文章能帮你解决什么?
- 普通表单字段怎么接, Form(...) 的正确打开方式
- 单文件和多文件上传的实战写法,以及异步读取的坑
- 文件大小限制怎么做才安全
- 小文件与大文件在内存处理上的本质区别,什么时候该落盘
🧩 第一部分:先搞懂表单数据怎么接
好,咱们先来最简单的场景。前端提交一个普通登录表单,用户名和密码。
很多人一上来就用 Form(...) ,但不知道为什么非要用它,不用行不行?
你可能会问:FastAPI 不是自己就能解析 JSON 吗?
对啊,但表单数据是 application/x-www-form-urlencoded 或 multipart/form-data ,不是 JSON。
你得明确告诉 FastAPI:这个字段从表单里拿,不是从路径参数或查询字符串里来。
from fastapi import FastAPI, Form
app = FastAPI()
@app.post("/login")
async def login(
username: str = Form(...),
password: str = Form(...)
):
return {"user": username}
注意那个 Form(...) 里的三个点,代表必填。如果你想给默认值,就直接 Form("guest") 。
可别偷懒用 Optional 加 None 又不设 Form 默认值,如果前端不传这个字段,直接 422,又要排查半天。
📁 第二部分:单文件上传,不止 UploadFile 那么简单
接下来重点来了,文件上传。
FastAPI 给了咱们 UploadFile ,这货比 Starlette 原生的 UploadFile 好用不少,自带异步接口。
from fastapi import FastAPI, UploadFile, File
@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
contents = await file.read()
return {"filename": file.filename, "size": len(contents)}
这里有个超容易翻车的点:就是 await file.read() 会把整个文件内容读进内存。
你要是传个几百兆的文件,内存当场就飙上去了。所以对于小文件(比如头像),这么做没问题,但要是一视同仁,没作区别判断,大文件这么来一下,那就是给服务器埋雷了。
再说个我踩过的坑:那就是文件读一次就没了。
你如果先 await file.read() 一次,再想读第二次时,你就拿不到东西了。要想复用,得先把内容存到变量里。
📚 第三部分:多文件上传,List 写法最省心
前端需要一次传多张图?直接把参数类型设置为 List[UploadFile] 就行,别自己手写循环拼装,那纯粹是给自己找活干。
from typing import List
from fastapi import FastAPI, UploadFile, File
@app.post("/upload-multiple")
async def upload_files(files: List[UploadFile] = File(...)):
for file in files:
content = await file.read()
# 依次处理每个文件
return {"uploaded": [f.filename for f in files]}
是不是以为这样就完了?还没完。
多文件上传时,如果某个文件出错,前面成功的文件要不要回滚?
怎么给前端返回精确的“第三个文件格式不对”这种错误?
这些都需要业务层自己设计好,FastAPI 只负责把文件对象给你。
🛡️ 第四部分:文件大小限制与安全性,别等出事了再想
官方文档里的确提到可以基于 request.headers 里的 Content-Length 做大小判断,但根据以往的经验,别完全依赖它。
客户端完全可以伪造这个头部,或者分块传输编码根本没有这个字段。
真正靠谱的做法是:
- 在网关层(Nginx)先限制一波 client_max_body_size
- 在 FastAPI 应用里通过中间件或依赖,对已上传大小做累计检查
- 读文件时别一次性全读,用 file.read(size) 分块读,边读边写磁盘
咱直接看代码。分块存盘的核心思路就一句话:别一口吃成胖子,拿个小碗,一勺一勺舀到磁盘里。
我习惯用 aiofiles 这个库来做异步文件写入,避免阻塞事件循环。先装一下:
uv add aiofiles
然后上代码,假设我们要把上传的文件分块存到服务器本地:
import os
import aiofiles
from fastapi import FastAPI, UploadFile, File, HTTPException
app = FastAPI()
CHUNK_SIZE = 1024 * 1024 # 每次读 1MB,根据服务器内存调
@app.post("/upload-chunked")
async def upload_chunked(file: UploadFile = File(...)):
# 生成一个安全的目标路径,这里简单用原文件名,生产环境务必改成 UUID
save_path = os.path.join("/tmp/uploads", file.filename)
os.makedirs(os.path.dirname(save_path), exist_ok=True)
try:
# 用 aiofiles 以异步写方式打开目标文件
async with aiofiles.open(save_path, 'wb') as out_file:
# 读第一块
chunk = await file.read(CHUNK_SIZE)
while chunk:
await out_file.write(chunk)
chunk = await file.read(CHUNK_SIZE)
except Exception as e:
# 出错了要清理掉不完整的文件,别留垃圾
if os.path.exists(save_path):
os.remove(save_path)
raise HTTPException(status_code=500, detail=f"File save failed: {e}")
return {
"filename": file.filename,
"saved_path": save_path
}
🎯 几个必须划重点的细节:
-
CHUNK_SIZE别设太大也别太小。设 1MB 或 2MB 是个比较稳妥的值,太大跟一次读完没区别,太小了磁盘 I/O 频繁反而慢。这是我实测过几次后的经验值。 -
一定要异步写。如果你用同步的
open()加write(),FastAPI 的主线程会被堵住,并发直接就跪了。aiofiles让整个过程保持在异步上下文里。 -
while chunk:这个循环会一直跑到读不到数据为止,这正是我们想要的“流式读取”。文件再大,内存里永远只保留当前这一小块。 -
异常处理里的清理 绝对不能省。上次我就偷懒没删残废文件,结果
/tmp塞满了几十个写到一半的垃圾,排查了半天才发现。 -
真实项目中,
save_path记得用uuid重命名,别直接用file.filename,防止路径穿越攻击。
如果你想在存盘的同时做一下大小限制检查,可以在循环里累加一个 total_size,一旦超过阈值就终止并抛异常:
MAX_SIZE = 50 * 1024 * 1024 # 50MB
total_size = 0
chunk = await file.read(CHUNK_SIZE)
while chunk:
total_size += len(chunk)
if total_size > MAX_SIZE:
# 注意:此时 out_file 已经写了一些数据,需要清理
await out_file.close()
os.remove(save_path)
raise HTTPException(status_code=413, detail="File too large")
await out_file.write(chunk)
chunk = await file.read(CHUNK_SIZE)
这样,不管多大的文件过来,你的内存都稳如老狗,磁盘也不会被撑爆。
最后啰嗦一句:上传文件一定要校验类型。
别光看扩展名,用 python-magic 或 filetype 库去读文件头,那种把 .exe 改成 .jpg 传上来的坏心思不能不防。
filetype 纯 Python 实现,不需要系统依赖,更轻量,咱就用它。uv add filetype安装一下即可!
这里单独抽一个校验函数,方便在接口里调用:
import filetype
# 只允许这些类型的图片上传
ALLOWED_MIME = {"image/jpeg", "image/png", "image/webp"}
# 文件头最少读这么多个字节就够判断了
MAGIC_BYTES_LEN = 261
def validate_file_type(file: UploadFile, allowed_mimes: set):
"""
读文件头部魔数来判断真实类型。
这里先只读头部,不消耗整个文件,后面还能接着 read。
"""
# 保证读取指针在开头,不然拿不到正确头部
file.file.seek(0)
head = file.file.read(MAGIC_BYTES_LEN)
# 读完头记得把指针复位,否则后续分块读或存盘读不到完整内容
file.file.seek(0)
kind = filetype.guess(head)
if kind is None:
raise HTTPException(
status_code=400,
detail=f"无法识别文件类型,文件头部不是已知格式"
)
if kind.mime not in allowed_mimes:
raise HTTPException(
status_code=400,
detail=f"不支持的文件类型: {kind.mime},只允许: {', '.join(allowed_mimes)}"
)
return kind
@app.post("/upload-safe")
async def upload_safe(file: UploadFile = File(...)):
# 先校验类型,不通过直接拒绝,不会在磁盘落任何东西
validate_file_type(file, ALLOWED_MIME)
# 下面才开始真正的存盘逻辑……
# (这里接你分块存盘或直接 read 的代码)
contents = await file.read()
return {"filename": file.filename, "size": len(contents)}
🎯 这个实现里也藏着几个容易翻车的小细节:
-
file.file.seek(0)这步 绝对不能漏。UploadFile.file是一个类文件对象,读了头部后指针就偏移了,不复位的话,后续读文件会缺前面这一截,导致存下来的文件损坏。这个坑我当初排查到凌晨三点。 -
MAGIC_BYTES_LEN我习惯设 261 字节,足以覆盖绝大多数格式的文件头。但像 Office 2007 那种 zip 包裹格式,判断会稍微复杂,一般业务用场景够用了。 -
校验时机选在 存盘动作之前,这样可以第一时间拒绝恶意文件,避免无效 I/O 和潜在的安全风险。
-
filetype.guess()通过文件头魔数判断,.exe改成.jpg会被揪出来,这是扩展名和 MIME 欺骗搞不定的。但这种判断不是 100% 防得住所有伪装,有条件再加一层病毒扫描是更安全的。
🧠 第五部分:内存缓存 vs 硬盘暂存,怎么选?
这个问题的本质其实是个取舍。对于头像、缩略图这种大概率不到 1MB 的文件,你完全可以 await file.read() 后直接丢对象存储,速度快代码也清爽。
但对于用户上传的附件、视频这类可能很大的文件,就得用 分块读取 + 临时文件流 的模式。
Starlette 的 UploadFile 有个属性叫 file ,它底层是 tempfile.SpooledTemporaryFile ,小文件就在内存里,超过一定阈值自动落盘。
但这个阈值默认是 1MB,你可以根据自己服务器内存情况调一下。
我的选择就简单粗暴:只要不是确定要转存的临时数据,一律流式写硬盘,免得到时候 OOM 了再后悔。
🎯 一句话总结:用 Form 接表单,用 UploadFile 接文件,用 List 处理多文件,小文件图方便,大文件走流式落盘,层层设限兜底安全。
这篇文章里每一个注意点,几乎都是反复Debug排障后换来的。如果你看完觉得有用,记得收藏一下,免得下次写文件上传接口的时候又掉进同一个坑里。
顺便点个赞,加个关注,让我知道这样的经验总结对你有帮助,以后我就有动力继续把其他的踩坑经历都倒出来。
有什么别的问题,或者你也有更巧妙的方法,直接在后台留言,咱们一起聊聊。毕竟,天下码农是一家,坑不踩过怎么长大呢? 😉
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:jacktools123@163.com进行投诉反馈,一经查实,立即删除!
标签:
相关文章
最新发布
- Redis-Hash型与List型操作命令
- 8 年前的老代码 + 20 刀 AI token = 我的第一款独立产品
- 深度学习进阶(十二)可变形池化 deformable RS RoI Pooling
- 计算与判定:P、NP、NP-hard 和 NP-complete 问题
- 20253904 2025-2026-2 《网络攻防实践》第六周作业
- Qwen3.6-27B 等九款本地模型的测试结果
- 在线学习算力平台推荐-Hyper.AI
- 有监督 vs 全自主:两种 Agent 范式,你选对了吗?
- windows装gitlab服务器
- 广数 CNC 跨平台采集:状态、坐标、刀补、宏变量等读取,支持980tdi、986gs、988ta、980tb3i、980mdi、980tc3、980tdc等等

