怎样通过一个文件上传案例对比Flask和FastAPI的处理方式

访客 全栈框架 1

本文目录导读:

  1. Flask 实现文件上传
  2. FastAPI 实现文件上传
  3. 关键对比分析
  4. 测试代码对比
  5. Flask vs FastAPI 特点总结
  6. 选择建议

我将通过一个文件上传案例来对比 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 开发中越来越受欢迎,特别是在需要高性能和自动文档的场景下。

标签: Flask FastAPI

抱歉,评论功能暂时关闭!