Flask文件上传 #

一、文件上传概述 #

1.1 上传原理 #

文件上传使用HTTP POST请求,表单的enctype必须设置为multipart/form-data

1.2 配置上传 #

python
app = Flask(__name__)

# 上传目录
app.config['UPLOAD_FOLDER'] = 'uploads'

# 最大文件大小(字节)
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024  # 16MB

# 允许的扩展名
ALLOWED_EXTENSIONS = {'txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif'}

二、基本文件上传 #

2.1 HTML表单 #

html
<form method="POST" enctype="multipart/form-data">
    <input type="file" name="file">
    <button type="submit">上传</button>
</form>

2.2 处理上传 #

python
import os
from flask import request, redirect, url_for
from werkzeug.utils import secure_filename

@app.route('/upload', methods=['GET', 'POST'])
def upload_file():
    if request.method == 'POST':
        if 'file' not in request.files:
            return '没有文件'
        
        file = request.files['file']
        
        if file.filename == '':
            return '没有选择文件'
        
        if file and allowed_file(file.filename):
            filename = secure_filename(file.filename)
            file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
            return '上传成功'
    
    return render_template('upload.html')

def allowed_file(filename):
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS

三、使用Flask-WTF #

3.1 文件表单 #

python
from flask_wtf import FlaskForm
from flask_wtf.file import FileField, FileAllowed, FileRequired

class UploadForm(FlaskForm):
    file = FileField('文件', validators=[
        FileRequired(message='请选择文件'),
        FileAllowed(['jpg', 'png', 'gif'], message='只允许图片文件')
    ])
    submit = SubmitField('上传')

3.2 处理上传 #

python
@app.route('/upload', methods=['GET', 'POST'])
def upload():
    form = UploadForm()
    
    if form.validate_on_submit():
        file = form.file.data
        filename = secure_filename(file.filename)
        file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
        return redirect(url_for('uploaded_file', filename=filename))
    
    return render_template('upload.html', form=form)

四、多文件上传 #

4.1 多文件表单 #

html
<form method="POST" enctype="multipart/form-data">
    <input type="file" name="files" multiple>
    <button type="submit">上传</button>
</form>

4.2 处理多文件 #

python
@app.route('/multi-upload', methods=['POST'])
def multi_upload():
    files = request.files.getlist('files')
    
    for file in files:
        if file and allowed_file(file.filename):
            filename = secure_filename(file.filename)
            file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
    
    return '上传成功'

4.3 使用Flask-WTF #

python
from flask_wtf.file import MultipleFileField

class MultiUploadForm(FlaskForm):
    files = MultipleFileField('文件', validators=[
        FileAllowed(['jpg', 'png'], '只允许图片')
    ])
    submit = SubmitField('上传')

@app.route('/multi-upload', methods=['GET', 'POST'])
def multi_upload():
    form = MultiUploadForm()
    
    if form.validate_on_submit():
        for file in form.files.data:
            filename = secure_filename(file.filename)
            file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
        return '上传成功'
    
    return render_template('multi_upload.html', form=form)

五、文件安全 #

5.1 secure_filename #

python
from werkzeug.utils import secure_filename

# 清理文件名
filename = secure_filename('../../../etc/passwd')
# 结果: 'etc_passwd'

filename = secure_filename('我的文件.jpg')
# 结果: '_.jpg'  (中文会被移除)

# 建议使用UUID作为文件名
import uuid
ext = os.path.splitext(file.filename)[1]
filename = str(uuid.uuid4()) + ext

5.2 文件类型验证 #

python
import imghdr

def validate_image(file):
    """验证图片类型"""
    header = file.read(512)
    file.seek(0)
    format = imghdr.what(None, header)
    if format not in ['jpeg', 'png', 'gif']:
        return False
    return True

def validate_file_type(file, allowed_types):
    """验证文件MIME类型"""
    import magic
    mime = magic.from_buffer(file.read(1024), mime=True)
    file.seek(0)
    return mime in allowed_types

5.3 文件大小验证 #

python
@app.route('/upload', methods=['POST'])
def upload():
    file = request.files['file']
    
    # 检查文件大小
    file.seek(0, os.SEEK_END)
    size = file.tell()
    file.seek(0)
    
    if size > 5 * 1024 * 1024:  # 5MB
        return '文件太大'
    
    # 保存文件
    filename = secure_filename(file.filename)
    file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
    return '上传成功'

六、图片处理 #

6.1 图片缩放 #

python
from PIL import Image

def resize_image(file, max_size=(800, 600)):
    """缩放图片"""
    img = Image.open(file)
    img.thumbnail(max_size)
    return img

@app.route('/upload-avatar', methods=['POST'])
def upload_avatar():
    file = request.files['avatar']
    
    if file and allowed_file(file.filename):
        img = resize_image(file, (200, 200))
        filename = str(uuid.uuid4()) + '.jpg'
        img.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
        return '上传成功'
    
    return '上传失败'

6.2 图片裁剪 #

python
def crop_image(file, box):
    """裁剪图片"""
    img = Image.open(file)
    cropped = img.crop(box)
    return cropped

6.3 生成缩略图 #

python
def create_thumbnail(file, size=(128, 128)):
    """创建缩略图"""
    img = Image.open(file)
    img.thumbnail(size)
    return img

七、文件访问 #

7.1 提供文件下载 #

python
from flask import send_from_directory

@app.route('/uploads/<filename>')
def uploaded_file(filename):
    return send_from_directory(app.config['UPLOAD_FOLDER'], filename)

# 指定下载文件名
@app.route('/download/<filename>')
def download_file(filename):
    return send_from_directory(
        app.config['UPLOAD_FOLDER'],
        filename,
        as_attachment=True,
        download_name='custom_name.txt'
    )

7.2 限制文件访问 #

python
from flask_login import login_required

@app.route('/uploads/<filename>')
@login_required
def uploaded_file(filename):
    return send_from_directory(app.config['UPLOAD_FOLDER'], filename)

八、云存储上传 #

8.1 上传到OSS #

python
import oss2

def upload_to_oss(file, filename):
    """上传到阿里云OSS"""
    auth = oss2.Auth('access_key', 'secret_key')
    bucket = oss2.Bucket(auth, 'endpoint', 'bucket_name')
    
    bucket.put_object(filename, file.read())
    
    return f'https://bucket_name.oss-cn-hangzhou.aliyuncs.com/{filename}'

8.2 上传到S3 #

python
import boto3

def upload_to_s3(file, filename, bucket_name):
    """上传到AWS S3"""
    s3 = boto3.client('s3')
    s3.upload_fileobj(file, bucket_name, filename)
    
    return f'https://{bucket_name}.s3.amazonaws.com/{filename}'

九、最佳实践 #

9.1 文件命名 #

python
import uuid
import os

def generate_filename(original_filename):
    """生成唯一文件名"""
    ext = os.path.splitext(original_filename)[1]
    return str(uuid.uuid4()) + ext

9.2 目录组织 #

python
import os
from datetime import datetime

def get_upload_path():
    """按日期组织目录"""
    today = datetime.now().strftime('%Y/%m/%d')
    path = os.path.join(app.config['UPLOAD_FOLDER'], today)
    os.makedirs(path, exist_ok=True)
    return path

9.3 异步处理 #

python
from celery import Celery

celery = Celery('tasks', broker='redis://localhost:6379/0')

@celery.task
def process_file(filepath):
    """异步处理文件"""
    pass

@app.route('/upload', methods=['POST'])
def upload():
    file = request.files['file']
    filename = save_file(file)
    
    # 异步处理
    process_file.delay(os.path.join(app.config['UPLOAD_FOLDER'], filename))
    
    return '上传成功'

十、总结 #

10.1 核心要点 #

要点 说明
表单设置 enctype=“multipart/form-data”
获取文件 request.files[‘name’]
安全处理 secure_filename()
文件验证 类型、大小检查
文件保存 file.save()
文件访问 send_from_directory()

10.2 下一步 #

现在你已经掌握了文件上传,接下来让我们学习 数据库概述,了解Flask数据库集成!

最后更新:2026-03-28