FastAPI 怎么处理文件上传下载？UploadFile 和 bytes 有什么区别？
简化版
FastAPI 接收文件有两种声明方式，区别很关键：file: bytes = File() 会把整个文件读进内存（一个 100MB 的文件就占 100MB 内存，几个并发就 OOM）；file: UploadFile 拿到的是一个包装了 SpooledTemporaryFile 的对象——小文件留在内存、超过阈值（默认 1MB）自动落到磁盘临时文件，还提供了 filename、content_type 和 async 的读写方法（await file.read()、await file.seek(0)）。所以除了「确定很小的文件」，一律用 UploadFile。安全上和其他框架一样有五道防线：文件名不可信（用 UUID 自己生成，别信 file.filename）、大小要限（FastAPI 没有内置的 MAX_CONTENT_LENGTH，要靠 Nginx 的 client_max_body_size 或自己在中间件里检查 Content-Length）、类型查魔数而不是信 content_type（那是客户端声明的）、存到 Web 根目录之外、下发前校验权限。下载侧：FileResponse 发送文件（内部是流式的，支持 Range 断点续传）、StreamingResponse 发送生成的内容、受保护文件的最佳方案是 X-Accel-Redirect 交给 Nginx（应用只做鉴权，传输零拷贝）。最重要的架构判断：大文件不该经过应用服务器——用对象存储的预签名 URL 让客户端直传，服务端只签发凭证和接收回调，这样不占 worker、不占带宽、天然支持断点续传。核心记忆：UploadFile 而不是 bytes；文件名自己生成、类型查魔数；FastAPI 没有内置大小限制；大文件走预签名直传。
详细版
两种接收方式对比：
bytes = File() | UploadFile | |
|---|---|---|
| 内存 | 全部读进内存 | 超过 1MB 落磁盘 |
| 元数据 | ❌ 只有内容 | ✅ filename/content_type |
| 读取 | 已是 bytes | await file.read() |
| 大文件 | ❌ 会 OOM | ✅ |
| 适用 | 头像缩略图等确定很小的 | 默认选它 |
from fastapi import FastAPI, UploadFile, File, Form, HTTPException, Depends
from fastapi.responses import FileResponse, StreamingResponse
import uuid, os, shutil, aiofiles
# ① ★基本上传★
@app.post("/upload")
async def upload(file: UploadFile): # ★★默认写法★★
content = await file.read() # ★异步读★
await file.seek(0) # ★★读完要 seek 回去★★
return {"filename": file.filename,
"content_type": file.content_type,
"size": file.size} # ★0.100+ 有 size★
# ② ★多文件 + 表单字段混合★
@app.post("/upload-many")
async def upload_many(
files: list[UploadFile],
title: str = Form(...), # ★★同时有表单字段时要用 Form★★
tags: list[str] = Form([]),
):
for f in files: ...
# ③ ★★流式保存（大文件必须这样）★★
CHUNK = 1024 * 1024
async def save_upload(file: UploadFile, dest: str) -> int:
size = 0
async with aiofiles.open(dest, "wb") as out: # ★异步文件 IO★
while chunk := await file.read(CHUNK): # ★★分块读★★
size += len(chunk)
if size > MAX_SIZE: # ★★边读边校验大小★★
await out.close(); os.remove(dest)
raise HTTPException(413, "文件过大")
await out.write(chunk)
return size
# ✗ 反例：content = await file.read() → ★整个文件进内存★
# ④ ★安全校验★
from werkzeug.utils import secure_filename # 或自己实现
MAGIC = {b"\xff\xd8\xff": ".jpg", b"\x89PNG\r\n\x1a\n": ".png",
b"%PDF-": ".pdf", b"GIF8": ".gif"}
ALLOWED = {".jpg", ".jpeg", ".png", ".pdf", ".gif"}
async def validate(file: UploadFile) -> str:
ext = os.path.splitext(file.filename or "")[1].lower()
if ext not in ALLOWED:
raise HTTPException(400, "不支持的类型")
head = await file.read(16)
await file.seek(0) # ★★必须 seek 回去★★
if not any(head.startswith(m) for m in MAGIC): # ★★查魔数★★
raise HTTPException(400, "文件内容与扩展名不符")
return ext
@app.post("/upload-safe")
async def upload_safe(file: UploadFile, user: CurrentUser):
ext = await validate(file)
name = f"{uuid.uuid4().hex}{ext}" # ★★自己生成文件名★★
subdir = datetime.now().strftime("%Y/%m") # ★目录分片★
path = os.path.join(UPLOAD_ROOT, subdir, name) # ★Web 根目录之外★
os.makedirs(os.path.dirname(path), exist_ok=True)
size = await save_upload(file, path)
await Upload.create(path=f"{subdir}/{name}", size=size,
original_name=file.filename, user_id=user.id)
return {"id": ..., "name": name}
# ⑤ ★★请求体大小限制（FastAPI 没有内置！）★★
class LimitUploadSizeMiddleware: # ★纯 ASGI 中间件★
def __init__(self, app, max_size: int):
self.app, self.max_size = app, max_size
async def __call__(self, scope, receive, send):
if scope["type"] == "http":
headers = dict(scope["headers"])
cl = headers.get(b"content-length")
if cl and int(cl) > self.max_size: # ★★先看 Content-Length★★
await send({"type": "http.response.start", "status": 413,
"headers": [(b"content-type", b"application/json")]})
await send({"type": "http.response.body",
"body": b'{"detail":"文件过大"}'})
return
await self.app(scope, receive, send)
app.add_middleware(LimitUploadSizeMiddleware, max_size=10*1024*1024)
# ★★但真正的防线是 Nginx 的 client_max_body_size★★
# ⑥ ★下载★
@app.get("/files/{fid}")
async def download(fid: int, user: CurrentUser, db: DbDep):
up = await db.get(Upload, fid)
if not up or up.user_id != user.id: # ★★权限校验★★
raise HTTPException(404)
return FileResponse( # ★内部流式 + 支持 Range★
os.path.join(UPLOAD_ROOT, up.path),
filename=up.original_name, # ★★自动处理中文名★★
media_type=up.mime,
)
# ★★受保护文件的最佳方案：X-Accel-Redirect★★
@app.get("/files/{fid}/fast")
async def download_fast(fid: int, user: CurrentUser, db: DbDep):
up = await db.get(Upload, fid); check(up, user)
return Response(headers={
"X-Accel-Redirect": f"/protected/{up.path}", # ★Nginx internal location★
"Content-Disposition": f'attachment; filename="{quote(up.original_name)}"',
})⚠️ 三个必须记住的点：①
UploadFile而不是bytes。声明成file: bytes = File()时，FastAPI 会把整个文件读进内存再交给你——上传一个 200MB 的视频就是 200MB 常驻内存，几个并发请求直接 OOM。UploadFile底层是SpooledTemporaryFile：内容小于阈值（Starlette 默认 1MB）时留在内存，超过就自动写入磁盘临时文件，对使用者接口完全一致。而且它还提供了filename、content_type、size以及异步的read/write/seek/close。② FastAPI/Starlette 没有内置的请求体大小限制——不像 Flask 有MAX_CONTENT_LENGTH。这意味着默认情况下，任何人都能往你的上传接口发一个 10GB 的请求（虽然UploadFile会落盘不至于 OOM，但会打满磁盘）。三层防护：最外层用 Nginx 的client_max_body_size（超出直接 413，请求体根本不进应用）、中间层写一个检查Content-Length的 ASGI 中间件、内层在分块读取时累加实际字节数（因为Content-Length可以伪造，chunked 编码时甚至没有这个头）。③file.filename和file.content_type都是客户端提供的，完全不可信。filename可能包含../做路径穿越、可能是超长字符串、可能为空；content_type改一下就能让.php伪装成image/png。所以：文件名自己用 UUID 生成（原始名存数据库用于展示和下载时还原），类型靠读取文件头的魔数判断（读完记得await file.seek(0)，否则后续保存会得到空文件或残缺文件）。
完整版教学
一、UploadFile 的机制
★ ★UploadFile 的本质★：
class UploadFile:
file: BinaryIO # ★★SpooledTemporaryFile★★
filename: str | None # ★客户端提供，不可信★
size: int | None # ★0.100+★
headers: Headers
content_type: str | None
# ★异步方法（内部用 run_in_threadpool 包装同步 IO）★
async def read(self, size: int = -1) -> bytes
async def write(self, data: bytes) -> None
async def seek(self, offset: int) -> None
async def close(self) -> None★ ★SpooledTemporaryFile 的分级策略★：
┌──────────────────────────────────────────────────┐
│ 内容 ★< max_size（默认 1MB）★ → ★留在 BytesIO★ │
│ 内容 ★≥ max_size★ → ★自动 rollover★ │
│ ★写入磁盘临时文件★ │
└──────────────────────────────────────────────────┘
★ ★对使用者透明★：接口完全一样
★ ★所以上传 1GB 文件不会撑爆内存★
★ ★★但 await file.read() 会把整个文件读进内存★★：
content = await file.read() # ★★1GB → 1GB 内存★★
✓ ★分块读★：
while chunk := await file.read(1024 * 1024):
process(chunk)
✓ ★或用 shutil.copyfileobj（同步，要丢线程池）★：
await run_in_threadpool(shutil.copyfileobj, file.file, out, 1024*1024)
★ ★调整 spool 阈值★：
# Starlette 的 MultiPartParser
from starlette.formparsers import MultiPartParser
MultiPartParser.spool_max_size = 10 * 1024 * 1024 # ★★改成 10MB★★
★ ★权衡：阈值大 = 更多小文件走内存（快）但内存占用高★
★ ★async 方法其实是线程池包装★：
await file.read()
# 内部：await run_in_threadpool(self.file.read, size)
★ 因为 ★文件 IO 在 Linux 上没有真正的异步★（O_NONBLOCK 对普通文件无效）
★ ★所以它占用的是那 40 个线程池 token★
→ ★大量并发上传时线程池可能成为瓶颈★
★ ★临时文件的清理★：
★ Starlette 在请求结束后会 close 掉 UploadFile
→ SpooledTemporaryFile 的 close 会删除临时文件
★ ✗ 但如果你把 file 对象存到了别处（比如 BackgroundTasks 里用）
→ ★请求结束时文件已经被关闭/删除了★
✓ ★要在后台处理就先把内容复制出来★：
@app.post("/upload")
async def upload(file: UploadFile, bt: BackgroundTasks):
tmp = f"/tmp/{uuid4().hex}"
await save_upload(file, tmp) # ★★先落到自己的临时文件★★
bt.add_task(process_file, tmp) # 后台处理这个路径
return {"ok": True}
★ ★UploadFile 与 Form 混用★：
@app.post("/x")
async def x(file: UploadFile, title: str = ★Form(...)★):
★ ★注意：一旦有文件，其他字段必须声明成 Form 而不是普通参数或 Body★
★ 因为整个请求是 multipart/form-data
★ ✗ 不能同时有 File 和 Body（JSON）→ ★协议上互斥★
✓ 需要传复杂结构：把 JSON 作为一个 Form 字段传，然后手动解析
metadata: str = Form(...) → json.loads(metadata)
★ ★依赖形式的复用★：
async def validated_image(file: UploadFile) -> UploadFile:
await check_magic(file); return file
ImageDep = Annotated[UploadFile, Depends(validated_image)]
@app.post("/avatar")
async def avatar(file: ImageDep): ... # ★★校验逻辑复用★★UploadFile 的核心是 SpooledTemporaryFile 的分级策略——小于阈值（默认 1MB）留在内存、超过就自动写入磁盘临时文件，对使用者完全透明，所以上传 1GB 文件不会撑爆内存。但 await file.read() 会把整个文件读进内存，大文件必须分块读。有个容易忽略的细节：UploadFile 的 async 方法其实是 run_in_threadpool 包装的同步 IO（因为普通文件在 Linux 上没有真正的异步），所以它占用那 40 个线程池 token，大量并发上传时可能成为瓶颈。临时文件在请求结束后会被自动删除——所以想在 BackgroundTasks 里处理必须先把内容复制到自己的临时文件。还有个协议层面的限制：有文件时其他字段必须用 Form，而且不能同时有 File 和 JSON Body（需要复杂结构就把 JSON 当作一个 Form 字段传）。
二、上传的安全防线
★ ★★① 文件名：路径穿越★★
file.filename = "../../../etc/cron.d/evil"
✗ open(os.path.join(UPLOAD_DIR, file.filename), "wb")
→ ★写到了系统目录 → RCE★
✓ ★服务端自己生成★：
ext = os.path.splitext(file.filename or "")[1].lower()[:10] # ★限长★
if ext not in ALLOWED: raise HTTPException(400)
name = f"{uuid.uuid4().hex}{ext}"
★ 好处：★杜绝穿越 + 不会同名覆盖 + 中文名照常存数据库 + 不泄露原名★
★ ★★② 大小：FastAPI 没有内置限制★★
★三层防护（缺一不可）：★
┌────────────────────────────────────────────────────┐
│ ★① Nginx：client_max_body_size 10m★ │
│ → ★超出直接 413，请求体不进应用★（最有效） │
│ ★② ASGI 中间件：检查 Content-Length★ │
│ → ★快速拒绝，但 Content-Length 可伪造★ │
│ ★③ 读取时累加实际字节★ │
│ → ★最可靠（chunked 编码时前两层都可能失效）★ │
└────────────────────────────────────────────────────┘
★ ★为什么三层都要：★
- ★chunked 传输编码没有 Content-Length★
- ★恶意客户端可以声明小的 Content-Length 却发大量数据★
★ ★★③ 类型：查魔数★★
✗ 信 file.content_type # ★客户端声明的★
✗ 信扩展名 # ★改名就绕过★
✓ 读文件头：
head = await file.read(16); await file.seek(0) # ★★seek 回去★★
if not head.startswith(b"\x89PNG"): raise ...
✓ ★用 python-magic★：
import magic
mime = magic.from_buffer(await file.read(2048), mime=True)
await file.seek(0)
✓ ★★图片最彻底：重新编码★★
from PIL import Image
img = Image.open(file.file)
img.verify(); file.file.seek(0)
img = Image.open(file.file).convert("RGB")
img.thumbnail((2000, 2000)) # ★顺便限尺寸★
img.save(dest, "JPEG", quality=85)
★ → ★清除 EXIF（含 GPS）+ 销毁藏在图片里的载荷★
★ ★④ 存储位置★：
✗ ★存在 StaticFiles 挂载的目录下★
app.mount("/static", StaticFiles(directory="static"))
→ ★上传的文件可被直接 URL 访问 = 无权限控制★
✓ ★存 Web 根目录之外★（/var/data/uploads）
✓ ★或对象存储★
★ ★⑤ 下发前校验权限★：
@app.get("/files/{fid}")
async def get_file(fid: int, user: CurrentUser):
up = await get_upload(fid)
if up.user_id != user.id and not user.is_admin:
raise HTTPException(404) # ★★用 404 而不是 403（不泄露存在性）★★
return FileResponse(...)
★ ★其他要防的★：
□ ★图片炸弹★：10KB 的 PNG 解压成 10GB
Image.MAX_IMAGE_PIXELS = 50_000_000
□ ★SVG 是 XML，能内嵌 <script>★ → ★禁止或用 nh3 清洗★
□ ★ZipSlip★：解压时成员名可能是 ../../x
□ ★上传目录禁止执行★（Nginx 配置 + 文件权限）
□ ★病毒扫描★（ClamAV，企业场景）
□ ★用户配额★（总大小/数量）
★ ★目录分片（大量文件时必须）★：
✗ 所有文件塞一个目录 → ★几十万文件时 ls/stat 极慢★
✓ 按日期：uploads/2026/08/03/xxx.jpg
✓ 按 hash 前缀：uploads/ab/cd/abcdef...jpg上传安全的五道防线和其他框架一致，但 FastAPI 的特殊点是「没有内置的大小限制」——三层防护缺一不可：Nginx 的 client_max_body_size（最有效，超出直接 413）、ASGI 中间件检查 Content-Length（快速拒绝但可伪造）、读取时累加实际字节数（最可靠，因为 chunked 编码时前两层都可能失效）。文件名要自己用 UUID 生成（杜绝穿越、不会同名覆盖、中文名照常存数据库）。类型必须查魔数（读完记得 await file.seek(0)），图片最彻底的做法是用 Pillow 重新编码——清除 EXIF（含 GPS）并销毁藏在图片里的载荷。存储位置绝不能放在 StaticFiles 挂载的目录下（那等于没有权限控制）。下发时用 404 而不是 403（不泄露资源存在性）。
三、下载与文件下发
★ ★三个 Response 类★：
┌────────────────────┬──────────────────────────────────┐
│ ★FileResponse★ │ ★发送磁盘上的文件★ │
│ │ ✓ ★内部流式、支持 Range 断点续传★ │
│ │ ✓ ★自动设 Content-Type/Length★ │
│ │ ✓ ★用 anyio 的线程池读文件★ │
│ ★StreamingResponse★ │ ★发送生成的内容★ │
│ Response │ 小的内存内容 │
└────────────────────┴──────────────────────────────────┘
return FileResponse(path,
filename="报告.pdf", # ★★自动生成 RFC 5987 的中文文件名★★
media_type="application/pdf",
headers={"Cache-Control": "private, max-age=3600"})
★ filename 参数会生成：
Content-Disposition: attachment; filename*=utf-8''%E6%8A%A5%E5%91%8A.pdf
★ ★发送内存里的内容★：
buf = io.BytesIO(pdf_bytes)
return StreamingResponse(buf, media_type="application/pdf",
headers={"Content-Disposition": 'attachment; filename="x.pdf"'})
★ ★★生产环境：不要让应用发文件★★：
┌──────────────────┬────────────────────────────────┐
│ ★FastAPI 发★ │ ★占用 worker + 线程池、无零拷贝★│
│ ★Nginx 发★ │ ★sendfile 零拷贝、支持 Range★ │
│ ★对象存储/CDN★ │ ★最优：完全不经过应用★ │
└──────────────────┴────────────────────────────────┘
★★方案：X-Accel-Redirect（兼得鉴权和性能）★★
# FastAPI 只做权限校验
@app.get("/files/{fid}")
async def download(fid: int, user: CurrentUser):
up = await get_and_check(fid, user) # ★应用层鉴权★
return Response(headers={
"X-Accel-Redirect": f"/protected/{up.path}", # ★★Nginx 接管★★
"Content-Disposition":
f"attachment; filename*=utf-8''{quote(up.original_name)}",
"Content-Type": up.mime,
})
# nginx.conf
location /protected/ {
★internal;★ # ★★外部无法直接访问★★
alias /var/data/uploads/;
}
★ ★Apache 用 X-Sendfile★
★ ★预签名 URL（★云上最优★）★：
@app.get("/files/{fid}/url")
async def get_url(fid: int, user: CurrentUser):
up = await get_and_check(fid, user)
url = await s3.generate_presigned_url(
"get_object", Params={"Bucket": B, "Key": up.key},
ExpiresIn=300) # ★★5 分钟有效★★
return {"url": url}
★ ✓ ★传输完全不经过应用服务器 + 可走 CDN★
★ ★Range 请求（断点续传/视频拖动）★：
★ FileResponse ★自动支持★（返回 206 Partial Content）
★ StreamingResponse ★不自动支持★，要自己处理：
range_header = request.headers.get("range")
# bytes=0-1023 → 解析 → seek → 返回 206 + Content-Range
★ ★大文件导出（生成型）★：
@app.get("/export.csv")
async def export(db: DbDep):
async def gen():
yield "id,name\n"
result = await db.stream(select(User))
async for u in result.scalars():
yield f"{u.id},{u.name}\n"
return StreamingResponse(gen(), media_type="text/csv",
headers={"Content-Disposition": 'attachment; filename="export.csv"'})
★ ✗ ★注意 BaseHTTPMiddleware 会破坏流式★
★ ✓ 超大导出更应该：★异步任务生成文件 → 通知用户下载★下载侧有三个 Response 类：FileResponse 发磁盘文件（内部流式、自动支持 Range 断点续传、filename 参数会自动生成 RFC 5987 的中文文件名）、StreamingResponse 发生成的内容（不自动支持 Range）、Response 发小的内存内容。生产环境的核心判断是「不要让应用发文件」——受保护文件用 X-Accel-Redirect（应用只做鉴权、Nginx 零拷贝传输，location 要配 internal），云上最优是预签名 URL（传输完全不经过应用服务器，还能走 CDN）。超大导出更应该做成异步任务生成文件再通知用户下载，而不是在请求里流式生成（那会长时间占用连接）。
四、大文件与直传
★ ★★WSGI/ASGI 都逃不掉的问题★★：
上传期间★连接一直占着★
ASGI 下比 WSGI 好（不占整个 worker，占一个 task + 线程池 token）
★ 但仍然：
- ★带宽经过应用服务器★
- ★磁盘临时文件★
- ★线程池 token 被占（UploadFile 的 read 是线程池）★
★ ★★正解：对象存储预签名直传★★
┌──────────────────────────────────────────────────────┐
│ ① 前端 → 后端：★请求上传凭证★ │
│ ② 后端：★校验权限 + 生成预签名 POST/PUT★ │
│ （★带大小限制、类型限制、过期时间★） │
│ ③ 前端 → ★对象存储：直接上传（不经过后端）★ │
│ ④ 前端 → 后端：★通知上传完成★ │
│ ⑤ 后端：★head_object 确认 + 落库★ │
└──────────────────────────────────────────────────────┘
@app.post("/upload-token")
async def upload_token(req: UploadTokenIn, user: CurrentUser):
key = f"uploads/{user.id}/{uuid4().hex}{req.ext}"
presigned = s3.generate_presigned_post(
Bucket=BUCKET, Key=key,
Fields={"Content-Type": req.content_type},
Conditions=[
["content-length-range", 0, 10 * 1024 * 1024], # ★★大小限制★★
{"Content-Type": req.content_type}, # ★★类型限制★★
],
ExpiresIn=600)
return {"key": key, **presigned}
@app.post("/upload-callback")
async def callback(key: str, user: CurrentUser, db: DbDep):
if not key.startswith(f"uploads/{user.id}/"): # ★★校验归属★★
raise HTTPException(403)
obj = await s3.head_object(Bucket=BUCKET, Key=key) # ★★确认真实存在★★
await Upload.create(key=key, size=obj["ContentLength"],
mime=obj["ContentType"], user_id=user.id)
★ ★关键：回调必须服务端确认，不能信前端报的 size/type★
★ ★分片上传（超大文件）★：
★ 优先用对象存储的 multipart upload API：
① create_multipart_upload → upload_id
② 前端逐片上传（可并发、可重试单片）
③ complete_multipart_upload
★ 自己实现的话要处理：
- ★分片归属校验（防止覆盖别人的分片）★
- ★过期分片清理（定时任务）★
- ★合并时校验总大小和 hash★
- ★秒传：先用文件 hash 查库★
★ ★秒传（hash 去重）★：
@app.post("/check-hash")
async def check(sha256: str, user: CurrentUser):
existing = await Upload.get_by_hash(sha256)
if existing:
await Upload.create_ref(existing, user) # ★★只建引用★★
return {"instant": True, "id": ...}
return {"instant": False}
★ ✓ 省存储、省带宽、体验好
★ ★上传进度★：
★ ✗ 服务端很难提供（数据是一次性交给应用的）
✓ ★前端用 XMLHttpRequest 的 upload.onprogress★
✓ 或分片上传时按片数算
✓ ★直传到 S3 时，SDK 自带进度回调★
★ ★超时配置★：
# uvicorn / gunicorn
--timeout-keep-alive 75
# gunicorn + uvicorn worker
--timeout 300 # ★★大文件必须调大★★
# nginx
client_max_body_size 100m;
client_body_timeout 300s;
proxy_request_buffering on; # ★★先收完再转发，保护应用★★大文件的正解是「对象存储预签名直传」——后端只做两件事：签发带大小限制、类型限制、过期时间的凭证，以及在回调里 head_object 确认文件真实存在并落库。关键安全点是「回调必须服务端确认，不能信前端报的 size 和 type」，还要校验 key 的归属（防止用户伪造别人的路径）。超大文件用对象存储的 multipart upload API（可并发、可重试单片），自己实现的话要处理分片归属校验、过期清理、合并校验。秒传靠内容 hash 去重（只建引用）。上传进度应该由前端做（XMLHttpRequest.upload.onprogress），服务端很难提供。超时配置上，gunicorn 的 --timeout 必须调大，Nginx 的 proxy_request_buffering on（先收完再转发）能保护应用。
五、图片处理与实践
★ ★图片处理的典型流程★：
@app.post("/avatar")
async def upload_avatar(file: UploadFile, user: CurrentUser):
# ★① 快速校验（大小、扩展名）★
if file.size and file.size > 5 * 1024 * 1024:
raise HTTPException(413, "图片不能超过 5MB")
# ★② 魔数校验★
head = await file.read(16); await file.seek(0)
if not is_image(head): raise HTTPException(400)
# ★③ ★CPU 密集的处理丢线程池★★
result = await run_in_threadpool(process_image, file.file, user.id)
return result
def process_image(fp, user_id): # ★同步函数★
Image.MAX_IMAGE_PIXELS = 50_000_000 # ★★防图片炸弹★★
img = Image.open(fp)
img.verify(); fp.seek(0) # ★verify 后必须重新 open★
img = Image.open(fp).convert("RGB") # ★★丢弃 EXIF 和异常数据★★
# 生成多个尺寸
out = {}
for name, size in [("thumb", 128), ("medium", 512), ("large", 1024)]:
im = img.copy(); im.thumbnail((size, size))
path = f"{UPLOAD_ROOT}/{user_id}/{name}_{uuid4().hex}.jpg"
im.save(path, "JPEG", quality=85, optimize=True)
out[name] = path
return out
★ ★为什么要 run_in_threadpool★：
★图片处理是 CPU 密集的★，在 async def 里直接做会★卡住事件循环★
★ ★更重的处理（视频转码）→ 丢 Celery★
★ ★为什么重新编码是最好的安全措施★：
┌────────────────────────────────────────────────┐
│ ★清除 EXIF★（★可能含 GPS 位置、设备信息★） │
│ ★销毁图片马★（附加在文件尾部的 PHP/脚本代码） │
│ ★统一格式和尺寸★（省存储、省带宽） │
│ ★验证了它确实是一张能解码的图片★ │
└────────────────────────────────────────────────┘
★ ★后台处理的正确姿势★：
✗ @app.post("/upload")
async def upload(file: UploadFile, bt: BackgroundTasks):
bt.add_task(process, file) # ★★请求结束文件就没了★★
✓ 先落到自己的持久位置，再传路径：
path = await save_upload(file, f"/data/tmp/{uuid4().hex}")
bt.add_task(process, path)
✓ ★更重的活交给 Celery★（BackgroundTasks 仍占用当前进程）
★ ★清理与运维★：
□ ★孤儿文件清理★（数据库记录删了但文件还在 → 定时对账）
□ ★上传中断的临时文件清理★
□ ★用户配额检查（上传前）★
□ ★磁盘水位告警★
□ ★访问日志（谁下载了什么）★
★ ★测试文件上传★：
def test_upload(client):
files = {"file": ("test.png", io.BytesIO(PNG_BYTES), "image/png")}
r = client.post("/upload", files=files)
assert r.status_code == 200
def test_upload_wrong_magic(client):
files = {"file": ("evil.png", io.BytesIO(b"<?php ..."), "image/png")}
r = client.post("/upload", files=files)
assert r.status_code == 400 # ★★魔数校验生效★★
def test_upload_too_large(client):
files = {"file": ("big.png", io.BytesIO(b"x" * 20_000_000), "image/png")}
assert client.post("/upload", files=files).status_code == 413图片处理的关键是把 CPU 密集的部分丢线程池（run_in_threadpool）——在 async def 里直接用 Pillow 处理会卡住事件循环；更重的活（视频转码）要交给 Celery。重新编码是最好的安全措施——一次性完成「清除 EXIF（含 GPS）、销毁图片马、统一格式尺寸、验证确实是能解码的图片」四件事。后台处理有个必踩的坑：BackgroundTasks 里不能直接用 UploadFile 对象（请求结束时临时文件已被删除），必须先把内容保存到自己的持久位置再传路径。测试上传用 files={"file": (name, BytesIO(content), content_type)}，重点测魔数校验和大小限制真的生效。
六、实践清单
★ 检查清单：
【接收】
□ ★用 UploadFile 而不是 bytes★
□ ★大文件分块读，不用 await file.read() 一次读完★
□ ★读了内容后 await file.seek(0)★
【安全】
□ ★文件名自己生成（UUID），原名存数据库★
□ ★扩展名白名单★
□ ★魔数校验（不信 content_type）★
□ ★图片重新编码（清 EXIF 和载荷）★
□ ★Image.MAX_IMAGE_PIXELS 防图片炸弹★
□ ★禁止或清洗 SVG★
□ ★存 Web 根目录之外（不在 StaticFiles 目录下）★
□ ★下发前校验权限（用 404 不用 403）★
【大小】
□ ★Nginx client_max_body_size★
□ ★ASGI 中间件检查 Content-Length★
□ ★读取时累加实际字节★
【性能】
□ ★CPU 密集处理 run_in_threadpool 或 Celery★
□ ★下载用 X-Accel-Redirect 或预签名 URL★
□ ★大文件走对象存储直传★
□ ★gunicorn --timeout 调大★
【运维】
□ ★目录分片★
□ ★孤儿文件清理★
□ ★用户配额★
★ 报错速查：
┌────────────────────────────────────┬──────────────────┐
│ 保存出来是空文件 │ ★读了内容没 seek(0)★│
│ 内存暴涨/OOM │ ★用了 bytes 或一次★ │
│ │ ★性 read()★ │
│ 413（自己没配） │ ★Nginx 限制★ │
│ 大文件上传超时 │ ★gunicorn timeout★ │
│ BackgroundTasks 里文件不存在 │ ★临时文件已删除★ │
│ 中文文件名下载后乱码 │ ★没用 filename 参数★│
│ 上传时其他字段收不到 │ ★没声明成 Form★ │
└────────────────────────────────────┴──────────────────┘
★ 一句话总结：
★"用 UploadFile 而不是 bytes（前者超过 1MB 自动落盘）；
FastAPI 没有内置大小限制，要靠 Nginx + 中间件 + 读取时累加三层；
文件名自己生成、类型查魔数（读完 seek(0)）、存 Web 根目录之外；
下载用 X-Accel-Redirect 交给 Nginx，大文件走对象存储预签名直传。"★检查清单分四块。报错速查表里最高频的两条：「保存出来是空文件」就是读了内容没 seek(0)、「上传时其他字段收不到」是没声明成 Form。还有一个容易忽略的：BackgroundTasks 里文件不存在是因为临时文件在请求结束时已被删除。
记忆钩子：「FastAPI 接收文件有两种声明，★区别关键★：★
file: bytes = File()会把整个文件读进内存★（200MB 视频就是 200MB 常驻内存，几个并发直接 OOM）；★file: UploadFile底层是 SpooledTemporaryFile——小于阈值（默认 1MB）留内存、超过自动落磁盘临时文件★，还提供 filename/content_type/size 和异步的 read/write/seek。★所以除了确定很小的文件一律用 UploadFile★。但注意★await file.read()仍会把整个文件读进内存，大文件必须分块读★，而且★UploadFile 的 async 方法其实是 run_in_threadpool 包装的同步 IO（普通文件没有真异步），所以占那 40 个线程池 token★。★FastAPI/Starlette 没有内置的请求体大小限制★（不像 Flask 有 MAX_CONTENT_LENGTH）→ ★三层防护缺一不可：① Nginx 的 client_max_body_size（最有效，超出直接 413 请求体不进应用）② ASGI 中间件检查 Content-Length（快但可伪造）③ 读取时累加实际字节（最可靠，因为 chunked 编码时前两层都可能失效）★。安全五道防线：★① filename 和 content_type 都是客户端提供的完全不可信★——文件名可能含 ../ 做路径穿越，★所以自己用 UUID 生成、原名存数据库★；★② 类型查魔数，读完必须 await file.seek(0)★（否则保存出空文件，这是最高频的低级错误）；★③ 图片最彻底的做法是用 Pillow 重新编码★——一次性清除 EXIF（含 GPS）、销毁图片马、统一格式、验证确实能解码，还要设 ★Image.MAX_IMAGE_PIXELS 防图片炸弹★、★禁止或清洗 SVG（XML 能内嵌 script）★；★④ 存 Web 根目录之外，绝不能放在 StaticFiles 挂载的目录下★（那等于没有权限控制）；★⑤ 下发前校验权限，用 404 而不是 403（不泄露资源存在性）★。下载侧：★FileResponse 内部流式且自动支持 Range 断点续传，filename 参数会自动生成 RFC 5987 的中文文件名★；★StreamingResponse 不自动支持 Range★；★生产环境不要让应用发文件——受保护文件用 X-Accel-Redirect（应用只鉴权、Nginx 零拷贝，location 要配 internal），云上最优是预签名 URL★。★大文件的正解是对象存储预签名直传★：后端只签发★带 content-length-range 和类型限制的凭证★，★回调必须服务端 head_object 确认，不能信前端报的 size/type，还要校验 key 的归属★。两个易忽略的坑：★BackgroundTasks 里不能直接用 UploadFile 对象（请求结束时临时文件已被删除），要先保存到自己的位置再传路径★；★有文件时其他字段必须声明成 Form，且不能同时有 File 和 JSON Body（协议上互斥）★。★CPU 密集的图片处理要 run_in_threadpool，更重的丢 Celery★。」
七、常见误区与追问
- 误区：
file: bytes = File()和file: UploadFile差不多，前者用起来还更方便。 内存行为完全不同。声明成bytes时，FastAPI 会在调用你的函数之前就把整个文件读进内存，然后把bytes对象传给你——上传一个 200MB 的视频，进程内存立刻涨 200MB（而且因为 Python 的内存管理，释放后也未必立刻还给操作系统）；5 个并发上传就是 1GB，很容易 OOM。而UploadFile底层是SpooledTemporaryFile：内容小于阈值（Starlette 默认 1MB）时留在BytesIO里、超过就自动 rollover 到磁盘临时文件，对使用者来说接口完全一致，但内存占用恒定。此外UploadFile还给了你filename、content_type、size这些元数据和异步方法。所以规则是：除非你确定文件一定很小（比如限制了 100KB 的头像），否则一律用UploadFile。注意即使用了UploadFile，await file.read()不带参数仍然会把全部内容读进内存——大文件要分块读。 - 误区：FastAPI 会自动限制上传大小，不用额外配置。 它没有任何内置限制——这和 Flask 的
MAX_CONTENT_LENGTH不同，是很多人从 Flask 转过来后的盲区。默认情况下，任何人都可以往你的上传接口发一个 10GB 的请求：UploadFile会老老实实地把它写进磁盘临时文件，结果就是磁盘被打满（/tmp满了之后整个系统都会出问题）。需要三层防护：① Nginx 的client_max_body_size是最有效的一层，超出时 Nginx 直接返回 413，请求体根本不会转发给应用；② 一个检查Content-Length请求头的 ASGI 中间件，能在读取任何数据之前快速拒绝；③ 在分块读取时累加实际字节数，超过限制就中止并删除已写入的部分。为什么三层都要？因为Content-Length可以被伪造（声明 1MB 却发 1GB），而且 chunked 传输编码时根本没有这个头——只有第三层是真正可靠的。 - 误区：读了文件头做魔数校验之后，直接保存就行。 会保存出空文件或残缺文件——因为读取会移动文件指针。
head = await file.read(16)之后，指针停在第 16 字节；接下来如果你await file.read()保存，得到的是第 17 字节之后的内容（前 16 字节丢了）；如果之前已经用await file.read()读完了全部内容做校验，那么再读就是空的，保存出来是 0 字节文件。每次读取之后都必须await file.seek(0)把指针复位。这是文件上传最高频的低级错误，而且现象很迷惑：接口返回成功、数据库有记录、但文件打不开。同样的道理适用于用 Pillow 校验：img.verify()之后必须重新Image.open()（verify 会消耗掉文件对象），而且中间要seek(0)。 - 误区：
BackgroundTasks里可以直接处理上传的UploadFile。 请求结束时临时文件已经被删除了。Starlette 在响应发送完成后会关闭UploadFile，而SpooledTemporaryFile的close()会删除底层的临时文件。所以bt.add_task(process_image, file)这种写法，任务真正执行时拿到的是一个已关闭的文件对象——报ValueError: I/O operation on closed file或者读到空内容。正确做法是先把内容保存到你自己控制的持久位置（/data/tmp/{uuid}.tmp），然后把路径传给后台任务，任务处理完再删除。顺带说，即使解决了这个问题，BackgroundTasks仍然运行在当前的 Web 进程里——视频转码、大图批处理这类重活会占用应用资源、影响正常请求的响应，应该交给 Celery 这类外部队列。 - 误区：把上传的文件存到
StaticFiles挂载的目录下最方便，前端直接拿 URL 就能访问。 这等于放弃了所有权限控制。app.mount("/static", StaticFiles(directory="static"))之后，那个目录下的任何文件都能被直接 URL 访问——用户的身份证照片、合同、私密图片全部裸奔；如果文件名用了自增 id 还能被直接枚举。更严重的是执行风险：虽然StaticFiles本身不会执行脚本，但如果你的部署架构里 Nginx 也指向了同一个目录，配置疏漏就可能让上传的 webshell 被执行。正确做法是存到 Web 根目录之外（/var/data/uploads），通过一个做了权限校验的路由下发；更进一步用X-Accel-Redirect（应用只鉴权、Nginx 零拷贝传输）或对象存储的预签名 URL。只有确实完全公开且经过重新编码的资源（如公开头像）才适合放在可直接访问的目录。 - 追问：
UploadFile的异步方法是真异步吗？ 不是真异步，是线程池包装。await file.read()内部实际执行的是await run_in_threadpool(self.file.read, size)——因为普通文件的 IO 在 Linux 上没有真正的异步支持（O_NONBLOCK对普通文件无效，磁盘 IO 总是”就绪”的，实际会阻塞），所以 asyncio 生态里的文件操作（包括aiofiles）本质上都是把同步 IO 丢进线程池。这带来两个实际影响：① 它占用 anyio 那个默认 40 token 的线程池——大量并发上传时，文件读写会和def路由抢线程，可能成为瓶颈（可以调大total_tokens）；② 不要以为「加了 await 就不阻塞」——如果你在async def里用普通的open()和f.write()写一个大文件，那是真的会阻塞事件循环的，必须用aiofiles或run_in_threadpool。真正的异步文件 IO 需要io_uring（Linux 5.1+），Python 生态还没有成熟的封装。 - 追问：为什么说「重新编码」是图片上传最好的安全措施？ 因为它一次性解决了四个问题。① 销毁隐藏的载荷——攻击者可以把 PHP 代码或 JS 附加在图片文件尾部（「图片马」），或者藏在 EXIF 的注释字段里；这些数据在图片解码时会被忽略，但如果服务器配置有疏漏就可能被执行。重新编码后输出的文件只包含像素数据，任何附加内容都消失了。② 清除 EXIF 隐私信息——手机拍的照片默认包含 GPS 坐标、拍摄时间、设备型号，用户上传头像时并不知道自己泄露了家庭住址；
convert("RGB")加重新save()会丢弃所有 EXIF。③ 验证它确实是一张合法图片——能被 Pillow 完整解码并重新编码，就排除了「伪装成图片的其他文件」。④ 顺便统一格式和尺寸——省存储、省带宽、前端展示也更一致。代价是 CPU 开销（所以要run_in_threadpool）和有损压缩的画质损失（可以用quality=85~90平衡）。要注意配合Image.MAX_IMAGE_PIXELS防「图片炸弹」——一个 10KB 的 PNG 可以解压成 50000×50000 像素、占用 10GB 内存。 - 追问：什么时候该用对象存储直传而不是经过应用上传？ 三个判断维度。① 文件大小——超过 10MB 就该考虑，超过 100MB 基本必须。因为上传期间连接一直占着：ASGI 下虽然不像 WSGI 那样占满一个 worker，但仍然占一个 task 加线程池 token，而且带宽和磁盘 IO 全部经过你的应用服务器（云上的出入带宽是要花钱的）。② 并发量——即使单个文件不大，如果有几百个用户同时上传，应用服务器的带宽和临时磁盘都会成为瓶颈。③ 是否需要断点续传——对象存储的 multipart upload API 原生支持分片、并发上传、单片重试，自己实现一套非常麻烦。直传的流程是：后端签发一个带
content-length-range（大小限制）、Content-Type限制和过期时间的预签名凭证 → 前端直接 POST 到对象存储 → 前端回调通知后端 → 后端head_object确认后落库。最关键的安全点在最后一步：必须服务端主动查询对象元数据确认文件真实存在及其大小类型，绝不能相信前端上报的信息（否则用户可以不上传就调回调，或者声称传了一个 10MB 的图片实际是 1KB 的垃圾）；同时要校验 key 的前缀属于当前用户，防止伪造别人的路径。
八、加强记忆
FastAPI 接收文件有两种声明方式，区别很关键：file: bytes = File() 会把整个文件读进内存（200MB 视频就是 200MB 常驻内存，几个并发直接 OOM）；file: UploadFile 底层是 SpooledTemporaryFile——小于阈值（默认 1MB）留在内存、超过自动落磁盘临时文件，还提供了 filename/content_type/size 和异步的 read/write/seek。所以除了确定很小的文件，一律用 UploadFile。但要注意 await file.read() 不带参数时仍会把整个文件读进内存，大文件必须分块读；而且 UploadFile 的 async 方法其实是 run_in_threadpool 包装的同步 IO（普通文件没有真异步），所以它占用那 40 个线程池 token。FastAPI/Starlette 没有内置的请求体大小限制（不像 Flask 有 MAX_CONTENT_LENGTH）——三层防护缺一不可：① Nginx 的 client_max_body_size（最有效，超出直接 413、请求体不进应用）、② ASGI 中间件检查 Content-Length（快但可伪造）、③ 读取时累加实际字节（最可靠，因为 chunked 编码时前两层都可能失效）。安全五道防线：① filename 和 content_type 都是客户端提供的、完全不可信——文件名可能含 ../ 做路径穿越，所以自己用 UUID 生成、原名存数据库；② 类型查魔数，读完必须 await file.seek(0)（否则保存出空文件，这是最高频的低级错误）；③ 图片最彻底的做法是用 Pillow 重新编码——一次性清除 EXIF（含 GPS）、销毁图片马、统一格式、验证确实能解码，还要设 Image.MAX_IMAGE_PIXELS 防图片炸弹、禁止或清洗 SVG（XML 能内嵌 script）；④ 存 Web 根目录之外，绝不能放在 StaticFiles 挂载的目录下（那等于没有权限控制）；⑤ 下发前校验权限，用 404 而不是 403（不泄露资源存在性）。下载侧：FileResponse 内部流式且自动支持 Range 断点续传，filename 参数会自动生成 RFC 5987 的中文文件名；StreamingResponse 不自动支持 Range；生产环境不要让应用发文件——受保护文件用 X-Accel-Redirect（应用只鉴权、Nginx 零拷贝，location 要配 internal），云上最优是预签名 URL。大文件的正解是对象存储预签名直传：后端只签发带 content-length-range 和类型限制的凭证，回调必须服务端 head_object 确认、不能信前端报的 size 和 type，还要校验 key 的归属。两个容易忽略的坑：BackgroundTasks 里不能直接用 UploadFile 对象（请求结束时临时文件已被删除），要先保存到自己的位置再传路径；有文件时其他字段必须声明成 Form，且不能同时有 File 和 JSON Body（协议上互斥）。最后，CPU 密集的图片处理要 run_in_threadpool，更重的活丢 Celery。