Flask 文件上传与下载完全指南
1. 引言
文件上传和下载是Web应用中常见的功能需求，Flask提供了简单而强大的方式来处理文件操作。本文将详细介绍如何在Flask应用中实现安全高效的文件上传和下载功能，包括基础实现、安全防护、性能优化等关键知识点。
2. 基础文件上传
2.1 简单文件上传表单
<!-- templates/upload.html -->
<form method="POST" enctype="multipart/form-data">
<input type="file" name="file">
<input type="submit" value="Upload">
</form>
2.2 Flask处理上传
from flask import Flask, request, redirect, url_for
from werkzeug.utils import secure_filename
import os
app = Flask(__name__)
app.config['UPLOAD_FOLDER'] = 'uploads'
os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
@app.route('/upload', methods=['GET', 'POST'])
def upload_file():
if request.method == 'POST':
if 'file' not in request.files:
return redirect(request.url)
file = request.files['file']
if file.filename == '':
return redirect(request.url)
if file:
filename = secure_filename(file.filename)
file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
return 'File uploaded successfully'
return '''
<!doctype html>
<title>Upload new File</title>
Upload new File
<form method=post enctype=multipart/form-data>
<input type=file name=file>
<input type=submit value=Upload>
</form>
'''
关键点：
- 必须设置
enctype="multipart/form-data" - 使用
secure_filename防止路径遍历attack - 检查
file.filename是否为空
3. 文件上传进阶
3.1 限制文件类型
ALLOWED_EXTENSIONS = {'txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif'}
def allowed_file(filename):
return '.' in filename and \
filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
@app.route('/upload', methods=['POST'])
def upload_file():
if 'file' not in request.files:
return redirect(request.url)
file = request.files['file']
if file.filename == '':
return redirect(request.url)
if file and allowed_file(file.filename):
filename = secure_filename(file.filename)
file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
return 'File uploaded successfully'
return 'Invalid file type'
3.2 限制文件大小
from flask import Flask, Request
from werkzeug.exceptions import RequestEntityTooLarge
app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 # 16MB
@app.errorhandler(RequestEntityTooLarge)
def handle_file_too_large(e):
return 'File exceeds maximum size limit (16MB)', 413
4. 文件下载
4.1 从服务器下载文件
from flask import send_from_directory
@app.route('/downloads/<filename>')
def download_file(filename):
return send_from_directory(
app.config['UPLOAD_FOLDER'],
filename,
as_attachment=True
)
4.2 动态生成文件下载
import io
from flask import Response
@app.route('/generate-csv')
def generate_csv():
# 创建内存中的CSV文件
data = io.StringIO()
data.write('id,name,email\n')
data.write('1,John,john@example.com\n')
data.write('2,Jane,jane@example.com\n')
# 返回文件下载
return Response(
data.getvalue(),
mimetype='text/csv',
headers={'Content-disposition': 'attachment; filename=users.csv'}
)
5. 安全防护
5.1 文件名安全处理
from werkzeug.utils import secure_filename
@app.route('/upload', methods=['POST'])
def upload_file():
file = request.files['file']
filename = secure_filename(file.filename)
# 防止目录遍历attack
filename = os.path.basename(filename)
file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
5.2 文件内容验证
import magic # python-magic库
def is_valid_file(file_stream):
# 检查文件实际类型
file_type = magic.from_buffer(file_stream.read(1024), mime=True)
file_stream.seek(0) # 重置指针
return file_type in ['image/jpeg', 'image/png']
@app.route('/upload-image', methods=['POST'])
def upload_image():
file = request.files['file']
if not is_valid_file(file.stream):
return 'Invalid image file', 400
# 处理文件...
6. 云存储集成
6.1 AWS S3上传示例
import boto3
from flask import current_app
s3 = boto3.client('s3',
aws_access_key_id=current_app.config['AWS_ACCESS_KEY'],
aws_secret_access_key=current_app.config['AWS_SECRET_KEY'])
@app.route('/upload-to-s3', methods=['POST'])
def upload_to_s3():
file = request.files['file']
s3.upload_fileobj(
file,
current_app.config['S3_BUCKET'],
secure_filename(file.filename)
)
return 'File uploaded to S3'
6.2 从S3下载
@app.route('/download-from-s3/<filename>')
def download_from_s3(filename):
s3_object = s3.get_object(
Bucket=current_app.config['S3_BUCKET'],
Key=filename
)
return Response(
s3_object['Body'].read(),
mimetype=s3_object['ContentType'],
headers={'Content-Disposition': f'attachment; filename={filename}'}
)
7. 性能优化
7.1 分块上传
@app.route('/chunk-upload', methods=['POST'])
def chunk_upload():
chunk = request.files['chunk']
chunk_number = request.form['chunkNumber']
total_chunks = request.form['totalChunks']
filename = secure_filename(request.form['filename'])
# 保存分块到临时目录
temp_dir = os.path.join(app.config['UPLOAD_FOLDER'], 'temp', filename)
os.makedirs(temp_dir, exist_ok=True)
chunk.save(os.path.join(temp_dir, chunk_number))
# 检查是否所有分块都已上传
if len(os.listdir(temp_dir)) == int(total_chunks):
# 合并文件
with open(os.path.join(app.config['UPLOAD_FOLDER'], filename), 'wb') as f:
for i in range(1, int(total_chunks) + 1):
with open(os.path.join(temp_dir, str(i)), 'rb') as chunk_file:
f.write(chunk_file.read())
# 清理临时文件
shutil.rmtree(temp_dir)
return 'File uploaded successfully'
return 'Chunk uploaded'
7.2 流式上传
@app.route('/stream-upload', methods=['POST'])
def stream_upload():
def custom_stream_factory(total_content_length, filename, content_type, content_length=None):
return open(os.path.join(app.config['UPLOAD_FOLDER'], secure_filename(filename)), 'wb')
return '', 204
8. 前端集成
8.1 AJAX文件上传
// 前端JavaScript
document.getElementById('upload-form').addEventListener('submit', function(e) {
e.preventDefault();
let formData = new FormData();
formData.append('file', document.getElementById('file-input').files[0]);
fetch('/upload', {
method: 'POST',
body: formData
})
.then(response => response.text())
.then(data => console.log(data))
.catch(error => console.error('Error:', error));
});
8.2 进度条显示
// 前端带进度条的AJAX上传
function uploadWithProgress(file) {
let xhr = new XMLHttpRequest();
xhr.open('POST', '/upload', true);
xhr.upload.onprogress = function(e) {
if (e.lengthComputable) {
let percent = Math.round((e.loaded / e.total) * 100);
console.log(percent + '% uploaded');
}
};
xhr.onload = function() {
if (xhr.status === 200) {
console.log('Upload complete');
}
};
let formData = new FormData();
formData.append('file', file);
xhr.send(formData);
}
9. 测试与调试
9.1 测试文件上传
import pytest
import io
def test_file_upload(client):
data = {
'file': (io.BytesIO(b"test file content"), 'test.txt')
}
response = client.post(
'/upload',
data=data,
content_type='multipart/form-data'
)
assert response.status_code == 200
assert b'File uploaded successfully' in response.data
9.2 测试文件下载
def test_file_download(client):
# 先上传测试文件
test_file_path = os.path.join(app.config['UPLOAD_FOLDER'], 'test.txt')
with open(test_file_path, 'w') as f:
f.write("test content")
# 测试下载
response = client.get('/downloads/test.txt')
assert response.status_code == 200
assert response.headers['Content-Disposition'] == 'attachment; filename=test.txt'
assert response.data == b"test content"
10. 总结与最佳实践
10.1 文件上传关键点
-
安全防护：
- 始终使用
secure_filename - 验证文件内容和类型
- 限制文件大小和扩展名
- 始终使用
-
目录管理：
- 为上传文件创建专用目录
- 定期清理旧文件
- 考虑按日期/用户组织文件结构
-
性能考虑：
- 对大文件实现分块上传
- 考虑使用云存储服务
- 实现进度反馈机制
10.2 文件下载关键点
-
响应控制：
- 使用正确的MIME类型
- 设置合适的
Content-Disposition - 实现范围请求支持(大文件)
-
安全考虑：
- 验证下载权限
- 防止目录遍历attack
- 考虑生成临时下载链接
-
性能优化：
- 实现文件压缩
- 使用X-Sendfile(如Nginx)
- 考虑CDN分发
10.3 生产环境建议
-
存储方案：
- 小文件：本地存储或数据库
- 大文件：云存储(S3等)
- 敏感文件：加密存储
-
监控与维护：
- 记录文件操作日志
- 监控存储空间使用
- 定期备份重要文件
-
扩展功能：
- 实现文件预览功能
- 添加virus扫描功能
- 支持文件版本控制
通过合理运用这些技术和最佳实践，您可以构建出安全、高效且易于维护的文件上传下载功能，满足各种业务场景需求。
