如果你能一路刷到这篇 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_okTrue) 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, length1024 * 1024) return { internal_name: internal_name, original_name: file.filename, saved_path: str(save_path), }重点就在shutil.copyfileobj(file.file, buffer, length1024*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(contentdata, media_typeapplication/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_typeapplication/octet-stream, headers{ Content-Disposition: fattachment; filename{internal_name} }, )3.2 分块读盘的生成器为什么每次读 64KBf.read(64 * 1024)这里每次读 64KB。为什么是这个值太小比如 4KB会产生大量小系统调用CPU 开销偏高太大比如 8MB则内存占用上升而且遇到网络慢时一个 chunk 要等很久才发出去反而造成延迟。我试过从 16KB 到 1MB 的区间64KB 到 256KB 是性能和内存比较平衡的范围。个人习惯用 64KB移动端弱网环境下也比较稳。还需要注意media_type和Content-Disposition的配合。如果你希望浏览器弹出下载框而不是直接打开预览不要写media_typeapplication/octet-stream之外的值并确保Content-Disposition里带上了attachment关键字。如果原始文件名有中文光用filenamexxx大概率会乱码。标准做法是用filename*编码格式from urllib.parse import quote encoded_name quote(safe_name) headers { Content-Disposition: fattachment; filename*UTF-8{encoded_name} }这个写法的好处是兼容现代浏览器中文、日文、emoji 文件名都能正常显示。3.3 Range 头与 206 状态码很多播放器、下载工具、浏览器下载管理器都支持断点续传它们会发送一个Range请求头例如Range: bytes1024-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( pathstr(file_path), filenameinternal_name, media_typeapplication/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: fbytes {start}-{end}/{file_size}, Accept-Ranges: bytes, Content-Length: str(chunk_size), } return StreamingResponse( file_chunk_generator(str(file_path), startstart, lengthchunk_size), status_code206, headersheaders, ) return StreamingResponse( file_chunk_generator(str(file_path)), media_typeapplication/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 自带的BackgroundTasksfrom 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_code404, 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文件完全可能是个脚本或者恶意内容。稳妥的做法是读取文件的前几个字节和已知格式的魔数比对。常见格式的文件头格式文件头十六进制JPEGFF D8 FFPNG89 50 4E 47GIF47 49 46 38PDF25 50 44 46ZIP50 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否则临时文件会一直占着文件描述符日志里记录文件名时做截断避免原始文件名里的特殊字符破坏日志格式最后所有对外暴露的下载接口都要考虑“按需生成”的场景不要把所有处理好的文件无限期留存在磁盘里。我自己做这类文件中转服务最大的体会是能用小内存方案就绝不整文件读能异步就绝不同步卡接口能流式响应就绝不先拼完整字符串。这三个判断一旦想清楚后面大部分问题都能提前避开。
