05-表单处理

Python 3.11+

本章讲解 Flask-WTF 表单创建、验证和处理。

第一部分：Flask-WTF 基础

1.1 实际场景

你正在开发用户注册功能，需要收集用户名、邮箱、密码等信息，并进行验证（邮箱格式、密码长度等）。

问题：如何创建并验证 Web 表单？

1.2 安装

bash

pip install flask-wtf email-validator

1.3 基本配置

python

from flask import Flask
from flask_wtf import FlaskForm

app: Flask = Flask(__name__)
app.config["SECRET_KEY"] = "your-secret-key"

# CSRF 配置
app.config["WTF_CSRF_ENABLED"] = True
app.config["WTF_CSRF_TIME_LIMIT"] = 3600

第二部分：创建表单

2.1 实际场景

注册表单需要用户名、邮箱、密码、确认密码字段，每个字段有不同的验证规则。

问题：如何定义表单类和验证规则？

2.2 基础表单

python

from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField
from wtforms.validators import DataRequired, Email, Length, EqualTo

class LoginForm(FlaskForm):
    username: StringField = StringField("用户名", validators=[DataRequired()])
    password: PasswordField = PasswordField("密码", validators=[DataRequired()])
    submit: SubmitField = SubmitField("登录")

2.3 完整注册表单

python

from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, TextAreaField, IntegerField, BooleanField
from wtforms.validators import (
    DataRequired, Email, Length, EqualTo,
    Optional, NumberRange
)

class RegisterForm(FlaskForm):
    # 文本字段
    username: StringField = StringField(
        "用户名",
        validators=[
            DataRequired(message="用户名不能为空"),
            Length(min=3, max=20, message="用户名长度3-20字符")
        ]
    )
    
    email: StringField = StringField(
        "邮箱",
        validators=[
            DataRequired(message="邮箱不能为空"),
            Email(message="请输入有效的邮箱地址")
        ]
    )
    
    password: PasswordField = PasswordField(
        "密码",
        validators=[
            DataRequired(message="密码不能为空"),
            Length(min=6, message="密码至少6位")
        ]
    )
    
    confirm_password: PasswordField = PasswordField(
        "确认密码",
        validators=[
            DataRequired(message="请确认密码"),
            EqualTo("password", message="两次密码不一致")
        ]
    )
    
    bio: TextAreaField = TextAreaField(
        "个人简介",
        validators=[Optional(), Length(max=500)]
    )
    
    age: IntegerField = IntegerField(
        "年龄",
        validators=[Optional(), NumberRange(min=0, max=150)]
    )
    
    agree_terms: BooleanField = BooleanField(
        "同意服务条款",
        validators=[DataRequired(message="必须同意服务条款")]
    )
    
    submit: SubmitField = SubmitField("注册")

第三部分：模板中使用表单

3.1 实际场景

表单定义好了，需要在 HTML 页面中渲染出来，并显示验证错误信息。

问题：如何在模板中渲染表单？

3.2 渲染表单

html

<form method="POST" action="{{ url_for('register') }}">
    {{ form.hidden_tag() }}  <!-- CSRF token -->
    
    <div>
        {{ form.username.label }}
        {{ form.username(size=20) }}
        {% if form.username.errors %}
            <ul class="errors">
            {% for error in form.username.errors %}
                <li>{{ error }}</li>
            {% endfor %}
            </ul>
        {% endif %}
    </div>
    
    <div>
        {{ form.email.label }}
        {{ form.email() }}
        {% for error in form.email.errors %}
            <span class="error">{{ error }}</span>
        {% endfor %}
    </div>
    
    <div>
        {{ form.password.label }}
        {{ form.password() }}
    </div>
    
    <div>
        {{ form.confirm_password.label }}
        {{ form.confirm_password() }}
    </div>
    
    {{ form.submit() }}
</form>

第四部分：处理表单数据

4.1 实际场景

用户提交注册表单，你需要验证表单数据，如果验证成功则创建用户。

问题：如何在视图函数中处理表单提交？

4.2 视图函数

python

from flask import Flask, render_template, redirect, url_for, flash, request

app: Flask = Flask(__name__)


@app.route("/register", methods=["GET", "POST"])
def register() -> str:
    form: RegisterForm = RegisterForm()
    
    if request.method == "POST" and form.validate_on_submit():
        # 获取表单数据
        username: str = form.username.data
        email: str = form.email.data
        password: str = form.password.data
        
        # 处理数据...
        
        flash("注册成功！", "success")
        return redirect(url_for("login"))
    
    return render_template("register.html", form=form)

4.3 自定义验证

python

from wtforms.validators import ValidationError
from flask_sqlalchemy import SQLAlchemy

db: SQLAlchemy = SQLAlchemy()


class RegisterForm(FlaskForm):
    username: StringField = StringField("用户名")
    
    # 自定义验证器
    def validate_username(self, username: StringField) -> None:
        user: User | None = User.query.filter_by(username=username.data).first()
        if user:
            raise ValidationError("用户名已存在")
    
    def validate_email(self, email: StringField) -> None:
        user: User | None = User.query.filter_by(email=email.data).first()
        if user:
            raise ValidationError("邮箱已被注册")

第五部分：文件上传表单

5.1 实际场景

用户需要上传头像图片，只允许 jpg、png、gif 格式。

问题：如何创建文件上传表单并限制文件类型？

5.2 文件上传表单

python

from flask_wtf import FlaskForm
from flask_wtf.file import FileField, FileAllowed, FileRequired
from wtforms import SubmitField

class UploadForm(FlaskForm):
    avatar: FileField = FileField(
        "头像",
        validators=[
            FileRequired(message="请选择文件"),
            FileAllowed(["jpg", "png", "gif"], message="只支持 jpg, png, gif 格式")
        ]
    )
    
    document: FileField = FileField(
        "文档",
        validators=[
            FileAllowed(["pdf", "doc", "docx"], message="只支持 PDF 和 Word 文档")
        ]
    )
    
    submit: SubmitField = SubmitField("上传")

第六部分：完整示例

6.1 表单类

python

# forms.py
from flask_wtf import FlaskForm
from wtforms import (
    StringField, TextAreaField, SelectField, BooleanField, SubmitField
)
from wtforms.validators import DataRequired, Email, Length

class ContactForm(FlaskForm):
    name: StringField = StringField(
        "姓名",
        validators=[DataRequired(), Length(max=50)],
        render_kw={"placeholder": "请输入姓名"}
    )
    
    email: StringField = StringField(
        "邮箱",
        validators=[DataRequired(), Email()],
        render_kw={"placeholder": "example@mail.com"}
    )
    
    subject: SelectField = SelectField(
        "主题",
        choices=[
            ("general", "一般咨询"),
            ("bug", "问题反馈"),
            ("business", "商务合作"),
            ("other", "其他")
        ],
        validators=[DataRequired()]
    )
    
    message: TextAreaField = TextAreaField(
        "留言内容",
        validators=[DataRequired(), Length(min=10, max=1000)],
        render_kw={"rows": 5, "placeholder": "请输入留言内容"}
    )
    
    subscribe: BooleanField = BooleanField("订阅Newsletter", default=True)
    
    submit: SubmitField = SubmitField("提交")

6.2 模板

html

<!-- templates/contact.html -->
<!DOCTYPE html>
<html>
<head>
    <title>联系我们</title>
    <style>
        .form-group { margin-bottom: 15px; }
        label { display: block; margin-bottom: 5px; font-weight: bold; }
        input, textarea, select { 
            width: 100%; 
            padding: 8px; 
            border: 1px solid #ddd; 
            border-radius: 4px;
        }
        .error { color: red; font-size: 0.9em; }
        .success { color: green; }
    </style>
</head>
<body>
    <h1>联系我们</h1>
    
    {% with messages = get_flashed_messages(with_categories=true) %}
        {% if messages %}
            {% for category, message in messages %}
                <div class="{{ category }}">{{ message }}</div>
            {% endfor %}
        {% endif %}
    {% endwith %}
    
    <form method="POST">
        {{ form.hidden_tag() }}
        
        <div class="form-group">
            {{ form.name.label }}
            {{ form.name() }}
            {% for error in form.name.errors %}
                <span class="error">{{ error }}</span>
            {% endfor %}
        </div>
        
        <div class="form-group">
            {{ form.email.label }}
            {{ form.email() }}
        </div>
        
        <div class="form-group">
            {{ form.subject.label }}
            {{ form.subject() }}
        </div>
        
        <div class="form-group">
            {{ form.message.label }}
            {{ form.message() }}
        </div>
        
        <div class="form-group">
            {{ form.subscribe() }} {{ form.subscribe.label }}
        </div>
        
        <div class="form-group">
            {{ form.submit() }}
        </div>
    </form>
</body>
</html>

6.3 视图函数

python

@app.route("/contact", methods=["GET", "POST"])
def contact() -> str:
    form: ContactForm = ContactForm()
    
    if form.validate_on_submit():
        # 处理表单数据
        name: str = form.name.data
        email: str = form.email.data
        subject: str = form.subject.data
        message: str = form.message.data
        
        # 保存到数据库或发送邮件
        save_message(name, email, subject, message)
        
        flash("感谢您的留言！", "success")
        return redirect(url_for("contact"))
    
    return render_template("contact.html", form=form)

第七部分：L3 专家层

7.1 CSRF Token 的生成与验证原理（HMAC）

CSRF（Cross-Site Request Forgery）保护的核心在于生成一个不可预测的 Token，并将其嵌入表单中。Flask-WTF 使用 HMAC（Hash-based Message Authentication Code）实现。

HMAC 签名流程：

+------------------+        +-------------------+        +------------------+
|   SECRET_KEY     |        |    csrf_token     |        |   HMAC-SHA256    |
|  (服务器持有)     |------->|  (随机生成)        |------->|  (摘要计算)       |
+------------------+        +-------------------+        +------------------+
                                                          |
                                                          v
                                                 +------------------+
                                                 |  signed_token =  |
                                                 |  token + '.' +   |
                                                 |  timestamp + '.' |
                                                 |  signature       |
                                                 +------------------+

python

import hmac
import hashlib
import secrets
import time
from typing import Tuple

def generate_csrf_token(secret_key: str, time_limit: int = 3600) -> str:
    """生成 CSRF Token（简化版实现）"""
    token: str = secrets.token_urlsafe(32)
    timestamp: int = int(time.time())
    message: bytes = f"{token}.{timestamp}".encode("utf-8")
    signature: str = hmac.new(
        secret_key.encode("utf-8"),
        message,
        hashlib.sha256
    ).hexdigest()
    return f"{token}.{timestamp}.{signature}"

def validate_csrf_token(secret_key: str, signed_token: str, time_limit: int = 3600) -> bool:
    """验证 CSRF Token"""
    try:
        token, timestamp_str, signature = signed_token.split(".")
        timestamp: int = int(timestamp_str)
        
        # 检查时效性
        if int(time.time()) - timestamp > time_limit:
            return False
        
        # 重新计算签名并比较
        message: bytes = f"{token}.{timestamp}".encode("utf-8")
        expected: str = hmac.new(
            secret_key.encode("utf-8"),
            message,
            hashlib.sha256
        ).hexdigest()
        
        return hmac.compare_digest(expected, signature)
    except ValueError:
        return False

关键设计：

secrets.token_urlsafe(32) 生成加密安全的随机值（256 bit 熵）
时间戳防止 Token 被无限期重用
hmac.compare_digest() 防止时序攻击（Timing Attack）

7.2 表单验证的装饰器模式

WTForms 的验证器链本质上是责任链模式（Chain of Responsibility），每个验证器独立执行，错误累积返回。

python

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Callable

@dataclass
class ValidationResult:
    """验证结果"""
    is_valid: bool = True
    errors: list[str] = field(default_factory=list)

class Validator(ABC):
    """验证器基类"""
    @abstractmethod
    def __call__(self, value: str) -> ValidationResult:
        ...

class DataRequired(Validator):
    def __init__(self, message: str = "字段不能为空") -> None:
        self.message: str = message
    
    def __call__(self, value: str) -> ValidationResult:
        if not value or not value.strip():
            return ValidationResult(is_valid=False, errors=[self.message])
        return ValidationResult()

class Length(Validator):
    def __init__(self, min_len: int = 0, max_len: int | None = None, message: str = "") -> None:
        self.min_len: int = min_len
        self.max_len: int | None = max_len
        self.message: str = message
    
    def __call__(self, value: str) -> ValidationResult:
        if len(value) < self.min_len:
            return ValidationResult(is_valid=False, errors=[self.message or f"长度不能少于{self.min_len}"])
        if self.max_len and len(value) > self.max_len:
            return ValidationResult(is_valid=False, errors=[self.message or f"长度不能超过{self.max_len}"])
        return ValidationResult()

def validate_field(validators: list[Validator], value: str) -> ValidationResult:
    """执行验证器链"""
    result: ValidationResult = ValidationResult()
    for validator in validators:
        v_result: ValidationResult = validator(value)
        if not v_result.is_valid:
            result.is_valid = False
            result.errors.extend(v_result.errors)
    return result

验证器执行流程：

  用户输入
     |
     v
+-----------+    失败    +-----------+    失败    +-----------+
|DataRequired|--------->|   Length   |--------->|   Email    |
|  (非空)    |          |  (长度)    |          |  (格式)    |
+-----------+          +-----------+          +-----------+
     |                      |                      |
     v 成功                 v 成功                 v 成功
+------------------------------------------------------+
|                 验证通过，返回数据                      |
+------------------------------------------------------+

7.3 文件上传的 multipart/form-data 解析过程

HTTP 文件上传使用 multipart/form-data 编码类型，浏览器将表单数据分割为多个"部分"（part），每部分有独立的 Header 和 Body。

请求体结构：

POST /upload HTTP/1.1
Content-Type: multipart/form-data; boundary=----WebKitFormBoundary7MA4YWxkTrZu0gW

------WebKitFormBoundary7MA4YWxkTrZu0gW
Content-Disposition: form-data; name="username"

baxiang
------WebKitFormBoundary7MA4YWxkTrZu0gW
Content-Disposition: form-data; name="avatar"; filename="photo.jpg"
Content-Type: image/jpeg

<二进制文件数据...>
------WebKitFormBoundary7MA4YWxkTrZu0gW--

Werkzeug 解析流程：

python

from io import BufferedReader
from typing import BinaryIO

def parse_multipart_boundary(raw_data: bytes, boundary: bytes) -> dict[str, str | BinaryIO]:
    """简化版 multipart 解析器"""
    parts: dict[str, str | BinaryIO] = {}
    boundary_line: bytes = b"--" + boundary
    
    # 按 boundary 分割
    segments: list[bytes] = raw_data.split(boundary_line)
    
    for segment in segments[1:-1]:  # 跳过首尾空段
        if segment.startswith(b"\r\n"):
            segment = segment[2:]
        
        # 分离 Header 和 Body
        header_end: int = segment.find(b"\r\n\r\n")
        headers: bytes = segment[:header_end]
        body: bytes = segment[header_end + 4:]
        
        # 解析 Content-Disposition
        disposition: str = headers.decode("utf-8").split("\r\n")[0]
        name_start: int = disposition.find('name="') + 6
        name_end: int = disposition.find('"', name_start)
        field_name: str = disposition[name_start:name_end]
        
        # 检查是否有 filename（文件字段）
        if "filename=" in disposition:
            parts[field_name] = body  # 实际实现会包装为 FileStorage
        else:
            parts[field_name] = body.decode("utf-8")
    
    return parts

性能考量：

方案	内存占用	适用场景	缺点
全量加载到内存	O(file_size)	小文件（<10MB）	大文件导致 OOM
流式解析（Werkzeug 默认）	O(buffer_size)	任意大小	实现复杂
临时文件落盘	O(disk_space)	超大文件（>100MB）	I/O 开销大

设计动机：

设计决策	原因
boundary 随机生成	防止与文件内容冲突（boundary 注入攻击）
流式解析默认启用	避免恶意大文件耗尽服务器内存
MAX_CONTENT_LENGTH 前置检查	在解析前拒绝超限请求，节省 CPU

7.4 知识关联

                    表单处理知识体系
                         |
        +----------------+----------------+
        |                |                |
     创建层          验证层          安全层
        |                |                |
   +----+----+      +----+----+      +----+----+
   |  字段    |      |  验证器  |      |  CSRF   |
   |  类型    |      |  链式    |      |  Token  |
   +----+----+      +----+----+      +----+----+
        |                |                |
        v                v                v
   +---------+      +---------+      +---------+
   | WTForms |      | 责任链  |      | HMAC    |
   | 类定义  |      | 模式    |      | 签名    |
   +---------+      +---------+      +---------+
                         |
                         v
                   +----+----+      +---------+
                   | 自定义  |----->| 数据库  |
                   | 验证器  |      | 查询    |
                   +---------+      +---------+

知识点	说明
Flask-WTF	表单处理扩展
字段类型	String, Password, TextArea 等
验证器	DataRequired, Email, Length 等
CSRF 保护	hidden_tag()
自定义验证	validate_字段名
文件上传	FileField, FileAllowed

05-表单处理 ​

第一部分：Flask-WTF 基础 ​

1.1 实际场景 ​

1.2 安装 ​

1.3 基本配置 ​

第二部分：创建表单 ​

2.1 实际场景 ​

2.2 基础表单 ​

2.3 完整注册表单 ​

第三部分：模板中使用表单 ​

3.1 实际场景 ​

3.2 渲染表单 ​

第四部分：处理表单数据 ​

4.1 实际场景 ​

4.2 视图函数 ​

4.3 自定义验证 ​

第五部分：文件上传表单 ​

5.1 实际场景 ​

5.2 文件上传表单 ​

第六部分：完整示例 ​

6.1 表单类 ​

6.2 模板 ​

6.3 视图函数 ​

第七部分：L3 专家层 ​

7.1 CSRF Token 的生成与验证原理（HMAC） ​

7.2 表单验证的装饰器模式 ​

7.3 文件上传的 multipart/form-data 解析过程 ​

7.4 知识关联 ​

05-表单处理

第一部分：Flask-WTF 基础

1.1 实际场景

1.2 安装

1.3 基本配置

第二部分：创建表单

2.1 实际场景

2.2 基础表单

2.3 完整注册表单

第三部分：模板中使用表单

3.1 实际场景

3.2 渲染表单

第四部分：处理表单数据

4.1 实际场景

4.2 视图函数

4.3 自定义验证

第五部分：文件上传表单

5.1 实际场景

5.2 文件上传表单

第六部分：完整示例

6.1 表单类

6.2 模板

6.3 视图函数

第七部分：L3 专家层

7.1 CSRF Token 的生成与验证原理（HMAC）

7.2 表单验证的装饰器模式

7.3 文件上传的 multipart/form-data 解析过程

7.4 知识关联