Flask请求响应与表单处理

Web开发专题 · 掌握Flask的表单与请求响应处理

专题: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-urlencodedmultipart/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 请求处理建议

6.2 响应构建建议

6.3 表单处理建议

扩展阅读:除Flask-WTF外,还可以使用 Flask-RESTful 构建RESTful API(使用 reqparse 解析请求),或 Marshmallow 进行更复杂的序列化/反序列化。对于纯API项目,推荐 Flask + Marshmallow + Webargs 的组合,更轻量且灵活。