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