在使用FastAPI进行文件上传开发时，看似简单的操作背后隐藏着诸多容易踩中的“陷阱”。从大文件上
传导致的内存溢出，到文件名安全引发的路径遍历风险，再到临时文件堆积造成的磁盘告警，每一个问
题都可能影响系统的稳定性与安全性。本文将围绕分块存盘、类型校验与安全兜底三个核心维度，为您
详细解析FastAPI文件上传中的常见问题及解决方案，帮助您构建稳健、安全的文件上传功能。
一、分块存盘：破解大文件上传难题
1. 传统上传方式的痛点
当用户上传大文件时，若采用FastAPI默认的file.read()方法一次性读取文件内容，会将整个文件加载到
服务器内存中。一旦文件大小达到数GB级别，服务器内存会被瞬间耗尽，不仅会导致当前请求处理失败，
还可能影响其他服务的正常运行。此外，网络不稳定时，大文件上传一旦中断，就需要重新上传全部内容，
严重影响用户体验。
2. 分块上传的实现思路
分块上传通过将大文件分割成多个固定大小的小块（通常为2MB - 10MB），分批次传输到服务器，最后
在服务器端将这些小块合并成完整文件。这种方式不仅能有效降低服务器内存压力，还支持断点续传，即
某个块传输失败时，只需重新传输该块即可，无需整个文件重新上传。
3. 前端与后端代码实现
前端实现
前端需要将文件分割成指定大小的块，并为每个块添加标识信息，如文件唯一ID、块索引、总块数等。以
下是一个简单的JavaScript示例：
function uploadFile(file) {
const chunkSize = 2 * 1024 * 1024; // 2MB
const totalChunks = Math.ceil(file.size / chunkSize);
const fileId = Math.random().toString(36).substring(2);
for (let i = 0; i < totalChunks; i++) {
const start = i * chunkSize;
const end = Math.min(start + chunkSize, file.size);
const chunk = file.slice(start, end);
const formData = new FormData();
formData.append('file_id', fileId);
formData.append('chunk_index', i);
formData.append('total_chunks', totalChunks);
formData.append('chunk', chunk);
fetch('/uploadchunk/', {
method: 'POST',
body: formData
});
}
}
后端实现
FastAPI后端需要接收每个文件块并临时存储，验证块的完整性和顺序，所有块上传完成后合并文件。以下是核心代码示例：
from fastapi import FastAPI, File, Form, UploadFile
import os
from pathlib import Path
app = FastAPI()
TEMP_DIR = Path("temp_chunks")
UPLOAD_DIR = Path("uploads")
TEMP_DIR.mkdir(exist_ok=True)
UPLOAD_DIR.mkdir(exist_ok=True)
@app.post("/uploadchunk/")
async def upload_chunk(
file_id: str = Form(...),
chunk_index: int = Form(...),
total_chunks: int = Form(...),
chunk: UploadFile = File(...)
):
# 保存块到临时目录
chunk_path = TEMP_DIR / f"{file_id}_{chunk_index}"
with open(chunk_path, "wb") as f:
f.write(await chunk.read())
# 检查是否所有块都已上传
uploaded_chunks = [f for f in TEMP_DIR.glob(f"{file_id}_*")]
if len(uploaded_chunks) == total_chunks:
# 合并所有块
with open(UPLOAD_DIR / file_id, "wb") as outfile:
for i in range(total_chunks):
chunk_path = TEMP_DIR / f"{file_id}_{i}"
with open(chunk_path, "rb") as infile:
outfile.write(infile.read())
# 删除临时块文件
os.remove(chunk_path)
return {"message": "文件上传完成", "file_id": file_id}
return {"message": "块上传成功", "uploaded_chunks": len(uploaded_chunks)}
二、类型校验：筑牢文件安全第一道防线
1. 文件类型校验的重要性
若不对上传文件的类型进行校验，用户可能会上传恶意脚本、病毒文件等，给系统带来安全风险。例如，攻击者上传一个包含恶意代码的PHP文件，若服务器未进行类型校验并直接执行该文件，可能会导致服务器被控制。
2. 实现文件类型校验
在FastAPI中，可以通过UploadFile对象的content_type属性获取文件的MIME类型，并与允许的文件类型列表进行对比。以下是一个示例：
from fastapi import FastAPI, File, UploadFile, HTTPException
from typing import List
app = FastAPI()
ALLOWED_TYPES = {"image/jpeg", "image/png", "image/gif", "application/pdf", "text/plain"}
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
if file.content_type not in ALLOWED_TYPES:
raise HTTPException(
status_code=400,
detail=f"文件类型不允许，仅支持{', '.join(ALLOWED_TYPES)}"
)
# 处理文件逻辑
return {"filename": file.filename, "content_type": file.content_type}
需要注意的是，仅通过content_type进行校验并不完全可靠，因为攻击者可以修改文件的MIME类型。更严谨的方式是结合文件内容进行校验，例如通过读取文件的前几个字节来判断文件的真实类型。
三、安全兜底：全方位规避潜在风险
1. 文件名安全处理
用户上传的文件名可能包含中文、特殊字符甚至恶意构造的路径（如../../../etc/passwd），直接使用原始文件名可能导致跨平台兼容性问题、路径遍历安全漏洞以及数据库存储异常。以下是一个文件名安全处理的示例：
from urllib.parse import quote
import re
def sanitize_filename(filename: str) -> str:
# 移除非法字符
clean = re.sub(r'[\\/*?:"<>|]', "", filename)
# 处理Unicode
return quote(clean.encode('utf-8')).replace("%", "_")
该函数会移除文件名中的非法字符，并对Unicode字符进行编码处理，确保文件名在不同操作系统和数据库中都能安全使用。
2. 临时文件管理
FastAPI默认会将上传文件存储在内存或磁盘临时目录，但不会自动清理这些临时文件。若不及时清理，临时文件会不断堆积，导致服务器磁盘空间不足。可以通过注册程序退出时的清理函数，结合定时任务扫描过期文件，构建自动化的临时文件清理系统。以下是一个示例：
import atexit
import glob
import os
import time
TEMP_DIR = "/tmp/fastapi_uploads"
os.makedirs(TEMP_DIR, exist_ok=True)
def cleanup():
for f in glob.glob(f"{TEMP_DIR}/*"):
try:
if os.path.isfile(f):
# 删除超过24小时的临时文件
if time.time() - os.path.getmtime(f) > 24 * 3600:
os.unlink(f)
except Exception as e:
print(f"清理失败 {f}: {e}")
atexit.register(cleanup)
3. 下载优化与安全
在提供文件下载功能时，直接使用FileResponse虽然简单，但面对大文件下载时缺乏断点续传支持和下载速度限制。可以通过实现StreamingResponse来支持断点续传，并限制下载速度，提升用户体验的同时，避免服务器因大文件下载而负载过高。以下是一个示例：
from fastapi import Request
from fastapi.responses import StreamingResponse
import os
def parse_range(range_header, file_size):
range_str = range_header.split('=')
start, end = range_str.split('-')
start = int(start) if start else 0
end = int(end) if end else file_size - 1
return start, end
async def smart_download(file_path: str, request: Request):
file_size = os.path.getsize(file_path)
range_header = request.headers.get("Range")
def chunk_generator():
with open(file_path, "rb") as f:
if range_header:
start, end = parse_range(range_header, file_size)
f.seek(start)
remaining = end - start + 1
while remaining > 0:
chunk_size = min(1024 * 1024, remaining)
yield f.read(chunk_size)
remaining -= chunk_size
else:
while chunk := f.read(1024 * 1024):
yield chunk
headers = {"Accept-Ranges": "bytes"}
if range_header:
start, end = parse_range(range_header, file_size)
headers["Content-Range"] = f"bytes {start}-{end}/{file_size}"
headers["Content-Length"] = str(end - start + 1)
return StreamingResponse(chunk_generator(), status_code=206, headers=headers)
else:
headers["Content-Length"] = str(file_size)
return StreamingResponse(chunk_generator(), headers=headers)
四、总结
FastAPI为文件上传提供了便捷的处理机制，但在实际开发中，需要充分考虑大文件上传、文件类型安全、文件名安全等诸多问题。通过分块存盘可以有效解决大文件上传带来的内存压力和中断重传问题，类型校验能阻止恶意文件上传，而安全兜底措施则从文件名处理、临时文件管理、下载优化等多个方面全方位保障系统的稳定性与安全性。希望本文的内容能帮助您在FastAPI文件上传开发中避开常见的“陷阱”，构建出更加稳健、安全的应用。