FastAPI教程-上传文件 作者：马育民 • 2026-08-14 14:55 • 阅读：10001 # 介绍 需要上传头像、照片、文件时，需要实现上传文件功能 # 上传文件的参数类型 视图函数的参数类型 `UploadFile`，默认值是：`File`： ``` async def upload_save(file: UploadFile = File(...)): ``` ### UploadFile 属性 - `.filename`：文件名 - `.file`：文件对象 - `.size`：大小 - `file.content_type`：MIME 类型，如 image/png ### UploadFile 方法 - `await file.read()`：读取全部字节 - `await file.read(size)`：读取指定大小分片 - `await file.close()`：关闭文件 # 例子1 ```python from fastapi import FastAPI,File,UploadFile import os # 保存附件的目录 save_dir = "./upload" # 目录不存在就创建 os.makedirs(save_dir, exist_ok=True) app = FastAPI() @app.post("/upload/save") async def upload_save(file: UploadFile = File(...)): out_path = os.path.join(save_dir, file.filename) # 写入磁盘 with open(out_path, "wb") as f: f.write(await file.read()) return {"saved_path": out_path, "filename": file.filename} ``` ### 测试 访问swagger：http://127.0.0.1:8000/docs ，按照下图操作： [](http://www.malaoshi.top/upload/0/0/1GW3rY6rYysi.png) 执行结果： [](http://www.malaoshi.top/upload/0/0/1GW3rY7Q8ZDm.png) ### 缺点 `await file.read()` 会一次性读入内存，大文件不推荐。 ### 解决 使用流式写入 # 例子2-流式写入 读一部分，写一部分，不是全部读取到内存 ``` from fastapi import FastAPI,File,UploadFile import os # 保存附件的目录 save_dir = "./upload" # 目录不存在就创建 os.makedirs(save_dir, exist_ok=True) # 1MB 分片 READ_SIZE = 1024 * 1024 app = FastAPI() @app.post("/upload/save") async def upload_save(file: UploadFile = File(...)): out_path = os.path.join(save_dir, file.filename) # 写入磁盘 with open(out_path, "wb") as f: # 读一部分，写一部分 while chunk := await file.read(READ_SIZE): f.write(chunk) return {"saved_path": out_path, "filename": file.filename} ``` ### 缺点 `with open(out_path, "wb") as f` 是同步磁盘 IO ，当执行 `f.write()` 写磁盘时，**会卡住整个协程事件循环**；高并发上传大文件时，**其他请求会被拖慢、延迟升高**。 ### 解决 - 把文件写入放到线程池（**缺点：**写起来有点复杂） - 用 aiofiles 异步文件库（**缺点：**需要安装库） # 用 aiofiles 异步写数据 ### 安装 ##### pip安装 ``` pip install aiofiles ``` ##### uv安装 ``` uv add aiofiles ``` ### 代码 ``` from fastapi import FastAPI,File,UploadFile import os import aiofiles # 保存附件的目录 save_dir = "./upload" # 目录不存在就创建 os.makedirs(save_dir, exist_ok=True) # 1MB 分片 READ_SIZE = 1024 * 1024 app = FastAPI() @app.post("/upload/save") async def upload_save(file: UploadFile = File(...)): out_path = os.path.join(save_dir, file.filename) # 异步写入磁盘 async with aiofiles.open(out_path, "wb") as f: while chunk := await file.read(READ_SIZE): await f.write(chunk) return {"saved_path": out_path, "filename": file.filename} ``` 原文出处：/show_1GW3rY7tbFGo.html