本文目录导读:
我将通过一个文件上传案例来对比 Flask 和 FastAPI 的处理方式,包括代码、异步支持和文档生成等方面的差异。
Flask 实现文件上传
# flask_upload.py
from flask import Flask, request, jsonify, render_template_string
import os
from werkzeug.utils import secure_filename
import time
app = Flask(__name__)
app.config['UPLOAD_FOLDER'] = 'uploads'
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 # 16MB
# 确保上传目录存在
os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
# 简单的上传页面
UPLOAD_HTML = '''
<!DOCTYPE html>
<html>
<head>Flask 文件上传</title>
</head>
<body>
<h1>Flask 文件上传示例</h1>
<form method="POST" action="/upload" enctype="multipart/form-data">
<input type="file" name="file" required>
<input type="submit" value="上传">
</form>
</body>
</html>
'''
@app.route('/')
def index():
return render_template_string(UPLOAD_HTML)
@app.route('/upload', methods=['POST'])
def upload_file():
start_time = time.time()
# 检查是否有文件
if 'file' not in request.files:
return jsonify({'error': '没有文件部分'}), 400
file = request.files['file']
# 检查文件名是否为空
if file.filename == '':
return jsonify({'error': '没有选择文件'}), 400
# 保存文件
if file:
filename = secure_filename(file.filename)
filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename)
# Flask 是同步的,大文件上传会阻塞
file.save(filepath)
process_time = time.time() - start_time
return jsonify({
'message': '文件上传成功',
'filename': filename,
'size': os.path.getsize(filepath),
'process_time': f'{process_time:.3f}s'
})
if __name__ == '__main__':
app.run(debug=True)
FastAPI 实现文件上传
# fastapi_upload.py
from fastapi import FastAPI, File, UploadFile, HTTPException, Form
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
import os
import shutil
import time
from typing import Optional
app = FastAPI(title="FastAPI 文件上传示例")
# 确保上传目录存在
UPLOAD_FOLDER = 'uploads'
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
@app.get("/", response_class=HTMLResponse)
async def index():
"""返回上传页面"""
html_content = '''
<!DOCTYPE html>
<html>
<head>
<title>FastAPI 文件上传</title>
</head>
<body>
<h1>FastAPI 文件上传示例</h1>
<form method="POST" action="/upload" enctype="multipart/form-data">
<input type="file" name="file" required>
<input type="text" name="description" placeholder="文件描述">
<input type="submit" value="上传">
</form>
</body>
</html>
'''
return html_content
@app.post("/upload")
async def upload_file(
file: UploadFile = File(...),
description: Optional[str] = Form(None) # 额外的表单字段
):
"""
上传文件
- 支持异步处理
- 自动验证文件类型
- 支持额外字段
"""
start_time = time.time()
# 验证文件类型
allowed_types = {'image/jpeg', 'image/png', 'application/pdf', 'text/plain'}
if file.content_type not in allowed_types:
raise HTTPException(
status_code=400,
detail=f"不支持的文件类型: {file.content_type}"
)
# 生成安全的文件名
safe_filename = file.filename.replace(" ", "_")
filepath = os.path.join(UPLOAD_FOLDER, safe_filename)
try:
# 异步读取并写入文件
with open(filepath, "wb") as buffer:
# 逐块读取,不会阻塞
while chunk := await file.read(1024 * 1024): # 1MB chunks
buffer.write(chunk)
except Exception as e:
raise HTTPException(status_code=500, detail=f"文件保存失败: {str(e)}")
process_time = time.time() - start_time
return {
"message": "文件上传成功",
"filename": safe_filename,
"original_name": file.filename,
"content_type": file.content_type,
"size": f"{os.path.getsize(filepath) / 1024:.2f} KB",
"description": description,
"process_time": f"{process_time:.3f}s"
}
@app.post("/upload-multiple")
async def upload_multiple_files(
files: list[UploadFile] = File(...)
):
"""上传多个文件"""
results = []
for file in files:
safe_filename = file.filename.replace(" ", "_")
filepath = os.path.join(UPLOAD_FOLDER, safe_filename)
with open(filepath, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
results.append({
"filename": safe_filename,
"size": f"{os.path.getsize(filepath) / 1024:.2f} KB"
})
return {"uploaded_files": results, "count": len(results)}
if __name__ == '__main__':
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
关键对比分析
1 异步处理
# Flask - 同步阻塞
@app.route('/upload', methods=['POST'])
def upload_file():
file = request.files['file']
file.save(filepath) # 大文件上传时阻塞整个应用
# FastAPI - 异步非阻塞
@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
while chunk := await file.read(1024 * 1024): # 异步读取
buffer.write(chunk) # 不会阻塞其他请求
2 文件处理 API
# Flask - 使用 werkzeug from werkzeug.utils import secure_filename file = request.files['file'] # FileStorage 对象 filename = secure_filename(file.filename) # 需要手动清理文件名 # FastAPI - 使用 Starlette from fastapi import UploadFile file: UploadFile = File(...) # 直接获取上传文件 safe_filename = file.filename # 自动处理
3 错误处理
# Flask - 返回元组或使用 abort
return jsonify({'error': '文件类型错误'}), 400
# 或
abort(400, '文件类型错误')
# FastAPI - 抛出 HTTPException
from fastapi import HTTPException
raise HTTPException(status_code=400, detail="不支持的文件类型")
4 自动 API 文档
# Flask - 需要手动创建文档 # 没有自动生成的 API 文档 # FastAPI - 自动生成 Swagger 文档 # 访问 http://localhost:8000/docs 查看 # 访问 http://localhost:8000/redoc 查看
测试代码对比
# test_upload.py
import requests
import time
def test_flask_upload():
"""测试 Flask 上传"""
files = {'file': ('test.txt', b'Hello Flask!')}
start = time.time()
response = requests.post('http://localhost:5000/upload', files=files)
elapsed = time.time() - start
print(f"Flask: {response.json()}, 耗时: {elapsed:.3f}s")
def test_fastapi_upload():
"""测试 FastAPI 上传"""
files = {'file': ('test.txt', b'Hello FastAPI!')}
data = {'description': '测试文件'}
start = time.time()
response = requests.post('http://localhost:8000/upload',
files=files, data=data)
elapsed = time.time() - start
print(f"FastAPI: {response.json()}, 耗时: {elapsed:.3f}s")
Flask vs FastAPI 特点总结
| 特性 | Flask | FastAPI |
|---|---|---|
| 异步支持 | 需要额外安装插件 | 原生异步支持 |
| 自动文档 | 需第三方库 (Flask-RESTful) | 自动生成 Swagger/ReDoc |
| 性能 | 同步,大文件时需谨慎 | 异步,高并发表现优秀 |
| 文件验证 | 手动实现 | 内置类型验证 |
| 学习曲线 | 简单直观 | 中等(需理解异步) |
| 社区生态 | 成熟,插件丰富 | 快速发展中 |
| 类型提示 | 可选 | 强制且自动验证 |
选择建议
选择 Flask 当:
- 项目简单,不需要高并发
- 团队对 Flask 更熟悉
- 需要大量成熟的第三方插件
- 传统 MVC 架构
选择 FastAPI 当:
- 需要高性能和异步处理
- 需要自动 API 文档
- 团队接受异步编程
- 微服务架构
- 需要类型安全
两种框架都能很好地处理文件上传,选择取决于项目具体需求和技术栈偏好,FastAPI 在现代 API 开发中越来越受欢迎,特别是在需要高性能和自动文档的场景下。