文件上传：从收到字节到安全落地

写一个能接收文件的接口，只要几行代码。写一个不会被大文件拖垮、不会把伪装文件直接公开、失败后也不留下垃圾的上传接口，才是真正的工程题。

这一章我们围绕同一个场景动手：给课程 42 上传课堂资料。调用方可以在一次请求里提交标题和多个文件；服务端逐个检查文件，把通过检查的内容写入内部目录，返回对象键、大小和摘要。我们会先让它接住请求，再一点点补上内存、类型、命名、清理、并发和对象存储边界。

先看清上传请求的形状

普通 JSON 请求像一个完整的数据对象，文件上传则更像一个带分隔线的包裹。它的媒体类型是 multipart/form-data，请求头里的 boundary 指定了分隔各部分的边界。每一部分都有自己的说明，其中可能是普通文本，也可能是带文件名和内容类型的二进制数据。

FastAPI 要解析这种请求，需要安装 python-multipart：

python -m pip install fastapi uvicorn python-multipart

先写一个只返回元数据的端点：

from typing import Annotated
 
from fastapi import FastAPI, File, Form, UploadFile
 
app = FastAPI()
 
 
@app.post("/lessons/{lesson_id}/materials")
async def upload_materials(
    lesson_id: int,
    title: Annotated[str, Form(min_length=1, max_length=60)],
    files: Annotated[list[UploadFile], File()],
) -> dict[str, object]:
    return {
        "lesson_id": lesson_id,
        "title": title,
        "files": [
            {
                "filename": upload.filename,
                "content_type": upload.content_type,
                "size": upload.size,
            }
            for upload in files
        ],
    }

这里有三种来源不同的参数：

• lesson_id 来自路径 /lessons/{lesson_id}/materials；
• title 是多段表单里的普通字段；
• files 是同名文件字段组成的列表。

用 curl 发一次请求：

curl -X POST http://127.0.0.1:8000/lessons/42/materials \
  -F 'title=课堂资料' \
  -F 'files=@cover.png;type=image/png' \
  -F 'files=@slides.pdf;type=application/pdf'

同名的两个 files 部分会被收集成 list[UploadFile]。title 仍然是字符串，路径里的 42 则按整数校验。换句话说，混合上传不是先发一段 JSON，再额外附一个文件；它从头到尾只有一个多段表单请求体。

单文件、可选文件与多文件

三种常用声明方式只差类型：

from typing import Annotated
 
from fastapi import File, UploadFile
 
 
required_file: Annotated[UploadFile, File()]
optional_file: Annotated[UploadFile | None, File()] = None
multiple_files: Annotated[list[UploadFile], File()]

列表类型解决的是“把同名文件字段收集起来”，不是数量上限。生产接口仍要主动检查 len(files)。如果业务只允许三份资料，第四份应该尽早拒绝，而不是先保存再假装没看见。

Starlette 的底层表单解析器还会限制字段数、文件数和普通字段每部分的大小，用来抵抗大量空字段消耗 CPU 和内存。要注意，这些解析限制不等于业务上的单文件字节上限；文件内容能否超过 2 MiB、50 MiB 或其他阈值，仍应在代理层和应用保存流程中分别限制。

bytes 和 UploadFile 不是两种写法而已

FastAPI 可以直接把文件参数声明成 bytes：

from typing import Annotated
 
from fastapi import File
 
 
@app.post("/tiny-files")
async def inspect_tiny_file(
    file: Annotated[bytes, File(max_length=64 * 1024)],
) -> dict[str, int]:
    return {"size": len(file)}

请求一个五字节的小文件：

curl -X POST http://127.0.0.1:8000/tiny-files \
  -F 'file=@note.txt;type=text/plain'

响应是：

{
  "size": 5
}

这种写法很直接，因为路由函数拿到的已经是完整 bytes。代价也同样直接：文件内容必须完整进入内存。即使给 File 设置了 max_length，校验发生时完整字节已经被读取，所以它适合明确很小的内容，不适合视频、压缩包或尺寸不可控的附件。

UploadFile 的思路不同。它给你的是一个上传文件对象：

• filename 是客户端提交的原始文件名；
• content_type 是客户端声明的媒体类型；
• size 是解析请求内容时统计出的字节数，通常比相信整个请求的 Content-Length 更贴近当前文件；
• headers 保存这个多段部分自己的请求头；
• file 是一个标准的类文件对象；
• read、seek、write、close 提供异步调用方式。

小文件在内存，大文件转入临时文件

UploadFile.file 背后是 SpooledTemporaryFile。数据较小时，它先留在内存缓冲区；超过框架配置的卷存阈值后，内容转到系统临时文件。调用 fileno() 或显式触发 rollover() 也可能让它提前落盘。

这带来两个容易误解的结论。

第一，UploadFile 会控制解析阶段的峰值内存，但它不是“文件永远不进内存”。小文件仍可能留在内存里，多个并发上传的缓冲区也会相加。

第二，临时文件不是业务存储。请求结束并关闭上传对象后，它会被清理；不能把 upload.file 的临时位置当作长期对象键，更不能把它交给一个请求结束后才运行的任务，指望届时还可以继续读取。

文件指针会移动

读取文件头后，指针停在已读取位置：

header = await upload.read(16)
await upload.seek(0)

少了 seek(0)，后续保存就会丢掉前 16 个字节。这个问题很隐蔽：接口可能返回成功，磁盘上也确实有文件，但图片或文档已经损坏。

如果接下来调用的是同步库，可以把 upload.file 交给它；在 async def 路由中，阻塞式库调用要放进线程池，后面会把这条边界写进完整示例。

校验不是只看 content_type

浏览器说“这是图片”，并不能证明它真是图片。原始文件名和 content_type 都来自客户端，请求发送者可以随意填写。可靠的上传检查应当分层，而不是寻找一个万能条件。

数量和总量先挡在外面

一次允许三个文件，就在任何落盘动作之前检查：

MAX_FILE_COUNT = 3
 
if len(files) > MAX_FILE_COUNT:
    for upload in files:
        await upload.close()
    raise HTTPException(status_code=400, detail="一次最多上传 3 个文件")

单文件上限和单次请求总量是两回事。三个文件各自没超过 20 MiB，总量仍可能达到 60 MiB。实际项目通常同时设置：

• 反向代理或网关的整个请求体上限；
• 表单解析器的字段数和文件数上限；
• 应用保存时的单文件累计字节上限；
• 用户、课程或租户的配额；
• 磁盘或对象存储侧的容量告警。

只在读取完成后查看 upload.size，可以快速拒绝已解析出的超限文件，却不能替代边读边累计。后者才可以在数据继续写入业务存储前停下来，并保证规则与真正保存的字节一致。

扩展名和声明类型要互相对应

先定义明确的允许列表：

TYPE_RULES = {
    ".png": ("image/png", b"\x89PNG\r\n\x1a\n"),
    ".jpg": ("image/jpeg", b"\xff\xd8\xff"),
    ".jpeg": ("image/jpeg", b"\xff\xd8\xff"),
    ".pdf": ("application/pdf", b"%PDF-"),
}

允许列表比“拒绝危险后缀”的黑名单更容易推理。需求是只收图片和 PDF，就只开放这些格式；不要试图枚举世界上所有可能危险的脚本后缀。

接着让后缀与声明类型对应：

from pathlib import Path
 
from fastapi import HTTPException, UploadFile
 
 
def check_metadata(upload: UploadFile) -> tuple[str, bytes]:
    basename = client_basename(upload.filename)
    suffix = Path(basename).suffix.lower()
 
    if suffix not in TYPE_RULES:
        raise HTTPException(status_code=415, detail="不支持的文件扩展名")
 
    expected_type, signature 

这里返回 415 Unsupported Media Type，因为问题是服务端不接受这种媒体内容。大小超限更适合 413 Content Too Large；字段缺失或表单结构不符合声明时，FastAPI 通常返回 422。状态码稳定后，前端才能准确区分“换个格式”“压缩后重试”和“补齐字段”。

文件头能识破最浅的一层伪装

PNG、JPEG 和 PDF 开头都有可识别的字节特征，常被称为魔数或文件签名。保存循环读到第一块时，可以检查 startswith(signature)。这能挡住把普通文本简单改名为 .png 的情况。

但魔数不是完整内容审查。一个文件可以拥有正确开头，后面仍是损坏数据；复合格式里也可能藏有脚本或恶意载荷。真实图片业务可以再用图像解码器完整解码，并限制像素数，防止超大尺寸图片在解压时吃光内存。文档、压缩包、音视频也应使用对应解析器或扫描服务，而不是只看前几个字节。

原始文件名只用来展示

最危险的保存方式看起来往往最自然：

destination = UPLOAD_ROOT / upload.filename  # 错误：不要直接拼接客户端文件名

客户端文件名可能重名，可能包含 / 或 \，也可能利用不同操作系统对尾部空格、点号和 Unicode 字符的处理差异。即便 Path(...).name 能去掉当前系统认识的路径，跨平台分隔符仍要考虑。

我们先把原始值压成一个只用于记录和展示的 basename：

def client_basename(filename: str | None) -> str:
    if not filename:
        raise HTTPException(status_code=400, detail="文件名不能为空")
 
    basename = filename.replace("\\", "/").rsplit("/", 1)[-1]
    if basename in {""

真正写入磁盘的名称由服务端生成：

file_id = uuid.uuid4().hex
final_path = lesson_dir / f"{file_id}{suffix}"

这样做同时解决了路径穿越和重名覆盖。数据库可以分别保存：

original_name = "春季课程封面.png"
object_key    = "lesson-42/8ac0...e912.png"
content_type  = "image/png"
size          = 128430
sha256        = "..."
status        = "quarantined"

对外响应只返回对象键或业务资源编号，不返回服务器绝对路径。绝对路径会暴露部署结构，也把客户端和某台机器的目录绑定在一起。

边读边验，失败时不留半个文件

一次性 await upload.read() 很适合演示，不适合保存不可控文件。更稳妥的过程是：每次读固定大小的一块，累计字节，更新摘要，写入带 .uploading 标记的临时文件；全部成功并刷新后，再原子改名为正式文件。

下面是同步保存核心。它不读取任何服务端既有文件，只消费已经由框架解析好的上传流，并且所有路径都由服务端内部值构造：

import hashlib
import os
from dataclasses import dataclass
from pathlib import Path
from typing import BinaryIO
 
from fastapi import HTTPException
 
 
CHUNK_SIZE = 1024 * 1024
MAX_FILE_SIZE = 2 * 1024 * 1024
 
 
@dataclass(frozen=True)
class

这段代码有几个容易被忽略的细节。

"xb" 要求临时文件必须是新文件，避免意外覆盖。total 统计的是保存循环真正读到的内容，不依赖客户端头部。摘要在同一个循环里更新，不需要为了算哈希再读一遍。flush() 和 fsync() 把 Python 缓冲与操作系统缓冲推进到稳定存储，再用 os.replace() 在同一文件系统内把临时名称切换为正式名称。

异常分支捕获 BaseException，因此普通异常和任务取消都会删除临时文件，然后把原异常继续抛出。这里不能用一个空泛的 except: return 吞掉错误，否则调用方收到成功，服务端却只留下半个文件。

多文件请求还要处理部分成功

假设第一个文件保存成功，第二个文件被识别为伪装文件。一次请求的业务语义如果是“要么全收，要么全不收”，就要删除第一个已保存文件：

stored: list[StoredFile] = []
 
try:
    for upload in files:
        stored.append(await store_upload(lesson_id, upload))
except BaseException:
    for item in stored:
        (UPLOAD_ROOT / item.object_key).unlink(missing_ok=True)
    for upload in files:
        await upload.close()
    raise

这是一种本地文件系统上的补偿动作，不是真正数据库事务。跨对象存储、数据库和消息队列时，更稳妥的设计是先把记录标成 uploading 或 quarantined，只有所有步骤成功才改为 ready；清理任务定期删除长期未完成的对象。

async 不会把同步磁盘变成异步磁盘

前面的 copy_checked() 使用普通文件对象，source.read()、target.write()、fsync() 都是阻塞调用。如果在 async def 路由里直接运行整段函数，事件循环会被占住，其他请求也要等。

把这段阻塞 I/O 整体交给线程池：

import uuid
 
from fastapi import UploadFile
from fastapi.concurrency import run_in_threadpool
 
 
async def store_upload(lesson_id: int, upload: UploadFile) -> StoredFile:
    suffix, signature = check_metadata(upload)
    lesson_dir = UPLOAD_ROOT / f"lesson-{lesson_id}"
    lesson_dir.mkdir(mode=0o750, parents=True, exist_ok=

FastAPI 的 UploadFile.read() 等异步方法，本身也会把底层文件操作安排到线程池。逐块 await upload.read() 是可用的；这里把完整同步复制函数交给线程池，是为了让读取、写入、刷新和原子改名保持在同一条清晰的同步流程里。

线程池不是无限资源。每个并发上传都会占一个工作线程，还会占临时文件、磁盘带宽和目标空间。应用层应配合并发信号量、用户配额和网关限速，不要因为事件循环没有阻塞，就认为机器可以同时接收任意数量的大文件。

BackgroundTasks 适合什么

BackgroundTasks 会在响应发送后运行任务，适合短小、可丢失后重建、与当前进程关系紧密的动作，例如写一条审计日志或清理一个确定的临时标记。

它不等于独立任务队列：

• 任务仍在当前应用进程；
• 进程退出时任务可能丢失；
• 它没有天然的持久化、重试、超时和跨机器调度；
• 重型转码或扫描仍会争用当前服务的 CPU、内存和线程资源。

更关键的是，请求结束后 UploadFile 会被关闭。因此后台任务只能接收已经保存好的对象键或资源编号，不能接收 UploadFile，再慢慢读临时上传流。

from fastapi import BackgroundTasks
 
 
def append_audit_log(object_key: str) -> None:
    # 小型、可快速完成的本地动作
    ...
 
 
@app.post("/audit-example", status_code=202)
async def audit_example(
    background_tasks: BackgroundTasks,
    file: UploadFile,
) -> dict[str, str]:
    stored = await store_upload(42, 

病毒扫描、视频转码、OCR 和大模型处理更适合独立任务系统。接口完成隔离保存和数据库记录后返回 202 Accepted，任务消息里只携带资源编号。工作进程根据编号重新获取对象，自行创建数据库连接，成功后更新状态，失败则记录原因并按策略重试。

把端点收成一个完整版本

现在把各部分接起来。下面的代码与前面的请求使用同一个 /lessons/{lesson_id}/materials 端点：

from __future__ import annotations
 
import hashlib
import os
import uuid
from dataclasses import dataclass
from pathlib import Path
from typing import Annotated, BinaryIO
 
from fastapi import FastAPI, File, Form, HTTPException, UploadFile, status
from fastapi.concurrency import run_in_threadpool
 
 
app = FastAPI()
UPLOAD_ROOT = Path(__file__

这个实现故意顺序保存多个文件。顺序执行更容易保证失败补偿，也避免一个请求内部同时抢占多条磁盘写入。真的需要并行时，要先明确总并发上限、取消传播和部分成功如何回滚，不能只把循环替换成 gather()。

成功请求会看到什么

用一个 13 字节的 PNG 测试内容和一个 15 字节的 PDF 测试内容请求端点后，返回 201 Created：

{
  "lesson_id": 42,
  "title": "课堂资料",
  "files": [
    {
      "object_key": "lesson-42/bb31bece861348a5becee449a7cfed74.png",
      "size": 13,
      "sha256": "401326cb307d7998aeec563482b28743a05152934359f5ceb397ff42cb9122b6"
    },
    {
      "object_key": "lesson-42/dec7b33a2a664eac916f1b60736ab181.pdf",
      "size": 15,
      "sha256": "b9ac328b92d5b1bb84e1c8b85e62e2b769c68892f1e9d5daa9709390523ffc44"

内部编号每次都会重新生成，因此你运行时看到的对象键会不同；大小和相同字节内容的 SHA-256 摘要保持一致。

伪装文件怎样失败

把普通文本命名成 fake.png，同时声明为 image/png，扩展名和声明类型虽然对得上，第一块内容却没有 PNG 签名。响应是：

{
  "detail": "文件内容与类型不一致"
}

状态码为 415，对应的 .uploading 文件已经删除。如果它前面还有本次请求刚保存成功的文件，外层补偿逻辑也会删掉它们。

还应该测哪些失败路径

一个上传端点至少要覆盖这些测试：

• 缺少 title 或 files 时返回字段校验错误；
• 第四个文件触发数量上限；
• 空文件被拒绝；
• 大小刚好等于上限时成功，多一个字节时返回 413；
• 扩展名不在允许列表时返回 415；
• 扩展名与声明类型冲突时返回 415；
• 文件头与声明格式冲突时返回 415；
• 文件名包含 ../、反斜杠或超长字符时不会逃出上传根目录；
• 第二个文件失败时，第一个文件被补偿删除；
• 模拟写盘异常或任务取消后，没有残留 .uploading 文件；
• 多个并发请求不会复用同一个内部名称。

从本地磁盘切到对象存储

本地磁盘适合开发和单机部署。应用扩成多个实例后，文件写在哪台机器、实例重建后是否还在、静态内容怎样分发，都会变成额外问题。对象存储更适合把内容和应用进程解耦。

小文件经过应用服务

最接近本章代码的路线是：客户端把文件发给 FastAPI，应用完成认证、大小和格式校验，再把流写入对象存储。好处是规则集中，调用方简单；代价是每个字节都经过应用实例，占用入口带宽、线程和连接时间。

对象存储 SDK 往往提供同步文件对象接口。可以继续把 upload.file 交给 SDK，并把同步上传调用放进线程池。对象键仍由服务端生成，例如：

quarantine/lessons/42/8ac0...e912.png

数据库保存对象键、原始文件名、摘要、上传者和状态，不保存带时效签名的访问 URL。下载时再按授权生成短期地址，或由下载端点代理响应。

大文件直传对象存储

视频或大型归档不必穿过 FastAPI。更合适的协议是：

分片上传创建后必须完成或中止。没有收尾的分片会继续占用存储，因此要配置生命周期规则清理长期未完成的上传，并记录上传编号以便主动中止。

预签名地址是临时授权，不是内容安全检查。即使客户端直传，仍要把对象先放进隔离前缀或私有存储桶；审核通过后再复制、改状态或开放读取。相同对象键的覆盖语义也要明确，最简单的做法仍是每次生成全新内部键。

上线前沿着生命周期检查

文件上传不是在 return 时结束。沿着“接收、验证、保存、处理、发布、删除”检查，遗漏会少很多。

接收阶段

• 网关限制整个请求体和请求速率；
• 框架与 python-multipart 保持更新；
• 限制表单字段数、文件数和业务单次总量；
• 接口先做认证、授权与租户配额检查；
• 超时和客户端断开时能取消后续工作。

验证阶段

• 扩展名使用允许列表；
• 声明类型只当作一条线索；
• 检查文件头并按格式完整解析；
• 图片限制像素数，压缩包限制展开后的数量、总大小和嵌套深度；
• 有需要时执行病毒扫描和内容审核。

保存阶段

• 服务端生成内部名称，原始文件名只用于展示；
• 上传目录不可执行，最好与应用代码和公开静态目录分离；
• 边读边累计大小和摘要；
• 使用临时名称，成功后再原子改名或提交对象；
• 所有异常和取消路径都清理临时产物；
• 数据库只记录相对对象键，不暴露绝对路径。

处理与发布阶段

• 小任务可以在响应后执行，重任务进入可持久化队列；
• 后台任务只接资源编号或对象键，不持有 UploadFile；
• 资源状态清楚区分 uploading、quarantined、processing、ready 和 failed；
• 下载前重新做授权，设置正确的媒体类型和下载响应头；
• 删除业务记录时同步安排对象删除，并对失败清理做重试。