在 Web 应用开发中，文件上传功能是用户交互的重要组成部分，无论是上传图片、文档，还是其他类型的文件，都需要一个可靠且高效的解决方案。FastAPI 作为高性能的 Python Web 框架，凭借其简洁的语法和强大的功能，为文件上传提供了优雅且高效的实现方式。本文将深入探讨 FastAPI 中文件上传的各种场景与实践，带你从基础入门到进阶应用，全面掌握这一核心功能
一、FastAPI 文件上传基础：单文件上传
在 FastAPI 中，处理文件上传的关键在于File和UploadFile这两个工具。UploadFile是对SpooledTemporaryFile的封装，它支持异步操作，并且在性能和资源管理上表现出色，非常适合处理 HTTP 请求中的文件数据
1. 简单单文件上传示例
from fastapi import FastAPI, File, UploadFile
app = FastAPI()
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
return {"filename": file.filename}在上述代码中：
@app.post("/uploadfile/")定义了一个处理 POST 请求的路由，路径为/uploadfile/。file: UploadFile = File(...)表示该接口接收一个文件，File(...)强调这是一个必填参数。- 接口返回值仅包含上传文件的文件名
file.filename，实际应用中可根据需求进行更复杂的操作。
2. 读取文件内容
获取文件名只是第一步，更多时候我们需要读取文件内容。FastAPI 的 UploadFile 支持异步读取，示例如下：
from fastapi import FastAPI, File, UploadFile
app = FastAPI()
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
contents = await file.read()
file_size = len(contents)
return {
"filename": file.filename,
"file_size": file_size,
"content_type": file.content_type
}这里通过await file.read()异步读取文件内容，计算文件大小，并获取文件的content_type（文件类型），以便对文件进行更细致的处理和验证
3. 保存上传文件
实际应用中，我们通常需要将上传的文件保存到服务器。以下是一个将文件保存到本地指定目录的示例：
import os
from fastapi import FastAPI, File, UploadFile
app = FastAPI()
UPLOAD_DIR = "uploads"
os.makedirs(UPLOAD_DIR, exist_ok=True)
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
file_path = os.path.join(UPLOAD_DIR, file.filename)
with open(file_path, "wb") as buffer:
buffer.write(await file.read())
return {"message": "File saved successfully", "file_path": file_path}代码中先创建保存文件的目录，再将文件内容写入到指定路径，确保文件能够持久化存储在服务器上
二、进阶应用：多文件上传与大文件处理
1. 多文件上传
当需要同时上传多个文件时，只需将UploadFile定义为列表形式，示例如下：
from fastapi import FastAPI, File, UploadFile
app = FastAPI()
@app.post("/uploadfiles/")
async def create_upload_files(files: list[UploadFile] = File(...)):
file_infos = []
for file in files:
contents = await file.read()
file_infos.append({
"filename": file.filename,
"file_size": len(contents)
})
return {"uploaded_files": file_infos}通过接收list[UploadFile]类型的参数，接口可以处理多个文件，遍历文件列表并获取每个文件的信息，实现多文件上传的功能
2. 大文件处理
对于大文件上传，直接一次性读取文件内容可能会导致内存不足。FastAPI 结合 Python 的文件操作，可以采用分块读取的方式处理大文件。以下是一个简单的分块读取示例：
from fastapi import FastAPI, File, UploadFile
app = FastAPI()
@app.post("/uploadlargefile/")
async def create_upload_large_file(file: UploadFile = File(...)):
CHUNK_SIZE = 1024 * 1024 # 1MB
file_size = 0
with open("large_file.tmp", "wb") as buffer:
while True:
chunk = await file.read(CHUNK_SIZE)
if not chunk:
break
buffer.write(chunk)
file_size += len(chunk)
return {"message": "Large file uploaded successfully", "file_size": file_size}此示例中，每次读取固定大小（1MB）的文件块写入临时文件，直到文件读取完毕，有效降低了内存占用，适合处理大文件上传场景
三、文件上传的验证与安全
1. 文件类型验证
为了确保上传的文件符合要求，我们可以对文件类型进行验证。一种简单的方式是通过文件扩展名或content_type进行判断：
from fastapi import FastAPI, File, UploadFile, HTTPException
app = FastAPI()
ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png"}
ALLOWED_CONTENT_TYPES = {"image/jpeg", "image/png"}
def is_allowed_file(filename, content_type):
return (
any(filename.endswith(ext) for ext in ALLOWED_EXTENSIONS)
and content_type in ALLOWED_CONTENT_TYPES
)
@app.post("/uploadimage/")
async def create_upload_image(file: UploadFile = File(...)):
if not is_allowed_file(file.filename, file.content_type):
raise HTTPException(status_code=400, detail="Invalid file type")
return {"message": "Image uploaded successfully", "filename": file.filename}2. 安全注意事项
- 文件重命名：为避免文件名冲突和安全隐患，建议对上传的文件进行重命名，例如使用 UUID 作为文件名。
- 存储路径限制：严格限制文件的保存路径，防止用户通过文件名或路径进行恶意操作（如目录穿越攻击）。
- 病毒扫描：在生产环境中，对上传文件进行病毒扫描是保障系统安全的重要措施，可以集成第三方病毒扫描服务实现。