简介
WTForms是一个支持多个web框架的form组件，主要用于对用户请求数据进行验证。
用户登录注册示例
1. 用户登录
当用户登录时候，需要对用户提交的用户名和密码进行多种格式校验。如：
用户不能为空；用户长度必须大于6；
密码不能为空；密码长度必须大于12；密码必须包含 字母、数字、特殊字符等（自定义正则）；
from flask import Blueprint,render_template,request,session,redirectfrom ..utils.sql import SQLHelperaccount = Blueprint('account',__name__)from wtforms import Formfrom wtforms.fields import simple,core,html5from wtforms import validatorsfrom wtforms import widgetsclass LoginForm(Form): user = simple.StringField( validators=[ validators.DataRequired(message='用户名不能为空.'), # validators.Length(min=6, max=18, message='用户名长度必须大于%(min)d且小于%(max)d') ], widget=widgets.TextInput(), render_kw={'class': 'form-control'} ) pwd = simple.PasswordField( validators=[ validators.DataRequired(message='密码不能为空.'), # validators.Length(min=8, message='用户名长度必须大于%(min)d'), # validators.Regexp(regex="^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)(?=.*[$@$!%*?&])[A-Za-z\d$@$!%*?&]{8,}", # message='密码至少8个字符，至少1个大写字母，1个小写字母，1个数字和1个特殊字符') ], widget=widgets.PasswordInput(), render_kw={'class': 'form-control'} )@account.route('/login',methods=['GET',"POST"])def login(): if request.method == 'GET': form = LoginForm() return render_template('login.html',form=form) form = LoginForm(formdata=request.form) if not form.validate(): return render_template('login.html', form=form) obj = SQLHelper.fetch_one("select id,name from users where name=%(user)s and pwd=%(pwd)s", form.data) if obj: session.permanent = True session['user_info'] = {'id':obj['id'], 'name':obj['name']} return redirect('/index') else: return render_template('login.html',msg='用户名或密码错误',form=form) 使用方法和django的form组件类似
前端页面
Title 登录
2. 用户注册
注册页面需要让用户输入：用户名、密码、密码重复、性别、爱好等。
class RegisterForm(Form): name = simple.StringField( label='用户名', validators=[ validators.DataRequired() ], widget=widgets.TextInput(), render_kw={'class': 'form-control'}, default='alex' ) pwd = simple.PasswordField( label='密码', validators=[ validators.DataRequired(message='密码不能为空.') ], widget=widgets.PasswordInput(), render_kw={'class': 'form-control'} ) pwd_confirm = simple.PasswordField( label='重复密码', validators=[ validators.DataRequired(message='重复密码不能为空.'), validators.EqualTo('pwd', message="两次密码输入不一致") ], widget=widgets.PasswordInput(), render_kw={'class': 'form-control'} ) email = html5.EmailField( label='邮箱', validators=[ validators.DataRequired(message='邮箱不能为空.'), validators.Email(message='邮箱格式错误') ], widget=widgets.TextInput(input_type='email'), render_kw={'class': 'form-control'} ) gender = core.RadioField( label='性别', choices=( (1, '男'), (2, '女'), ), coerce=int ) city = core.SelectField( label='城市', choices=SQLHelper.fetch_all('select id,name from city',{},None), # choices=( # (1, '篮球'), # (2, '足球'), # ), coerce=int ) hobby = core.SelectMultipleField( label='爱好', choices=( (1, '篮球'), (2, '足球'), ), coerce=int ) favor = core.SelectMultipleField( label='喜好', choices=( (1, '篮球'), (2, '足球'), ), widget=widgets.ListWidget(prefix_label=False), option_widget=widgets.CheckboxInput(), coerce=int, default=[1, 2] ) def __init__(self, *args, **kwargs): super(RegisterForm, self).__init__(*args, **kwargs) self.city.choices = SQLHelper.fetch_all('select id,name from city',{},None) def validate_name(self, field): """ 自定义pwd_confirm字段规则，例：与pwd字段是否一致 :param field: :return: """ # 最开始初始化时，self.data中已经有所有的值 # print(field.data) # 当前name传过来的值 # print(self.data) # 当前传过来的所有的值：name,gender..... obj = SQLHelper.fetch_one('select id from users where name=%s',[field.data,]) if obj: raise validators.ValidationError("用户名已经存在") # 继续后续验证 # raise validators.StopValidation("用户名已经存在") # 不再继续后续验证@account.route('/register',methods=['GET','POST'])def register(): if request.method == 'GET': form = RegisterForm() return render_template('register.html',form=form) form = RegisterForm(formdata=request.form) if form.validate(): print(form.data) else: print(form.errors) return "sdfasdfasdf" 这里我们的城市是从数据库取值的，为了保证每次的值都是最新的，所以我们自定义了一个__init__方法，每次实例化时都从数据库中取城市
validate_name相当于form组件中的钩子
前端页面
Title 用户注册
3. meta
#!/usr/bin/env python# -*- coding:utf-8 -*-from flask import Flask, render_template, request, redirect, sessionfrom wtforms import Formfrom wtforms.csrf.core import CSRFfrom wtforms.fields import corefrom wtforms.fields import html5from wtforms.fields import simplefrom wtforms import validatorsfrom wtforms import widgetsfrom hashlib import md5app = Flask(__name__, template_folder='templates')app.debug = Trueclass MyCSRF(CSRF): """ Generate a CSRF token based on the user's IP. I am probably not very secure, so don't use me. """ def setup_form(self, form): self.csrf_context = form.meta.csrf_context() self.csrf_secret = form.meta.csrf_secret return super(MyCSRF, self).setup_form(form) def generate_csrf_token(self, csrf_token): gid = self.csrf_secret + self.csrf_context token = md5(gid.encode('utf-8')).hexdigest() return token def validate_csrf_token(self, form, field): print(field.data, field.current_token) if field.data != field.current_token: raise ValueError('Invalid CSRF')class TestForm(Form): name = html5.EmailField(label='用户名') pwd = simple.StringField(label='密码') class Meta: # -- CSRF # 是否自动生成CSRF标签 csrf = True # 生成CSRF标签name csrf_field_name = 'csrf_token' # 自动生成标签的值，加密用的csrf_secret csrf_secret = 'xxxxxx' # 自动生成标签的值，加密用的csrf_context csrf_context = lambda x: request.url # 生成和比较csrf标签 csrf_class = MyCSRF # -- i18n # 是否支持本地化 # locales = False locales = ('zh', 'en') # 是否对本地化进行缓存 cache_translations = True # 保存本地化缓存信息的字段 translations_cache = {}@app.route('/index/', methods=['GET', 'POST'])def index(): if request.method == 'GET': form = TestForm() else: form = TestForm(formdata=request.form) if form.validate(): print(form) return render_template('index.html', form=form)if __name__ == '__main__': app.run() 其他：
1. metaclass
class MyType(type): def __init__(self, *args, **kwargs): print('MyType创建类',self) super(MyType, self).__init__(*args, **kwargs) def __call__(self, *args, **kwargs): obj = super(MyType, self).__call__(*args, **kwargs) print('类创建对象', self, obj) return objclass Foo(object,metaclass=MyType): user = 'wupeiqi' age = 18obj = Foo() class MyType(type): def __init__(self, *args, **kwargs): super(MyType, self).__init__(*args, **kwargs) def __call__(cls, *args, **kwargs): v = dir(cls) obj = super(MyType, cls).__call__(*args, **kwargs) return objclass Foo(MyType('MyType', (object,), {})): user = 'wupeiqi' age = 18obj = Foo() class MyType(type): def __init__(self, *args, **kwargs): super(MyType, self).__init__(*args, **kwargs) def __call__(cls, *args, **kwargs): v = dir(cls) obj = super(MyType, cls).__call__(*args, **kwargs) return objdef with_metaclass(arg,base): return MyType('MyType', (base,), {})class Foo(with_metaclass(MyType,object)): user = 'wupeiqi' age = 18obj = Foo() 2. 实例化流程分析
# 源码流程 1. 执行type的 __call__ 方法，读取字段到静态字段 cls._unbound_fields 中； meta类读取到cls._wtforms_meta中 2. 执行构造方法 a. 循环cls._unbound_fields中的字段，并执行字段的bind方法，然后将返回值添加到 self._fields[name] 中。 即： _fields = { name: wtforms.fields.core.StringField(), } PS：由于字段中的__new__方法，实例化时：name = simple.StringField(label='用户名')，创建的是UnboundField(cls, *args, **kwargs)，当执行完bind之后，才变成执行 wtforms.fields.core.StringField() b. 循环_fields，为对象设置属性 for name, field in iteritems(self._fields): # Set all the fields to attributes so that they obscure the class # attributes with the same names. setattr(self, name, field) c. 执行process，为字段设置默认值：self.process(formdata, obj, data=data, **kwargs) 优先级：obj,data,formdata; 再循环执行每个字段的process方法，为每个字段设置值： for name, field, in iteritems(self._fields): if obj is not None and hasattr(obj, name): field.process(formdata, getattr(obj, name)) elif name in kwargs: field.process(formdata, kwargs[name]) else: field.process(formdata) 执行每个字段的process方法，为字段的data和字段的raw_data赋值 def process(self, formdata, data=unset_value): self.process_errors = [] if data is unset_value: try: data = self.default() except TypeError: data = self.default self.object_data = data try: self.process_data(data) except ValueError as e: self.process_errors.append(e.args[0]) if formdata: try: if self.name in formdata: self.raw_data = formdata.getlist(self.name) else: self.raw_data = [] self.process_formdata(self.raw_data) except ValueError as e: self.process_errors.append(e.args[0]) try: for filter in self.filters: self.data = filter(self.data) except ValueError as e: self.process_errors.append(e.args[0]) d. 页面上执行print(form.name) 时，打印标签 因为执行了： 字段的 __str__ 方法 字符的 __call__ 方法 self.meta.render_field(self, kwargs) def render_field(self, field, render_kw): other_kw = getattr(field, 'render_kw', None) if other_kw is not None: render_kw = dict(other_kw, **render_kw) return field.widget(field, **render_kw) 执行字段的插件对象的 __call__ 方法，返回标签字符串 3. 验证流程分析
a. 执行form的validate方法，获取钩子方法 def validate(self): extra = {} for name in self._fields: inline = getattr(self.__class__, 'validate_%s' % name, None) if inline is not None: extra[name] = [inline] return super(Form, self).validate(extra) b. 循环每一个字段，执行字段的 validate 方法进行校验(参数传递了钩子函数) def validate(self, extra_validators=None): self._errors = None success = True for name, field in iteritems(self._fields): if extra_validators is not None and name in extra_validators: extra = extra_validators[name] else: extra = tuple() if not field.validate(self, extra): success = False return success c. 每个字段进行验证时候 字段的pre_validate 【预留的扩展】 字段的_run_validation_chain，对正则和字段的钩子函数进行校验 字段的post_validate【预留的扩展】