← 返回Web开发目录
← 返回学习笔记首页
专题: Python Web开发系统学习
关键词: Python, Web开发, Flask表单, WTForms, 请求处理, 文件上传, CSRF, Flask-WTF, 验证器
一、请求处理详解
Flask将客户端发送的HTTP请求封装为全局的 request 对象,通过该对象可以访问所有请求数据。需要先 from flask import request 导入。
1.1 获取查询参数 (request.args)
查询参数指URL中 ? 后面的键值对,适用于GET请求的过滤、分页等场景。request.args 是一个 ImmutableMultiDict 对象,使用 get() 方法获取值,不存在时返回 None 或默认值。
# URL: /search?q=flask&page=1&page=2
from flask import request
q = request.args.get('q') # 'flask'
page = request.args.get('page') # '1'(注意:始终返回字符串)
page_int = request.args.get('page', type=int) # 1(自动类型转换)
all_pages = request.args.getlist('page') # ['1', '2']
keys = request.args.keys() # 所有参数名列表
has_q = 'q' in request.args # True(判断是否存在)
1.2 获取表单数据 (request.form)
HTTP POST请求提交的 application/x-www-form-urlencoded 或 multipart/form-data 格式数据,通过 request.form 访问。用法与 request.args 完全相同。
# 表单: <input name="username" value="admin">
# <input name="tags" value="python">
# <input name="tags" value="flask">
username = request.form.get('username') # 'admin'
tags = request.form.getlist('tags') # ['python', 'flask']
remember = request.form.get('remember') # 'on'(复选框选中时)
remember_bool = request.form.get('remember') == 'on' # True/False
1.3 获取JSON数据
当客户端发送 Content-Type: application/json 请求时,使用 request.get_json() 解析JSON主体。强烈建议使用 get_json() 而非 request.json,前者在解析失败时返回 None 而非抛出异常。
# 请求体: {"name": "张三", "age": 28}
data = request.get_json()
if data is None:
return {"error": "无效的JSON数据"}, 400
name = data.get('name')
age = data.get('age')
# 带参数解析
data = request.get_json(force=True) # 忽略Content-Type强制解析
data = request.get_json(silent=True) # 解析失败返回None不报错
1.4 获取上传文件 (request.files)
上传的文件通过 request.files 获取,每个文件是一个 FileStorage 对象,包含文件名、保存方法等。
from werkzeug.utils import secure_filename
# 表单: <input type="file" name="photo">
file = request.files.get('photo')
if file and file.filename:
filename = secure_filename(file.filename) # 清理文件名
file.save(os.path.join('uploads', filename)) # 保存到指定路径
1.5 请求头访问 (request.headers)
HTTP请求头可以通过 request.headers 访问,它是一个 EnvironHeaders 对象,支持大小写不敏感的键名访问。
# 常见请求头获取
user_agent = request.headers.get('User-Agent')
content_type = request.headers.get('Content-Type')
referer = request.headers.get('Referer')
auth = request.headers.get('Authorization')
x_forwarded = request.headers.get('X-Forwarded-For')
# 遍历所有请求头
for key, value in request.headers:
print(f'{key}: {value}')
1.6 请求环境 (request.environ)
request.environ 返回WSGI环境的原始字典,包含所有底层环境变量。通常在需要访问底层服务器信息时使用。
# 常用环境变量
remote_addr = request.environ.get('REMOTE_ADDR') # 客户端IP
server_name = request.environ.get('SERVER_NAME') # 服务器名
server_port = request.environ.get('SERVER_PORT') # 服务器端口
wsgi_url_scheme = request.environ.get('wsgi.url_scheme') # http 或 https
# 简写方式
remote_addr = request.remote_addr # 更推荐的获取客户端IP方式
1.7 多语言支持 (request.accept_languages)
根据客户端的 Accept-Language 请求头实现内容协商,返回最佳语言选项。
from flask import request
best_lang = request.accept_languages.best_match(['zh-CN', 'en', 'ja'])
# 如果浏览器首选zh-CN,返回 'zh-CN'
# 也可以获取排序后的列表
all_langs = request.accept_languages
for lang in all_langs:
print(f'{lang[0]}: {lang[1]}') # 语言代码: 质量值
请求对象核心属性速查表:
request.args - GET查询参数 | request.form - POST表单数据 | request.json/get_json() - JSON数据 | request.files - 上传文件 | request.headers - 请求头 | request.method - HTTP方法 | request.url - 完整URL | request.cookies - Cookie数据
二、响应构建
Flask视图函数必须返回一个响应对象。Flask提供了多种构建响应方式,从简单的字符串到完整的 Response 对象。
2.1 返回字符串与状态码
最简单的响应方式是直接返回字符串,可附带状态码和响应头。Flask会自动将字符串包装为 Response 对象。
@app.route('/hello')
def hello():
return 'Hello, World!' # 状态码默认200
@app.route('/created')
def created():
return '资源创建成功', 201 # 返回字符串+状态码
@app.route('/custom-header')
def custom_header():
return '带自定义头的响应', 200, {'X-Custom': 'value'}
2.2 render_template() 模板响应
使用 render_template() 渲染Jinja2模板文件,返回HTML页面。模板文件默认存放在 templates/ 目录下。
from flask import render_template
@app.route('/profile/')
def profile(username):
return render_template('profile.html',
username=username,
age=25,
interests=['编程', '阅读', '跑步'])
2.3 jsonify() JSON响应
构建RESTful API时使用 jsonify() 返回JSON格式响应。它会自动设置 Content-Type: application/json。
from flask import jsonify
@app.route('/api/user/')
def get_user(uid):
user = {'id': uid, 'name': '张三', 'email': 'zhangsan@example.com'}
return jsonify(code=200, message='成功', data=user)
# 输出: {"code": 200, "message": "成功", "data": {...}}
# 也可以直接返回字典(Flask 2.0+)
@app.route('/api/status')
def status():
return {'status': 'ok', 'version': '1.0.0'}
2.4 redirect() 重定向
页面跳转使用 redirect(),常与 url_for() 配合使用,避免硬编码URL。
from flask import redirect, url_for
@app.route('/')
def index():
return redirect(url_for('login')) # 重定向到login视图
@app.route('/external')
def external():
return redirect('https://www.example.com') # 外部重定向
# 302临时重定向(默认)
# 301永久重定向
@app.route('/old-page')
def old_page():
return redirect(url_for('new_page'), code=301)
# 或使用: return redirect(url_for('new_page'), 301)
2.5 Response对象构建
直接构造 Response 对象可以获得最大灵活性,适用于设置Cookie、自定义状态码等高级场景。
from flask import Response, make_response
# 方式一: 直接构造
resp = Response('响应内容', status=200, mimetype='text/html')
# 方式二: make_response 包装
@app.route('/set-cookie')
def set_cookie():
resp = make_response('Cookie已设置')
resp.set_cookie('username', 'admin', max_age=3600) # 1小时过期
resp.set_cookie('token', 'abc123', httponly=True) # 仅HTTP访问
resp.delete_cookie('old_cookie') # 删除Cookie
return resp
2.6 设置响应头
设置自定义响应头有多种方式,按需选择即可。
# 方式一: 返回元组
@app.route('/headers1')
def headers1():
return '内容', 200, {'X-Custom': 'value', 'Cache-Control': 'no-cache'}
# 方式二: Response对象
@app.route('/headers2')
def headers2():
resp = make_response('内容')
resp.headers['X-Custom'] = 'value'
resp.headers['Cache-Control'] = 'public, max-age=3600'
resp.headers.add('Vary', 'Accept-Encoding') # 添加而非覆盖
return resp
2.7 文件下载 (send_file / send_from_directory)
Flask提供两个文件下载函数:send_file 发送单个文件,send_from_directory 从指定目录安全发送文件。
from flask import send_file, send_from_directory
# 发送内存中生成的文件
import io
@app.route('/download/report')
def download_report():
buf = io.BytesIO()
buf.write(b'报告内容...')
buf.seek(0)
return send_file(buf, as_attachment=True,
download_name='report.txt',
mimetype='text/plain')
# 发送服务器上的文件
@app.route('/uploads/')
def uploaded_file(filename):
return send_from_directory('uploads', filename)
# send_file支持更多选项
@app.route('/download/pdf')
def download_pdf():
return send_file('static/doc.pdf',
mimetype='application/pdf',
as_attachment=True, # 作为附件下载
download_name='文档.pdf') # 指定下载文件名
2.8 流式响应 (Response + generator)
对于大文件或实时数据,使用生成器实现流式响应可以显著降低内存占用。
from flask import Response, stream_with_context
# 大文件逐块读取
def generate_large_file():
with open('large_file.csv', 'r') as f:
while True:
chunk = f.read(8192) # 每次读取8KB
if not chunk:
break
yield chunk
@app.route('/stream/csv')
def stream_csv():
return Response(generate_large_file(),
mimetype='text/csv',
headers={'Content-Disposition': 'attachment;filename=data.csv'})
# 实时日志流(带上下文)
@app.route('/stream/logs')
def stream_logs():
def generate():
yield '开始监控...\n'
# 模拟实时日志
yield '2026-05-05 10:00:01 请求到达\n'
yield '2026-05-05 10:00:02 处理中...\n'
yield '2026-05-05 10:00:03 处理完成\n'
return Response(stream_with_context(generate()),
mimetype='text/plain')
核心原则: Flask视图函数可以返回 (str,)、(str, status)、(str, status, headers)、Response对象、dict(自动jsonify)或 render_template() 结果。所有形式最终都会被Flask转换为 Response 对象。
三、Flash消息
Flash消息系统用于在请求之间传递一次性提示消息,常用于表单提交后显示成功或错误信息。Flash消息存储在Session中,获取后自动清除。
3.1 基本使用
首先配置 SECRET_KEY,因为Flash消息依赖Session(Session又依赖Cookie签名)。
from flask import Flask, flash, get_flashed_messages, render_template
app = Flask(__name__)
app.secret_key = 'your-secret-key-here'
@app.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'POST':
username = request.form.get('username')
password = request.form.get('password')
if username == 'admin' and password == '123456':
flash('登录成功!欢迎回来!') # 添加成功消息
return redirect(url_for('dashboard'))
else:
flash('用户名或密码错误', 'error') # 带分类的Flash消息
return render_template('login.html')
3.2 获取Flash消息
在模板或视图函数中使用 get_flashed_messages() 获取所有待显示的消息。
# 视图函数中获取(较少用)
messages = get_flashed_messages()
for message in messages:
print(message)
# 带分类获取
messages = get_flashed_messages(with_categories=True)
for category, message in messages:
print(f'[{category}] {message}')
3.3 Flash消息分类
使用分类可以区分不同级别的消息,在模板中分别渲染不同样式。
# 添加不同分类的消息
flash('操作成功完成!', 'success')
flash('请检查输入信息', 'warning')
flash('系统出现错误', 'error')
flash('这是一条提示', 'info')
# 仅获取特定分类的消息
errors = get_flashed_messages(category_filter=['error'])
infos = get_flashed_messages(category_filter=['info', 'success'])
3.4 模板中显示Flash消息
在Jinja2模板中遍历 get_flashed_messages() 来渲染消息。
<!-- templates/base.html -->
{% with messages = get_flashed_messages(with_categories=true) %}
{% if messages %}
<div class="flash-messages">
{% for category, message in messages %}
<div class="flash flash-{{ category }}">
{{ message }}
</div>
{% endfor %}
</div>
{% endif %}
{% endwith %}
<!-- CSS 配合不同分类样式 -->
<style>
.flash { padding: 12px 18px; margin: 10px 0; border-radius: 4px; }
.flash-success { background: #d4edda; color: #155724; border: 1px solid #c3e6cb; }
.flash-error { background: #f8d7da; color: #721c24; border: 1px solid #f5c6cb; }
.flash-warning { background: #fff3cd; color: #856404; border: 1px solid #ffeeba; }
.flash-info { background: #d1ecf1; color: #0c5460; border: 1px solid #bee5eb; }
</style>
注意: Flash消息依赖Session,必须在应用上设置 SECRET_KEY。Flash消息是一次性的——获取后即被清除,不会在页面刷新后重复显示。
四、WTForms表单处理
WTForms是Flask最常用的表单处理库,通过 Flask-WTF 扩展集成,提供表单定义、验证、渲染和CSRF保护的一站式解决方案。
4.1 安装与配置
# 安装Flask-WTF(会自动安装WTForms)
pip install flask-wtf
# 应用配置
import os
from flask import Flask
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField
from wtforms.validators import DataRequired, Email, Length
app = Flask(__name__)
app.config['SECRET_KEY'] = os.urandom(24).hex() # 生成随机密钥
# app.config['WTF_CSRF_TIME_LIMIT'] = 3600 # CSRF令牌过期时间(秒)
4.2 表单类定义
继承 FlaskForm 定义表单类,字段名与HTML name 属性对应。
from flask_wtf import FlaskForm
from wtforms import (StringField, PasswordField, IntegerField,
SelectField, BooleanField, FileField, SubmitField,
TextAreaField, DateField)
from wtforms.validators import (DataRequired, Email, Length,
NumberRange, EqualTo, Regexp, URL)
class RegisterForm(FlaskForm):
username = StringField('用户名', validators=[
DataRequired('请输入用户名'),
Length(min=3, max=20, message='用户名长度3-20个字符')
])
email = StringField('邮箱', validators=[
DataRequired('请输入邮箱'),
Email('邮箱格式不正确')
])
password = PasswordField('密码', validators=[
DataRequired('请输入密码'),
Length(min=6, max=128, message='密码至少6位')
])
confirm_password = PasswordField('确认密码', validators=[
DataRequired('请确认密码'),
EqualTo('password', message='两次密码不一致')
])
age = IntegerField('年龄', validators=[
NumberRange(min=1, max=150, message='年龄范围1-150')
])
gender = SelectField('性别', choices=[
('', '请选择'), ('male', '男'), ('female', '女')
], validators=[DataRequired('请选择性别')])
agree = BooleanField('同意条款', validators=[
DataRequired('请同意条款')
])
avatar = FileField('头像')
submit = SubmitField('注册')
4.3 字段类型汇总
字段类型 说明 HTML渲染
StringField 单行文本输入 <input type="text">
PasswordField 密码输入 <input type="password">
IntegerField 整数输入 <input type="number">
FloatField 浮点数输入 <input type="number" step="any">
BooleanField 复选框 <input type="checkbox">
SelectField 下拉选择 <select><option>
SelectMultipleField 多选下拉 <select multiple>
TextAreaField 多行文本 <textarea>
FileField 文件上传 <input type="file">
SubmitField 提交按钮 <input type="submit">
HiddenField 隐藏字段 <input type="hidden">
DateField 日期选择 <input type="date">
DateTimeField 日期时间 <input type="datetime-local">
4.4 验证器详解
WTForms内置了丰富的验证器,通过 validators 列表参数添加。验证按列表顺序依次执行。
from wtforms.validators import (
DataRequired, # 字段不能为空
Email, # 邮箱格式验证
Length, # 字符串长度范围验证
NumberRange, # 数字范围验证
EqualTo, # 两字段值相等验证(确认密码)
Regexp, # 正则表达式验证
URL, # URL格式验证
AnyOf, # 值必须在指定列表中
NoneOf, # 值不能在指定列表中
InputRequired, # 与DataRequired类似,但允许空字符串
Optional, # 字段可选,为空时跳过其他验证
)
# 综合示例
class ProfileForm(FlaskForm):
phone = StringField('手机号', validators=[
DataRequired('请输入手机号'),
Regexp(r'^1[3-9]\d{9}$', message='手机号格式不正确')
])
website = StringField('个人网站', validators=[
Optional(), # 非必填
URL('请输入有效URL')
])
role = SelectField('角色', choices=[
('admin', '管理员'), ('user', '普通用户')
], validators=[AnyOf(['admin', 'user'], message='无效角色')])
4.5 自定义验证器
当内置验证器无法满足需求时,可编写自定义验证器。有两种方式:行内函数或可复用类。
from wtforms.validators import ValidationError
# 方式一: 行内自定义验证(方法名 validate_字段名)
class RegisterForm(FlaskForm):
username = StringField('用户名', validators=[DataRequired()])
def validate_username(self, field):
"""检查用户名是否已存在"""
if field.data == 'admin':
raise ValidationError('用户名 admin 已被注册')
if not field.data.isalnum():
raise ValidationError('用户名只能包含字母和数字')
# 方式二: 可复用验证器类
class UniqueUsername:
def __init__(self, message=None):
self.message = message or '用户名已存在'
def __call__(self, form, field):
# 查询数据库逻辑(示例)
existing_users = ['admin', 'root', 'test']
if field.data.lower() in existing_users:
raise ValidationError(self.message)
class RegisterForm(FlaskForm):
username = StringField('用户名', validators=[
DataRequired(), UniqueUsername()
])
4.6 模板中渲染表单
在Jinja2模板中使用WTForms渲染表单,自动生成HTML并包含CSRF令牌。
<!-- templates/register.html -->
<form method="POST" enctype="multipart/form-data">
{{ form.hidden_tag() }} <!-- 自动渲染CSRF令牌 + 所有Hidden字段 -->
<div class="form-group">
{{ form.username.label }}
{{ form.username(class="form-control", placeholder="请输入用户名") }}
{% for error in form.username.errors %}
<span class="error">{{ error }}</span>
{% endfor %}
</div>
<div class="form-group">
{{ form.email.label }}
{{ form.email(size=32) }}
{% if form.email.errors %}
<ul class="errors">
{% for error in form.email.errors %}
<li>{{ error }}</li>
{% endfor %}
</ul>
{% endif %}
</div>
<div class="form-group">
{{ form.gender.label }}
{{ form.gender(class="form-select") }}
</div>
<div class="form-group">
{{ form.agree() }} {{ form.agree.label }}
</div>
{{ form.submit(class="btn btn-primary") }}
</form>
4.7 视图函数处理表单
在视图函数中实例化表单并验证。
@app.route('/register', methods=['GET', 'POST'])
def register():
form = RegisterForm()
if form.validate_on_submit(): # 验证提交(POST + 验证通过)
# 获取验证后的数据
username = form.username.data
email = form.email.data
password = form.password.data
# 此处保存到数据库...
# db.session.add(User(username=username, email=email, password=password))
# db.session.commit()
flash(f'用户 {username} 注册成功!', 'success')
return redirect(url_for('login'))
# GET请求或验证失败,渲染表单页面
return render_template('register.html', form=form)
4.8 CSRF保护
Flask-WTF默认启用CSRF保护,且是WTForms的核心优势之一。需要正确配置才能生效。
# CSRF配置选项
app.config['SECRET_KEY'] = '难以猜测的密钥字符串' # 必需
app.config['WTF_CSRF_ENABLED'] = True # 默认True
app.config['WTF_CSRF_TIME_LIMIT'] = 3600 # 令牌有效期
app.config['WTF_CSRF_SSL_STRICT'] = True # HTTPS下严格检查
# AJAX请求传递CSRF令牌
# 在meta标签中设置
# <meta name="csrf-token" content="{{ csrf_token() }}">
#
# AJAX请求头
# const csrfToken = document.querySelector('meta[name="csrf-token"]').content;
# fetch('/api/data', {
# method: 'POST',
# headers: {'X-CSRFToken': csrfToken},
# body: JSON.stringify(data)
# })
# 禁用特定视图的CSRF(谨慎使用)
@csrf.exempt
@app.route('/webhook', methods=['POST'])
def webhook():
return 'ok'
最佳实践: 优先使用 form.validate_on_submit() 组合检查,而非手动判断 request.method == 'POST' 后再验证。前者同时检查请求方法和验证结果,代码更简洁。
五、文件上传处理
文件上传是Web开发中的常见需求,Flask通过 request.files 处理上传,结合WTForms的 FileField 和Werkzeug的 secure_filename 确保安全。
5.1 完整上传流程
import os
from werkzeug.utils import secure_filename
# 配置上传目录和允许类型
app.config['UPLOAD_FOLDER'] = 'uploads' # 上传保存目录
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 # 限制16MB
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'pdf'}
def allowed_file(filename):
"""检查文件扩展名是否允许"""
return '.' in filename and \
filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
@app.route('/upload', methods=['GET', 'POST'])
def upload_file():
if request.method == 'POST':
# 检查是否有文件
if 'file' not in request.files:
flash('没有选择文件', 'error')
return redirect(request.url)
file = request.files['file']
if file.filename == '':
flash('文件名为空', 'error')
return redirect(request.url)
if file and allowed_file(file.filename):
filename = secure_filename(file.filename) # 清理文件名
# 防止文件名冲突
name, ext = os.path.splitext(filename)
import uuid
filename = f'{name}_{uuid.uuid4().hex[:8]}{ext}'
file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
flash(f'文件 {filename} 上传成功!', 'success')
return redirect(url_for('upload_file'))
flash('不允许的文件类型', 'error')
return render_template('upload.html')
5.2 使用WTForms处理上传
from flask_wtf import FlaskForm
from flask_wtf.file import FileField, FileRequired, FileAllowed
class UploadForm(FlaskForm):
photo = FileField('选择照片', validators=[
FileRequired('请选择文件'),
FileAllowed(['jpg', 'png', 'gif', 'jpeg'], '仅支持图片文件!')
])
submit = SubmitField('上传')
@app.route('/upload-wtf', methods=['GET', 'POST'])
def upload_wtf():
form = UploadForm()
if form.validate_on_submit():
f = form.photo.data # FileStorage对象
filename = secure_filename(f.filename)
f.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
flash('上传成功!', 'success')
return redirect(url_for('upload_wtf'))
return render_template('upload_wtf.html', form=form)
5.3 上传模板
<!-- templates/upload.html -->
<form method="POST" enctype="multipart/form-data">
{{ form.hidden_tag() }}
<p>
{{ form.photo.label }}<br>
{{ form.photo() }}
{% for error in form.photo.errors %}
<span style="color:red">{{ error }}</span>
{% endfor %}
</p>
<p>{{ form.submit() }}</p>
</form>
5.4 上传大小限制
Flask通过 MAX_CONTENT_LENGTH 配置限制上传大小。超过限制时,Flask会抛出 413 Request Entity Too Large 异常。
# 全局限制
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 # 16 MB
# 自定义413错误页面
@app.errorhandler(413)
def too_large(e):
flash('文件过大,最大允许16MB', 'error')
return redirect(request.url), 413
# 限制特定路由(通过before_request装饰器实现)
@app.before_request
def limit_upload_size():
if request.path == '/upload' and request.content_length:
if request.content_length > 5 * 1024 * 1024: # 5MB
abort(413)
安全要点:
始终使用 secure_filename() 清理文件名,防止路径穿越攻击(如 ../../etc/passwd)
始终检查文件扩展名,限制允许上传的类型
始终设置 MAX_CONTENT_LENGTH,防止大文件耗尽服务器资源
不要信任用户提供的 Content-Type,仅依赖扩展名验证
上传文件保存在Web根目录之外(或使用专用子目录),避免直接通过URL访问
六、常用最佳实践总结
6.1 请求处理建议
使用 request.get_json(silent=True) 而非 request.json,避免解析异常导致500错误
获取数值参数时使用 type=int 参数,如 request.args.get('page', type=int)
对于可选查询参数,始终提供默认值,如 request.args.get('q', '')
使用 getlist() 获取多值参数(如复选框组、多选下拉)
6.2 响应构建建议
JSON API优先使用 jsonify() 或直接返回字典(Flask 2.0+)
重定向时使用 url_for() 而非硬编码URL,便于后期修改
大文件下载使用流式响应(generator)减少内存占用
统一错误响应格式,便于前端处理
6.3 表单处理建议
始终启用CSRF保护(Flask-WTF默认启用,不要关闭)
将表单类单独放在 forms.py 模块管理,保持代码组织清晰
使用 form.validate_on_submit() 替代手动判断 request.method == 'POST'
显示验证错误时,优先使用宏(macro)封装表单字段渲染,减少模板重复代码
考虑使用 Flask-WTF 的 FileAllowed 验证器替代手动扩展名检查
扩展阅读: 除Flask-WTF外,还可以使用 Flask-RESTful 构建RESTful API(使用 reqparse 解析请求),或 Marshmallow 进行更复杂的序列化/反序列化。对于纯API项目,推荐 Flask + Marshmallow + Webargs 的组合,更轻量且灵活。