如果你能一路刷到这篇 FastAPI 学习教程的第 7 部分，说明前面那些路由、请求参数、依赖注入的关卡已经过得差不多了。但我猜你心里早就堆着一个问题：光会用 GET、POST 返回 JSON 根本不够，真实系统里永远离不开文件上传、文件下载、后台跑一个耗时任务再把进度告诉前端。这一篇就集中处理这几个绕不开的硬骨头。
这一部分我会围绕一个非常常见的需求展开：某文件中转 Demo 项目——用户上传一个文件，后台做点处理（压缩、转格式、生成缩略图之类），然后生成一个新的下载链接。前端能通过接口看到任务进度，下载时文件可能很大，不能把内存吃满。涉及的核心内容包括 FastAPI 的 UploadFile 机制、StreamingResponse 流式响应、后台任务状态追踪，以及上了生产环境之后一定会踩到的坑。适合已经把 FastAPI 基础跑通、正准备动手做真实项目的读者。
先说好，这一章不是讲理论，我会直接把会遇到的实际问题一个个摊开讲，包括很多文档里不会写的细节。比如为什么直接return file_bytes会在前端打开文件时报错，为什么BackgroundTasks里跑 CPU 密集任务会卡住所有人，以及如何让下载支持断点续传。
1. 为什么上传下载是这个系列里最劝退的一课
1.1 一个真实场景把小问题全暴露出来
我给朋友维护过一个内部用的文件速递小服务。需求一点都不复杂：上传一个压缩包，后端解压后重打包并加上水印，然后给用户一个下载链接。听起来就是标准的“上传—处理—下载”三段式接口，但真正写起来时，头三天我就遇到了五六个此前完全没预料到的情况：
第一批测试文件是 700MB 的压缩包，浏览器直接卡死在等待响应；第二个文件只有一个文件名带中文的 PDF，下载回来后文件名乱码；第三次我用一个超大视频测试，服务进程内存涨到 2GB 多，直接触发了 OOM；还有一次上传过程中把网络切断了，留下了半截临时文件躺在磁盘上；最后是用户连续点了两次上传，任务状态乱套了。
这些都不是 FastAPI 本身的 bug，而是我没有搞清楚文件处理的底层机制。如果你也在做类似的事情，这篇就是为你写的排雷指南。我后面所有代码，都以这个“文件速递”需求为主线。
1.2 Form 表单和 JSON 到底差在哪
很多人第一次上传文件时会犯一个认知错误：把文件内容转成 base64 塞进 JSON 里发过去。这在技术上可行，但代价很大：文件体积膨胀约 33%，而且大包体的 JSON 对服务端内存、前端内存都不友好。实际上传文件应该走multipart/form-data格式，这也是浏览器原生支持的编码方式。
FastAPI 对不同场景给了三套方案：
- 普通 JSON 请求体：适合提交配置、元数据等小数据；
Form(...)表单字段：适合提交普通字段，比如上传者的昵称、备注；File(...)、UploadFile：适合真正的二进制文件。
它们在 FastAPI 中的定义方式完全不同：
from fastapi import FastAPI, File, UploadFile, Form app = FastAPI() @app.post("/upload") async def upload( file: UploadFile = File(...), remark: str = Form("") ): return { "filename": file.filename, "size_hint": None, "remark": remark, }我见过不少初学者把file直接声明成bytes = File(...)，这样也能收到文件，但 FastAPI 会把整个文件内容一次性读进内存，再赋值给变量。小图没问题，大文件就是灾难。所以只要不是几十 KB 的小文件，建议一律使用UploadFile类型，它在底层用的是临时文件机制。这点下一节详细说。
1.3 “把文件读进内存再处理”为什么会出事
如果把文件读取想成“喝水”，bytes = File(...)就像把整桶水一口气灌进嘴里，而UploadFile则是开了一条水管慢慢放。FastAPI 底层使用的SpooledTemporaryFile其实已经做了缓冲处理：小文件留在内存里，超过一定阈值（默认 1MB）后就自动滚落到磁盘上的临时文件。
但是注意一个容易误判的点：虽然UploadFile底层不会一次性把大文件全塞进内存，很多人的代码却会在这个环节自己制造问题：
content = await file.read() # 把整个文件读进内存这句话会把文件所有内容加载到内存里。如果你做的是视频转码、大压缩包解压之类的操作，前面的巧妙设计就全白费了。正确做法是分块读取、边读边写。
2. UploadFile 的握手细节：从 File 到分块落盘
2.1 三种接收文件的方式怎么选
先给一个可以直接抄作业的对照表：
| 接收方式 | 适用场景 | 内存表现 |
|---|---|---|
file: bytes = File(...) | 几 KB 的小图片、JSON 验证 | 全部读入内存 |
file: UploadFile = File(...) | 中型文件，需要边读边处理 | 底层临时文件缓冲，安全 |
files: list[UploadFile] = File(...) | 多文件批量上传 | 底层临时文件缓冲 |
这里额外说一下多文件上传。写法并不复杂：
from typing import list @app.post("/upload-many") async def upload_many(files: list[UploadFile] = File(...)): saved_names = [] for file in files: saved_names.append(file.filename) return {"saved": saved_names}这个功能非常适合做批量上传工具、图床、附件发送等场景。但注意，如果一次性传几十个文件，你仍需要逐个处理、分块落盘，否则几十个UploadFile同时打开，文件描述符会很快耗尽。我在测试时一次传 50 个小文件，默认系统文件描述符是 1024，一般没问题；但如果是每个文件都做了await file.read()且没及时关闭，就很容易触到上限。
2.2 分块写入的推荐姿势
我处理上传文件的固定套路是这样的：先给文件生成一个内部唯一名，用uuid避免重名，然后把UploadFile里的内容分块写入目标路径。这样即使后续处理失败，也不会污染用户的原始文件名。
import shutil import uuid from pathlib import Path UPLOAD_DIR = Path("/tmp/upload_storage") UPLOAD_DIR.mkdir(exist_ok=True) @app.post("/upload-safe") async def upload_safe(file: UploadFile = File(...)): internal_name = f"{uuid.uuid4().hex}_{file.filename}" save_path = UPLOAD_DIR / internal_name with save_path.open("wb") as buffer: shutil.copyfileobj(file.file, buffer, length=1024 * 1024) return { "internal_name": internal_name, "original_name": file.filename, "saved_path": str(save_path), }重点就在shutil.copyfileobj(file.file, buffer, length=1024*1024)。它会以 1MB 为单位循环读取源文件并写入目标文件，整个过程内存占用很稳定，不会随文件大小线性增长。我自己在当前机器上实测过，55MB 的文件用这种方式处理，峰值内存只增加了几十 MB。
这里还有个小细节容易被忽略：如果前端上传时设置了Accept-Encoding或使用某些压缩上传组件，UploadFile.file读出来的可能不是原始字节，而是压缩流。遇到这种情况，要么前端统一不开启上传压缩，要么后端先做解压处理，不然落盘的文件是损坏的。这个坑非常隐蔽，排查起来特别浪费时间。
2.3 文件名安全与类型校验
文件名安全是老生常谈，但每次都必须再强调一遍，因为真的会出事。用户上传的文件名可能包含../、绝对路径、特殊字符。如果直接把这个名字拼到路径里，轻则文件被存到错误位置，重则被利用来覆盖服务端其他文件。
我处理文件名的规则是三个“绝不”：
- 绝不直接用
file.filename做存储文件名； - 绝不在拼接路径时信任原始文件名；
- 绝不把原始文件名打印到日志里而不做转义。
实际做法是：存储文件名用uuid，原始文件名单独存一份用于下载时展示。上面代码里的internal_name = f"{uuid.uuid4().hex}_{file.filename}"还有一个隐患：如果file.filename是空字符串或包含路径分隔符，仍可能出问题。更稳妥的做法是只保留纯文件名部分：
from pathlib import Path safe_name = Path(file.filename).name # 去掉所有路径部分 if not safe_name or safe_name == ".": safe_name = "unnamed" internal_name = f"{uuid.uuid4().hex}_{safe_name}"类型校验方面，扩展名不能信。比如用户把一段 HTML 命名为photo.jpg上传，某些场景下可能被当作图片直接渲染，进而带来 XSS 风险。所以后端的校验应该分两层：第一层看扩展名白名单，第二层看文件头（magic bytes）。文件头校验我会在最后一节展开说。
3. 下载环节：StreamingResponse 与可续传下载
3.1 为什么不能直接 return 一个 bytes
到了下载环节，最常见的错误写法是把文件整个读进内存然后直接返回：
data = Path("output.zip").read_bytes() # 整个文件读进内存 return Response(content=data, media_type="application/zip")小文件这样写完全能跑，但文件一大，服务端内存会直接被打满。更准确地说，就算内存能扛住，响应也需要等到整个文件准备好才发给客户端，用户的感受就是“转圈很久，然后突然开始下载”。这个体验非常糟糕。
正确的姿势是用StreamingResponse配合一个生成器，让数据分块从磁盘流向客户端，而不是一次性进内存。这在 FastAPI 中实现起来非常直白：
from fastapi.responses import StreamingResponse def file_chunk_generator(file_path: str): with open(file_path, "rb") as f: while True: chunk = f.read(64 * 1024) if not chunk: break yield chunk @app.get("/download/{internal_name}") async def download_file(internal_name: str): file_path = UPLOAD_DIR / internal_name return StreamingResponse( file_chunk_generator(str(file_path)), media_type="application/octet-stream", headers={ "Content-Disposition": f'attachment; filename="{internal_name}"' }, )3.2 分块读盘的生成器：为什么每次读 64KB
f.read(64 * 1024)这里每次读 64KB。为什么是这个值？太小（比如 4KB）会产生大量小系统调用，CPU 开销偏高；太大（比如 8MB）则内存占用上升，而且遇到网络慢时，一个 chunk 要等很久才发出去，反而造成延迟。我试过从 16KB 到 1MB 的区间，64KB 到 256KB 是性能和内存比较平衡的范围。个人习惯用 64KB，移动端弱网环境下也比较稳。
还需要注意media_type和Content-Disposition的配合。如果你希望浏览器弹出下载框，而不是直接打开预览，不要写media_type="application/octet-stream"之外的值，并确保Content-Disposition里带上了attachment关键字。
如果原始文件名有中文，光用filename="xxx"大概率会乱码。标准做法是用filename*编码格式：
from urllib.parse import quote encoded_name = quote(safe_name) headers = { "Content-Disposition": f"attachment; filename*=UTF-8''{encoded_name}" }这个写法的好处是兼容现代浏览器，中文、日文、emoji 文件名都能正常显示。
3.3 Range 头与 206 状态码
很多播放器、下载工具、浏览器下载管理器都支持断点续传，它们会发送一个Range请求头，例如：
Range: bytes=1024-2047意思是“从第 1024 字节开始，读到第 2047 字节”。如果服务端支持，应该返回206 Partial Content，并带上Content-Range头。如果你直接返回200 OK和完整内容，部分下载工具会校验失败，导致整个文件重新下载。
这里先明确一个前置事实：FastAPI 生态里，FileResponse本身就支持 Range 请求，可以直接用。如果你只是想快速实现下载，用FileResponse最省事：
from fastapi.responses import FileResponse @app.get("/download-simple/{internal_name}") async def download_simple(internal_name: str): file_path = UPLOAD_DIR / internal_name return FileResponse( path=str(file_path), filename=internal_name, media_type="application/octet-stream", )但为了让你真正理解 Range 的底层逻辑，我建议手动用StreamingResponse实现一遍。解析思路是这样的：读Range头，提取start和end，如果合法就打开文件时seek(start)，然后只生成对应长度的字节流；如果Range头不存在，就返回完整文件。
from fastapi import Request from fastapi.responses import JSONResponse, Response def parse_range(range_header: str, file_size: int): if not range_header or not range_header.startswith("bytes="): return None parts = range_header[6:].split("-", 1) try: start = int(parts[0]) if parts[0] else 0 end = int(parts[1]) if len(parts) > 1 and parts[1] else file_size - 1 except ValueError: return None if start >= file_size or end < start: return None end = min(end, file_size - 1) return start, end @app.get("/download-range/{internal_name}") async def download_range(internal_name: str, request: Request): file_path = UPLOAD_DIR / internal_name file_size = file_path.stat().st_size range_info = parse_range(request.headers.get("range"), file_size) if range_info: start, end = range_info chunk_size = end - start + 1 headers = { "Content-Range": f"bytes {start}-{end}/{file_size}", "Accept-Ranges": "bytes", "Content-Length": str(chunk_size), } return StreamingResponse( file_chunk_generator(str(file_path), start=start, length=chunk_size), status_code=206, headers=headers, ) return StreamingResponse( file_chunk_generator(str(file_path)), media_type="application/octet-stream", headers={ "Accept-Ranges": "bytes", "Content-Length": str(file_size), }, )对应的生成器要支持偏移量：
def file_chunk_generator(file_path: str, start: int = 0, length: int | None = None): with open(file_path, "rb") as f: f.seek(start) remaining = length while True: if remaining is None: chunk = f.read(64 * 1024) else: chunk = f.read(min(64 * 1024, remaining)) if not chunk: break if remaining is not None: remaining -= len(chunk) yield chunk这一段代码实现的就是断点续传的核心逻辑。实际下载器、视频播放器的进度恢复、多线程下载资源，都依赖这个机制。
4. 耗时任务与状态追踪：不能让用户干等
4.1 BackgroundTasks 能做和不能做的事
如果你的“文件处理”只是转码前做个压缩，可能要几秒甚至几十秒。这时候如果让接口同步处理完再返回，前端就会一直干等，甚至触发超时。一个常见方案是使用 FastAPI 自带的BackgroundTasks：
from fastapi import BackgroundTasks @app.post("/upload/process") async def upload_process( file: UploadFile = File(...), background_tasks: BackgroundTasks = None, ): task_id = str(uuid.uuid4()) save_path = UPLOAD_DIR / f"{task_id}.tmp" with save_path.open("wb") as buffer: shutil.copyfileobj(file.file, buffer) background_tasks.add_task(process_file, task_id, str(save_path)) return {"task_id": task_id, "status": "queued"}这个写法能让接口立即返回task_id，后台慢慢处理。但这里有一个非常重要的认知边界：BackgroundTasks是在响应发送之后，由同一个进程执行的后台任务。它适合跑轻量任务，比如发通知邮件、清理临时文件、写日志。
但它不适合跑 CPU 密集型任务。因为在异步服务里，BackgroundTasks如果内部用了同步阻塞调用（比如 CPU 密集的图像处理循环），它会阻塞整个事件循环，导致所有并发请求都卡住。这个问题的表现形式非常诡异：明明只有一个任务在处理，其他用户却集体超时。
如果你确实需要在 FastAPI 进程里跑重任务，我建议把耗时逻辑丢进线程池执行：
import asyncio async def process_file(task_id: str, input_path: str): loop = asyncio.get_running_loop() # 在线程池里跑 CPU 密集型同步函数，避免阻塞事件循环 await loop.run_in_executor(None, heavy_cpu_work, task_id, input_path)不过话说回来，真正的生产环境如果任务量很大，更推荐用独立任务队列加独立 worker 进程。FastAPI 侧只管投递任务，worker 进程去处理。这种方式和 FastAPI 本身的耦合更低，也更容易扩展。这篇先不展开队列方案，但我会在后面提一句优化方向。
4.2 任务 ID 与轮询接口的设计
用户拿到task_id之后，怎么知道任务到哪一步了？常见做法是提供一个状态查询接口，前端定时轮询。内存字典存状态是最简单的方案：
import time TASK_STORE: dict[str, dict] = {} def heavy_cpu_work(task_id: str, input_path: str): try: TASK_STORE[task_id] = {"status": "running", "progress": 20, "message": "正在读取文件"} time.sleep(2) TASK_STORE[task_id] = {"status": "running", "progress": 60, "message": "正在压缩"} # 模拟耗时处理 time.sleep(3) result_path = input_path.replace(".tmp", ".result") TASK_STORE[task_id] = {"status": "done", "progress": 100, "message": "处理完成", "result_path": result_path} except Exception as exc: TASK_STORE[task_id] = {"status": "failed", "message": str(exc)} @app.get("/tasks/{task_id}") async def get_task_status(task_id: str): task = TASK_STORE.get(task_id) if not task: return JSONResponse(status_code=404, content={"detail": "task not found"}) result = { "status": task["status"], "progress": task.get("progress", 0), "message": task.get("message", ""), } if task.get("result_path"): result["download_url"] = f"/download-range/{Path(task['result_path']).name}" return result这个轮询接口设计很经典：创建任务时返回task_id，前端拿着这个 ID 每 1 到 2 秒调一次查询接口，根据status决定下一步。轮询间隔我一般建议不小于 1 秒，频率太高会给服务端带来无谓压力，太低用户又会觉得进度慢。
4.3 并发场景下的状态更新
用内存字典存状态，在单进程单实例的 Demo 里完全够用，但要注意两个并发相关问题：
第一，如果服务部署了多个 worker 进程（比如uvicorn --workers 4），每个进程的TASK_STORE是独立的。用户第一次请求落在 worker 1，查询状态却命中 worker 2，就会查不到任务。解决方案有几种：一是限制为单 worker 部署；二是用 Redis 等外部存储；三是把任务状态写进数据库。对生产环境，我建议至少把状态存到 Redis，这样才真正可靠。
第二，如果同一个用户连续提交两次任务，TASK_STORE里会有两个不同的task_id，它们互不干扰，但用户界面需要做去重。这块可以在前端处理：提交按钮提交后置为 loading，有新的task_id之后只记录最新的那个。
再补充一个细节：TASK_STORE里的result_path不需要长期保留。任务处理完成、用户下载成功后，最好把记录删掉，否则时间长了内存持续上涨，而且旧文件一直占着磁盘。我习惯在下载接口里做一次“下载后清理”，或者写一个定时清理脚本，只保留最近几小时的任务记录和文件。
5. 上生产前必查的清单
5.1 临时目录与磁盘清理
开发环境下，/tmp/upload_storage这种目录怎么用都行。但生产环境有几个必须处理的问题：
临时文件不会自己消失。我处理过一台小机器，一个月后磁盘被上传的临时文件塞满了，服务直接挂掉。要解决这个问题，不要指望手动清理，写一个简单的定时清理更可靠：
import os import time def clean_old_files(storage_dir: str, max_age_seconds: int = 3600): now = time.time() for name in os.listdir(storage_dir): path = os.path.join(storage_dir, name) if os.path.isfile(path) and now - os.path.getmtime(path) > max_age_seconds: os.remove(path)把这个函数做成定时任务，每小时跑一次，保留最近一小时的文件，其余通通删掉。注意如果你的任务处理耗时会超过一小时，要把max_age_seconds调大一点，否则用户处理到一半文件被清了。
另一个磁盘问题是大文件落盘时的“半截文件”。上传中断时，pathlib的.open("wb")会留下一个不完整的文件。最简单的兜底方案是：先写到一个临时后缀文件，全部写完后再os.rename成最终文件名。这样至少保证目录里出现的文件都是完整的：
temp_path = UPLOAD_DIR / f"{task_id}.tmp" final_path = UPLOAD_DIR / f"{task_id}.result" # 写完整后再改名 os.replace(temp_path, final_path)5.2 反向代理的超时与体积限制
绝大多数 FastAPI 应用上线时前面会挂一个反向代理（Nginx、Caddy 这类）。这里有两个经典问题：上传体积限制、请求超时时间。
Nginx 默认对请求体大小有限制，通常写着几十 MB。你直接传 700MB 的压缩包，请求会在代理层被拒掉，错误信息是413 Request Entity Too Large。需要显式配置：
client_max_body_size 2g;反向代理的超时也很关键。如果上传一个很大的文件需要很久，Nginx 默认的proxy_read_timeout只有 60 秒，很可能传到一半连接就被掐断。我一般会设置：
proxy_connect_timeout 60s; proxy_send_timeout 300s; proxy_read_timeout 300s; client_body_timeout 300s;我把这些数值按需调大：读超时和写超时都拉到 300 秒以上，上传接口才稳定。但注意，调大超时不是解决“任务处理慢”的办法。真正的处理逻辑应该走后台任务，接口本身必须快速响应，否则连接长时间挂着，代理和客户端两边的资源都会持续占用。
5.3 别信扩展名，要信文件的头几个字节
上传文件的类型校验只靠扩展名是远远不够的。一个.jpg文件完全可能是个脚本或者恶意内容。稳妥的做法是读取文件的前几个字节，和已知格式的魔数比对。常见格式的文件头：
| 格式 | 文件头（十六进制） |
|---|---|
| JPEG | FF D8 FF |
| PNG | 89 50 4E 47 |
| GIF | 47 49 46 38 |
25 50 44 46 | |
| ZIP | 50 4B 03 04 |
实现不复杂，读文件头时不要一次读太多字节，读 8 到 16 字节就够：
def check_magic_bytes(file_path: str, expected_hex: str): with open(file_path, "rb") as f: header = f.read(8) return header.hex().upper().startswith(expected_hex)这样做有个额外好处：很多用户在本地改扩展名，上传后服务端一眼就能识别出真实类型，避免后续处理环节因为格式不符而崩溃。注意这只是一个第一层过滤，不是安全银弹，但对于大多数普通应用来说已经足够挡住大部分粗心或低水平的投毒文件。
如果业务上允许的文件类型比较多样，可以用现成的第三方库做更完整的嗅探，效果更稳定。另外，如果服务器允许用户上传 HTML/SVG 这种带脚本能力的文件，一定要在下载或预览环节加上Content-Disposition: attachment以及X-Content-Type-Options: nosniff响应头，防止浏览器直接渲染执行。
关于文件处理这块，我还有几个建议：UploadFile 用完之后，要主动关闭file.file，否则临时文件会一直占着文件描述符；日志里记录文件名时做截断，避免原始文件名里的特殊字符破坏日志格式；最后，所有对外暴露的下载接口都要考虑“按需生成”的场景，不要把所有处理好的文件无限期留存在磁盘里。
我自己做这类文件中转服务最大的体会是：能用小内存方案就绝不整文件读，能异步就绝不同步卡接口，能流式响应就绝不先拼完整字符串。这三个判断一旦想清楚，后面大部分问题都能提前避开。
