本文目录
浏览器提交头像、导入 CSV、附件表单——HTTP body 不再是纯 JSON，而是 multipart/form-data：字段与文件块交替出现。服务端若自己解析字节流，要处理 boundary、filename、content-type，容易出错。FastAPI 用 UploadFile 和 File / Form 声明参数，Starlette 在后台把流式 body 拆成可 async 读取的文件对象。真正容易踩坑的是存储边界：文件读进内存还是 spool 到磁盘、谁算 hash、谁上传对象存储、数据库里存什么——这些应在 API / Service / Storage 之间划清，而不是在路由里堆满 open() 和 boto 调用。
multipart 与 UploadFile
客户端（或 TestClient）用表单字段 + 文件：
# 测试侧示意
files = {"file": ("report.csv", b"a,b\n1,2", "text/csv")}
data = {"category": "monthly"}
client.post("/imports", data=data, files=files)服务端：
from typing import Annotated
from fastapi import FastAPI, File, Form, UploadFile
app = FastAPI()
@app.post("/imports")
async def create_import(
category: Annotated[str, Form()],
file: Annotated[UploadFile, File()],
) -> dict[str, str]:
if file.content_type not in ("text/csv", "application/vnd.ms-excel"):
raise HTTPException(status_code=415, detail="unsupported media type")
content = await file.read()
# 交给 Service，不在此写业务
job_id = import_service.enqueue(category=category, filename=file.filename or "upload", data=content)
return {"job_id": job_id}UploadFile 提供：
filename、content_type元数据（不可信，需校验扩展名与魔数）await file.read()/async for chunk in file流式读取- 底层可能是 SpooledTemporaryFile：小文件在内存，超过阈值写临时磁盘
大文件应用 chunk 循环 写存储，避免一次 read() 吃光内存：
async def save_upload_stream(file: UploadFile, dest: Path, chunk_size: int = 1024 * 1024) -> int:
total = 0
with dest.open("wb") as out:
while chunk := await file.read(chunk_size):
total += len(chunk)
out.write(chunk)
await file.seek(0) # 若后续还要读，需 rewind 或禁止重复读
return totalFile 与 Form 一起用
多文件、可选文件：
@app.post("/albums/{album_id}/photos")
async def upload_photos(
album_id: int,
photos: Annotated[list[UploadFile], File()],
caption: Annotated[str | None, Form()] = None,
) -> dict[str, object]:
...list[UploadFile] 时 HTML 表单需同名多 file input。纯 JSON API 不能带文件——需改 multipart 或分「先拿上传 URL、再 PUT 对象存储」的预签名流程（大文件、移动端常用）。
预签名直传流程：API 返回 {upload_url, key}，客户端 PUT 到对象存储，再 POST /assets/complete 写元数据。API 不经手大 body，带宽与超时压力在 CDN/存储侧。Service 仍管 key 规则与权限，Storage 适配器提供 presign_put(key) -> url。
两种上传模式对照
课件里的「上传模式」核心不是 UploadFile 语法，而是字节走哪条路：
| 模式 | 数据路径 | 优点 | 代价 |
|---|---|---|---|
| 服务端中转 | 浏览器 → FastAPI（UploadFile）→ OSS/本地盘 | 实现简单，易在服务端扫毒/转码 | 占应用带宽与内存，大文件易超时 |
| 客户端直传 | 浏览器 →（预签名）→ OSS；API 只签发与确认 | 应用服务器轻，易扩 | 要处理回调/complete、权限与伪造 complete |
直传时 API 常见两步：POST /upload/sign 校验登录与文件元数据，短事务读 OSS 配置后返回临时凭证；客户端上传成功后 POST /upload/complete 写业务表。IAM/STS 策略应最小权限（只允许 Put 到指定前缀），不要把主账号密钥下发给浏览器。中转模式适合头像、小 CSV；视频、大包优先直传。
两种模式可以并存：小文件 UploadFile，大文件 sign+complete，Service 写 metadata 的接口尽量统一，避免两套仓储逻辑。
存储边界：三层各管什么
| 层 | 文件相关职责 |
|---|---|
| API | 收 UploadFile、大小/类型初筛、调 Service |
| Service | 病毒扫描策略、命名规则、写元数据记录、事务 |
| Storage 适配器 | 本地目录 / S3 / MinIO 的 put/get/delete |
class StorageBackend(Protocol):
async def put(self, key: str, data: bytes, content_type: str) -> str: ...
def url_for(self, key: str) -> str: ...
class LocalStorage:
def __init__(self, root: Path) -> None:
self.root = root
async def put(self, key: str, data: bytes, content_type: str) -> str:
path = self.root / key
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(data)
return key
def url_for(self, key: str) -> str:
return f"/static/{key}"Service：
class AssetService:
def __init__(self, storage: StorageBackend, repo: AssetRepository) -> None:
self.storage = storage
self.repo = repo
async def store_avatar(self, user_id: int, file: UploadFile) -> str:
raw = await file.read()
if len(raw) > 2 * 1024 * 1024:
raise AppError(code="FILE_TOO_LARGE", message="max 2MB")
key = f"avatars/{user_id}/{uuid.uuid4().hex}.webp"
await self.storage.put(key, raw, file.content_type or "application/octet-stream")
self.repo.save_asset(user_id=user_id, key=key)
return self.storage.url_for(key)数据库里存 key 或 URL，不要默认把 BLOB 塞进业务表——除非文件极小且强一致事务要求同事务提交（仍要评估备份体积）。
AssetRepository 只存 user_id, key, size, checksum, created_at；下载时 Service 调 storage.url_for 或生成限时签名。删除用户时要 cascade 删对象存储里的 key，通常 Service 编排 repo.list_keys + storage.delete，避免孤儿文件。
安全与校验
- 限制大小：反向代理
client_max_body_size+ 应用内read前查Content-Length（可伪造，以流计数为准）。 - 类型：不信
content_type，结合扩展名与文件头（magic bytes）。 - 路径：保存时用生成的 UUID key，禁止用户控制
../路径。 - 病毒扫描：异步队列处理，上传接口先收再扫，失败则删对象并更新状态。
图片还可做尺寸/比例校验、重新编码 strip EXIF，减少恶意 metadata。CSV 导入应对行数上限做 Service 级限制，避免一次上传拖垮 worker。
依赖注入 Storage
与 DB 相同，Storage 通过 Depends 注入，测试 override 为 InMemoryStorage：
def get_storage(settings: SettingsDep) -> StorageBackend:
return LocalStorage(root=settings.upload_dir)
StorageDep = Annotated[StorageBackend, Depends(get_storage)]路由只收 UploadFile 并传给 AssetService；Service 构造函数接受 StorageBackend，不 import 具体 S3 客户端类。
测试
TestClient 传 bytes 不需真磁盘：
def test_upload_avatar(client: TestClient) -> None:
app.dependency_overrides[get_storage] = lambda: InMemoryStorage()
try:
resp = client.post(
"/users/1/avatar",
files={"file": ("a.png", b"\x89PNG\r\n", "image/png")},
)
assert resp.status_code == 200
assert "url" in resp.json()
finally:
app.dependency_overrides.clear()集成测试可对临时目录 LocalStorage 断言文件存在。
同步 vs 异步上传路由
async def 路由应用 await file.read()；若在 sync 路由里接 UploadFile，Starlette 仍可能阻塞线程池读盘。大文件 + 慢存储（S3 upload）应用 async storage client 或 run_in_threadpool 包同步 SDK。Worker 超时（Uvicorn timeout）要大于最慢上传预期，或改异步任务：接口只收文件登记 job_id，后台 Celery/ARQ 慢慢传。
表单字段与 JSON 混用限制
同一请求不能既 application/json 又 multipart。前端改传 FormData 时，Query 参数仍可用，但 body 只能是表单。文档要在 OpenAPI 里写清 multipart/form-data schema，避免客户端仍 POST JSON 导致 422。列表页上传多图时，注意浏览器与 Starlette 对多 file 字段名的约定，集成测试用 TestClient files=[(...), (...)] 覆盖。
清理与合规
用户删号、 GDPR 删除请求时，Service 应列出该用户全部 object key 并调 storage 批量 delete，再删 DB 行。只删 DB 不删对象，桶里会堆孤儿文件，成本与合规都有风险。生命周期规则（S3 lifecycle expire）是兜底，不应替代应用层主动删。
图片处理与衍生版本
头像场景可在 Service 里调用 Pillow 生成 thumbnail，原图与缩略图两个 key 都存 storage。CPU 重的工作放任务队列，上传接口只 accept 并 enqueue。失败时更新 job 状态，客户端轮询或 WebSocket 通知——UploadFile 只负责入口字节，不负责整条 pipeline。
合规保留与审计
某些行业要求文件保留 N 年；storage lifecycle 与 DB retention_until 字段一致。审计 log 记谁在上传、checksum、大小，不记全文件内容。下载接口记 access log 满足追溯。
与三层架构对齐回顾
Upload 路由 → AssetService → AssetRepository + StorageBackend。路由不出现 boto3，Repository 不出现 Pillow。测试分别 override storage 与 repo，Service 单测用 Fake 二者。这与本系列第 13 篇分层原则一一对应。
断点续传与分片（概念）
超大文件可用分片 upload：客户端每片 POST，Service 记录 etags，最后 complete multipart upload。API 形态与单次 UploadFile 不同，但 Storage 适配器接口可复用。MVP 用单文件 UploadFile 即可，规模上来再演进，不必第一天就上 multipart S3 API。
常见误解
UploadFile.file 当普通同步 IO 在 async 路由里狂读。 优先 await file.read()；同步磁盘 IO 量大时用 run_in_threadpool。
上传完成等于可公开访问。 私有桶应走签名 URL 或鉴权下载路由，不要把内网 path 直接暴露。
一个路由里既解析 CSV 又发邮件又写库。 解析可以 Service 分步；路由只接文件。
忘记 await file.close()。 框架通常在请求结束清理；若在依赖或后台任务里复制 UploadFile，要自己管生命周期。
内容分发与 CDN
公开静态资源可走 CDN，API 只返回 signed URL。上传仍进源站或直传 bucket，CDN 只加速读。缓存头 Cache-Control 由 storage 或下载路由控制，与 UploadFile 接收无关但属于同一「文件生命周期」故事线。
文件进服务、元数据进库、对象进存储——边界清楚之后，后面的配置生命周期、事务长度、认证注入都会落在明确的层上，而不是挤在同一个路由函数里。
选型本地盘 vs 对象存储：开发用 LocalStorage，生产用 S3 兼容，接口都是 StorageBackend，Service 无感切换。迁移历史文件时 batch job 走 Service + Repository 更新 key，UploadFile 路径不参与。监控 bucket 用量与上传 413/415 比例，异常升高可能是攻击或客户端 bug。
表单上传适合中小文件与浏览器直传；移动端大视频倾向 presigned。两种入口可以共存，不同路由，同一 Service 写 metadata。文档写清 size limit 与 supported types，减少无效上传浪费带宽。
小结
multipart + UploadFile 接收文件，Storage 适配器持久化，Repository 存 metadata。大文件流式读写，安全校验不信任客户端 MIME。测试用 bytes 与 InMemoryStorage，生产可换 S3 而不改 Service 签名。
与 JSON API 并存时，OpenAPI 文档应对 multipart endpoint 单独示例，前端 axios/fetch 需用 FormData 而非 JSON.stringify。后端 async 路由 + 流式写盘，可避免 worker 被大文件 block。上传失败应返回统一 error envelope（415 类型不对、413 过大），与系列异常处理篇一致，便于客户端统一 toast 或表单错误展示。
下载接口返回 FileResponse 或 redirect 到 signed URL 时，仍保持 API 层薄：Service 校验权限，Storage 生成 URL，路由只组装 Response。
UploadFile 解决「怎么收」；Storage 解决「怎么存」；Repository 解决「记什么」——三者边界清楚，上传功能才不易长成路由里的泥球。
multipart 边界由客户端生成，服务端勿手动拼接 raw body；始终用框架提供的 UploadFile/Form 解析，减少安全漏洞与编码错误。
上传与下载对称设计：存 key、验权限、再读 storage，三步骤模式复用。
进度反馈：大文件上传可返回 job_id，客户端轮询 status endpoint；UploadFile 只负责入口，异步 pipeline 在 Service 与 worker 中完成，避免 HTTP 请求超时断开导致半成品文件。
存储适配器接口稳定后，换云厂商只改实现类，业务与路由无感。
校验失败尽早返回，避免读完整个大文件才 415。
流式校验 Content-Type 与 magic bytes，别等整文件落盘后再拒绝。
UploadFile 负责收字节，Storage 负责持久化，分工明确才更容易测、换实现。