第 14 章：文件上传、静态资源与模板

前面章节的接口都只在交换 JSON。但真实项目里还有头像、附件、Excel 导入、报表下载，以及「后台管理页面要不要后端渲染」这类架构问题。本章把文件上传的安全边界、静态资源挂载与 Jinja2 模板一次讲清。

---

学习目标

• 掌握 UploadFile 与 bytes + File() 的适用边界，理解大文件为何必须用前者
• 能独立完成单文件、多文件、文件 + 表单字段混用的上传接口
• 掌握大小、MIME、扩展名三层校验，并理解各层的局限
• 能用 uuid4 重命名 + 安全扩展名阻断目录穿越攻击
• 会挂载 StaticFiles，用它和 FileResponse / StreamingResponse 返回文件
• 能用 Jinja2Templates 渲染页面，并判断是否真需要服务端模板

---

14.1 前置依赖：python-multipart

File 与 Form 依赖 python-multipart 解析 multipart/form-data，模板需要 jinja2，两者都在 fastapi[standard] 里：

pip install "fastapi[standard]"      # 已包含 python-multipart 与 jinja2

未安装时报错很明确：

RuntimeError: Form data requires "python-multipart" to be installed.

后续小节沿这条链路逐个环节加固：

---

14.2 UploadFile vs bytes

维度	bytes + File()	UploadFile
底层	整个文件读入内存	SpooledTemporaryFile，超阈值自动落盘
元数据	无	filename / content_type / size
适用	固定小文件（头像 ≤ 1 MB、证书）	任意用户上传

❌ 不推荐：bytes = File(...) 会把整个文件读进内存。头像、证书这类固定小文件还行，用户上传一旦到 2 GB，进程内存直接爆掉。

✅ 推荐：

@app.post("/upload/stream")
async def upload_stream(
    file: Annotated[UploadFile, File(description="任意大小的上传文件")],
) -> dict[str, str | int | None]:
    content = await file.read()     # ✅ 从 SpooledTemporaryFile 读，不占常驻内存
    return {"filename": file.filename, "content_type": file.content_type, "size": len(content)}

🔑 UploadFile 的方法全是异步的——await file.read()、await file.seek(0)、await file.close()。漏写 await 会拿到协程对象而不是数据。

---

14.3 单文件、多文件与文件 + 表单混用

@app.post("/articles/{article_id}/cover")
async def upload_cover(
    article_id: int,
    file: Annotated[UploadFile, File(description="封面图，≤ 2 MB")],
    alt_text: Annotated[str, Form(max_length=100)] = "",     # 文件与表单字段可共存
) -> dict[str, str | int]:
    return {"article_id": article_id, "filename": file.filename or "", "alt": alt_text}

@app.post("/articles/{article_id}/attachments")
async def upload_attachments(
    article_id: int,
    files: Annotated[list[UploadFile], File()],              # 多文件用 list[UploadFile]
) -> dict[str, object]:
    return {"article_id": article_id, "count": len(files), "names": [f.filename for f in files]}

客户端表单字段名必须与形参名一致（-F "file=@cover.png" -F "alt_text=封面图"）；多文件时字段名必须全部相同，写成 file1、file2 会因找不到 files 而返回 422。

---

14.4 三层校验：大小、MIME、扩展名

file.size 由 Starlette 解析时填充，但可能为 None，不能作为唯一防线；稳妥做法是边读边累计，并用白名单收窄 MIME 与扩展名：

MAX_UPLOAD_BYTES = 5 * 1024 * 1024      # 5 MB
CHUNK_SIZE = 1024 * 1024
ALLOWED_CONTENT_TYPES = {"image/jpeg", "image/png", "image/webp"}
ALLOWED_SUFFIXES = {".jpg", ".jpeg", ".png", ".webp"}

async def read_limited(file: UploadFile, max_bytes: int = MAX_UPLOAD_BYTES) -> bytes:
    """分块读取并累计校验大小，超限立即中断。"""
    total, chunks = 0, []
    while chunk := await file.read(CHUNK_SIZE):
        total += len(chunk)
        if total > max_bytes:
            raise HTTPException(status_code=413, detail=f"文件超过 {max_bytes // 1048576} MB")
        chunks.append(chunk)
    await file.seek(0)      # ✅ 复位，供后续再次读取
    return b"".join(chunks)

def safe_suffix(filename: str | None) -> str:
    suffix = os.path.splitext(filename or "")[1].lower()
    return suffix if suffix in ALLOWED_SUFFIXES else ".bin"

MIME 校验只是一句 if file.content_type not in ALLOWED_CONTENT_TYPES: raise HTTPException(status_code=400, ...)。要真正判断类型则必须读文件头（magic bytes）：magic.from_file(str(path), mime=True)（pip install python-magic，系统还需 libmagic）。

校验层	依据	能否伪造	作用
大小	实际字节数	不能	防 DoS、防磁盘打满
MIME	请求头 Content-Type	能	快速拒绝明显不匹配的请求
扩展名	文件名后缀	能（可白名单收窄）	决定落盘后的处理方式
文件头	文件真实内容	极难	最终判定依据

⚠️ 把 evil.php 的 MIME 写成 image/png 就能通过 MIME 校验，它只能作为粗筛，绝不能作为唯一防线。

---

14.5 安全存储：目录穿越与重命名

这是文件上传里唯一能直接导致服务器被拿下的一环。

❌ 不推荐（直接用用户文件名拼路径）：

@app.post("/upload/unsafe")
async def upload_unsafe(file: Annotated[UploadFile, File()]) -> dict[str, str]:
    # ❌ filename="../../../../etc/passwd" 可覆盖系统文件
    # ❌ filename="../../../app/main.py" 可写入代码可执行路径
    dest = os.path.join("uploads", file.filename or "unknown")
    with open(dest, "wb") as out:
        shutil.copyfileobj(file.file, out)
    return {"path": dest}

✅ 推荐（文件名完全由服务端生成，用户输入只用于取扩展名）：

UPLOAD_DIR = Path("uploads").resolve()

def build_destination(original_name: str | None) -> Path:
    return UPLOAD_DIR / f"{uuid.uuid4().hex}{safe_suffix(original_name)}"

@app.post("/upload/safe")
async def upload_safe(file: Annotated[UploadFile, File()]) -> dict[str, str | int]:
    dest = build_destination(file.filename)
    total = 0
    # ✅ anyio.open_file 返回异步文件对象，写盘不阻塞事件循环
    async with await anyio.open_file(dest, "wb") as out:
        while chunk := await file.read(CHUNK_SIZE):
            total += len(chunk)
            if total > MAX_UPLOAD_BYTES:
                dest.unlink(missing_ok=True)   # 超限时清理写了一半的临时文件
                raise HTTPException(status_code=413, detail="文件超过 5 MB")
            await out.write(chunk)
    return {"file_id": dest.stem, "size": total}

另外两条硬性要求：上传目录不要落在代码可执行路径下，更不能让 Web 服务器把它当脚本目录解析；需要保留原始文件名时把它存数据库，落盘名始终用 UUID。生产环境优先用对象存储（S3 / MinIO / OSS），服务端只签发预签名 URL。

---

14.6 挂载静态资源

from fastapi.staticfiles import StaticFiles

app.mount("/static", StaticFiles(directory="static"), name="static")

第一个参数是 URL 前缀（浏览器访问 /static/logo.png），第二个是磁盘目录（不存在会直接报错，需提前创建）。name="static" 是反向路由名，供 request.url_for("static", path="logo.png") 与模板里的 url_for 生成链接，反代后更换前缀全靠它；加 html=True 则目录请求自动返回 index.html。

---

14.7 返回文件：FileResponse 与 StreamingResponse

@app.get("/files/{file_id}")
async def download_file(file_id: str) -> FileResponse:
    # ✅ 路径由服务端拼接后再校验归属，杜绝 ../ 穿越
    path = (UPLOAD_DIR / file_id).resolve()
    if not path.is_file() or UPLOAD_DIR not in path.parents:
        raise HTTPException(status_code=404, detail="文件不存在")
    return FileResponse(
        path=path,
        filename="季度报告.pdf",                 # Starlette 会做 RFC 5987 编码
        media_type="application/pdf",
        content_disposition_type="attachment",   # 改 "inline" 则浏览器内预览
    )

FileResponse 内部用 sendfile 零拷贝，不把文件读进内存，大文件下载一律优先用它。当内容不是磁盘上的完整文件（数据库导出、拼接 ZIP、代理下游流）时用流式响应：

def iter_rows_as_csv(rows: Iterator[dict[str, object]]) -> Iterator[bytes]:
    yield "id,name,email\n".encode()
    for row in rows:
        yield f"{row['id']},{row['name']},{row['email']}\n".encode()


@app.get("/export/users.csv")
async def export_users() -> StreamingResponse:
    return StreamingResponse(
        iter_rows_as_csv(fetch_all_users()),   # ✅ 逐块产出，内存占用恒定
        headers={"Content-Disposition": 'attachment; filename="users.csv"'},
    )

---

14.8 Jinja2 模板

from fastapi.templating import Jinja2Templates

templates = Jinja2Templates(directory="templates")


@app.get("/", response_class=HTMLResponse)
async def index(request: Request) -> HTMLResponse:
    return templates.TemplateResponse(
        request=request,                    # ✅ 新版签名中 request 是必需参数
        name="index.html",
        context={"title": "文章列表", "articles": await list_articles()},
    )

配套模板（母版 + 子模板继承）：

{# templates/base.html #}
<!DOCTYPE html>
<html lang="zh-CN">
<head><meta charset="utf-8"><title>{% block title %}{{ title }}{% endblock %}</title>
  <link rel="stylesheet" href="{{ url_for('static', path='style.css') }}"></head>
<body>{% block content %}{% endblock %}</body>
</html>

{# templates/index.html #}
{% extends "base.html" %}
{% block title %}{{ title }} · 我的站点{% endblock %}
{% block content %}
  <h1>{{ title }}</h1>
  <ul>{% for a in articles %}<li><a href="/articles/{{ a.id }}">{{ a.title }}</a></li>{% endfor %}</ul>
{% endblock %}

• {% extends %} 必须写在子模板第一行；静态资源用 url_for('static', path='...') 生成，不要硬编码 /static/...，否则换前缀时全线 404。模板默认 autoescape=True， 是安全的，只有确定可信时才用 |safe。

⚠️ request 必须作为关键字参数传给 TemplateResponse。旧写法 TemplateResponse("index.html", {"request": request}) 已废弃，新版本会因缺少 request 报错。

---

14.9 纯 API 服务 vs 需要页面的服务

纯 API + 前端框架适合移动端后端、SPA、BFF：前后端分离各自迭代，复杂交互交给前端状态管理，代价是首屏要等 JS 与接口往返、SEO 需额外做 SSR。服务端模板适合后台管理、内部工具与 SEO 内容站：直出 HTML 首屏更快、SEO 天然友好、一套代码改一处即可，代价是每次交互都要往返。判断标准：页面交互简单、需要 SEO 或首屏速度、团队没有独立前端 → 用模板；同一份数据要服务 App / 小程序 / 第三方，或页面交互复杂 → 纯 API + 前端框架。

---

常见坑与排查

现象	原因	解决
RuntimeError: Form data requires "python-multipart"	未安装解析库	pip install python-multipart 或 uv add python-multipart
上传的文件出现在 uploads 之外，甚至覆盖系统文件	直接用 file.filename 拼路径，目录穿越	用 uuid4().hex 生成落盘名，只从白名单取扩展名
上传大文件后内存暴涨或被 OOM Killer 杀掉	用了 bytes = File(...)，或 await file.read() 一次读完	改用 UploadFile + 分块读取并累计校验大小
读文件内容时拿到空字符串	之前读过一次，没 await file.seek(0) 复位	每次重新读取前 await file.seek(0)
启动报 Directory 'static' does not exist	StaticFiles(directory=...) 指向的目录不存在	提前创建目录，或 Path("static").mkdir(exist_ok=True)
TypeError: TemplateResponse() missing ... 'request'	用了旧版签名，把 request 塞进了 context	改为 TemplateResponse(request=request, name=..., context=...)
下载中文文件名乱码或变成 download.bin	手工拼 Content-Disposition，未做 RFC 5987 编码	交给 FileResponse(filename="报告.pdf")；自拼时用 filename*=UTF-8''...
422 提示缺少 files 字段	多文件上传时字段名不一致（file1、file2）	客户端统一用同一字段名重复提交

---

本章小结

要点	说明
依赖	File / Form 需 python-multipart，模板需 jinja2，fastapi[standard] 已包含
类型选择	小文件可用 bytes，用户上传一律 UploadFile（超阈值自动落盘）
异步方法	await file.read() / seek(0) / close()，漏 await 会拿到协程
多文件	Annotated[list[UploadFile], File()]，客户端字段名必须相同
大小校验	分块累计读取，不要只信 file.size，超限返回 413
类型校验	content_type 可伪造；扩展名白名单 + 文件头探测才可靠
安全落盘	绝不用用户文件名拼路径，uuid4().hex + 白名单扩展名
静态资源与返回文件	app.mount("/static", StaticFiles(...), name="static")；大文件用 FileResponse
模板	Jinja2Templates + TemplateResponse(request=..., name=..., context=...)

---

练习题

1. 实现 POST /avatars：只接受 image/jpeg、image/png、image/webp，不超过 2 MB，落盘名为 uuid4().hex，并把「原始文件名 + 落盘名」写入一张表。
2. 用 filename="../../evil.txt" 与 filename="/etc/hosts" 攻击 14.5 的 upload_unsafe，观察文件落到哪里，再用 upload_safe 验证攻击失效。
3. 把 14.7 的下载接口改造成需要登录的版本（用第 12 章的权限依赖保护），并说明为什么不能直接 app.mount 上传目录。
4. 用 Jinja2 写一个文章列表页：母版含导航与静态 CSS，子模板展示标题与摘要，静态资源链接全部用 url_for 生成。

---

下一章预告

文件上传和模板解决的是「请求—响应」模型下的问题。但有些场景服务端需要主动推数据给客户端，HTTP 的单向模型就不够用了——下一章进入 WebSocket。

👉 第 15 章：WebSocket 实时通信