简介
WTForms是一个支持多个web框架的form组件,主要用于对用户请求数据进行验证。
安装:
pip3 install wtforms
用户登录注册示例
1. 用户登录
当用户登录时候,需要对用户提交的用户名和密码进行多种格式校验。如:
密码不能为空;密码长度必须大于12;密码必须包含 字母、数字、特殊字符等(自定义正则);
#!/usr/bin/env python# -*- coding:utf-8 -*-from flask import Flask, render_template, request, redirectfrom wtforms import Formfrom wtforms.fields import corefrom wtforms.fields import html5from wtforms.fields import simplefrom wtforms import validatorsfrom wtforms import widgetsapp = Flask(__name__, template_folder='templates')app.debug = Trueclass LoginForm(Form): name = simple.StringField( label='用户名', 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( label='密码', 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'} )@app.route('/login', methods=['GET', 'POST'])def login(): if request.method == 'GET': form = LoginForm() return render_template('login.html', form=form) else: form = LoginForm(formdata=request.form) if form.validate(): print('用户提交数据通过格式验证,提交的值为:', form.data) else: print(form.errors) return render_template('login.html', form=form)if __name__ == '__main__': app.run()
Title 登录
2. 用户注册
注册页面需要让用户输入:用户名、密码、密码重复、性别、爱好等。
from flask import Flask, render_template, request, redirectfrom wtforms import Formfrom wtforms.fields import corefrom wtforms.fields import html5from wtforms.fields import simplefrom wtforms import validatorsfrom wtforms import widgetsapp = Flask(__name__, template_folder='templates')app.debug = Trueclass 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=( ('bj', '北京'), ('sh', '上海'), ) ) 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.favor.choices = ((1, '篮球'), (2, '足球'), (3, '羽毛球')) def validate_pwd_confirm(self, field): """ 自定义pwd_confirm字段规则,例:与pwd字段是否一致 :param field: :return: """ # 最开始初始化时,self.data中已经有所有的值 if field.data != self.data['pwd']: # raise validators.ValidationError("密码不一致") # 继续后续验证 raise validators.StopValidation("密码不一致") # 不再继续后续验证@app.route('/register', methods=['GET', 'POST'])def register(): if request.method == 'GET': form = RegisterForm(data={ 'gender': 1}) return render_template('register.html', form=form) else: form = RegisterForm(formdata=request.form) if form.validate(): print('用户提交数据通过格式验证,提交的值为:', form.data) else: print(form.errors) return render_template('register.html', form=form)if __name__ == '__main__': app.run()
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【预留的扩展】
补充:
要做编辑的时候,我们给form对象就不能用formdata参数传数据了
如果是formdata传的数据,组件内部会用getlist取值
form = LginForm(obj = 查询出的对象) 组件内部会用句点符取值
form = LoginForm(data = 字典)字典中的key就是字段名,value就是对应的字段值,组件内部会用索引取值
校验成功的数据:from.data 校验失败的数据:form.errors
局部钩子:validate_字段名(form) 参数form就是request.form,用get方法获取到值来定制校验规则
源码流程
1、定义LoginForm类时:
执行创建Form类的FormMeta类的new和init方法,封装了两个类属性:cls._unbound_fields = None cls._wtforms_meta = None
class FormMeta(type): def __init__(cls, name, bases, attrs): print(2222) type.__init__(cls, name, bases, attrs) cls._unbound_fields = None cls._wtforms_meta = None
Form类中还定义了字段类,字段类加括号执行new和init方法,在new方法中返回UnboundField对象,这个对象其实就是帮我们给每个字段做排序,为了能够在
页面上有顺序的显示
class Field(object): def __new__(cls, *args, **kwargs): if '_form' in kwargs and '_name' in kwargs: return super(Field, cls).__new__(cls) else: return UnboundField(cls, *args, **kwargs)
2、实例化LoginForm对象时:
要执行FormMeta的call方法,先获取Form的所有内容,dir()方法,然后获取到所有字段对应的UnboundField对象按顺序放到一个列表中,把这个列表赋值给_unbound_fields属性,
然后循环所有的继承类,如果继承类中有Meta这个属性,就把这个类的Meta属性值放到列表中,然后创建Meta类继承这个列表中的所有类,把Meta类赋值给_wtforms_meta属性
class FormMeta(type): def __call__(cls, *args, **kwargs): """ Construct a new `Form` instance. Creates the `_unbound_fields` list and the internal `_wtforms_meta` subclass of the class Meta in order to allow a proper inheritance hierarchy. """ print(11111) if cls._unbound_fields is None: fields = [] for name in dir(cls): if not name.startswith('_'): unbound_field = getattr(cls, name) if hasattr(unbound_field, '_formfield'): fields.append((name, unbound_field)) # We keep the name as the second element of the sort # to ensure a stable sort. fields.sort(key=lambda x: (x[1].creation_counter, x[0])) cls._unbound_fields = fields # Create a subclass of the 'class Meta' using all the ancestors. if cls._wtforms_meta is None: bases = [] for mro_class in cls.__mro__: if 'Meta' in mro_class.__dict__: bases.append(mro_class.Meta) cls._wtforms_meta = type('Meta', tuple(bases), {}) return type.__call__(cls, *args, **kwargs)
然后要执行Form类的new和init方法,在init方法中会先实例Meta对象(就是_wtforms_meta属性值),这个Meta对象就是用来帮我们生成csrf跨域请求隐藏的标签的,继续执行init方法,把这个LoginForm对象封装一个_fileds属性,属性值是字典,Key是LoginForm中对应的字段名,vlaue是一个个的字段对象,此时就不是UnboundField的了,而是对应的StringField对象了,然后又会循环这个字典,把字典的key也就是LoginForm中对应的字段名作为LoginForm对象的属性,属性值就是StringField对象,所以这个LoginForm对象直接点字段名就可以得到具体的字段对象StringField(当然还有别的)
class Form(with_metaclass(FormMeta, BaseForm)): Meta = DefaultMeta def __init__(self, formdata=None, obj=None, prefix='', data=None, meta=None, **kwargs): meta_obj = self._wtforms_meta() if meta is not None and isinstance(meta, dict): meta_obj.update_values(meta) super(Form, self).__init__(self._unbound_fields, meta=meta_obj, prefix=prefix) for name, field in iteritems(self._fields): setattr(self, name, field) self.process(formdata, obj, data=data, **kwargs)
3、当用LoginForm对象.字段名时:
.字段名首先得到对应的StringField对象,打印这个对象调用StringField的str方法,在str方法中又调用StringField继承的Field类的call()方法,调用Meta类(没有就找继承类DefaultMeta)的render_field方法,然后又调用StringField对象中widget参数对应的TextInput类继承Input类的call方法,这个方法就会返回一个input标签字符串,这也就是为什么我们直接点字段名就可以显示出对应的标签了
def __call__(self, **kwargs): return self.meta.render_field(self, kwargs)
class Input(object): def __call__(self, field, **kwargs): kwargs.setdefault('id', field.id) kwargs.setdefault('type', self.input_type) if 'value' not in kwargs: kwargs['value'] = field._value() return HTMLString('' % self.html_params(name=field.name, **kwargs))
4、校验流程
form.validate()做校验
在Form类的validate方法中,先拿到所有字段的钩子函数,然后把字段名作为key,钩子函数作为value值构建一个字典,然后把这个字典作为参数执行Form类的继承类BaseForm的validate方法
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)
然后循环所有的字段,判断字段名在不在之前构建好的字典,如果在就获取到这个字段对应的钩子函数,并把这个钩子函数作为参数执行当前字段对象的valiedate方法,也就是StringField的validate方法,没有就找父类的
BaseForm的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
先把当前字段的钩子函数和定义字段对象时的validators参数对应的校验规则做链式操作chain,然后执行_run_validation_chain方法,在这个方法中就把钩子函数和validators参数的校验规则一一对数据做验证,如果有错就把错误信息append到errors中
StringField的validate方法:
def validate(self, form, extra_validators=tuple()): self.errors = list(self.process_errors) stop_validation = False # Call pre_validate try: self.pre_validate(form) except StopValidation as e: if e.args and e.args[0]: self.errors.append(e.args[0]) stop_validation = True except ValueError as e: self.errors.append(e.args[0]) # Run validators if not stop_validation: chain = itertools.chain(self.validators, extra_validators) stop_validation = self._run_validation_chain(form, chain) # Call post_validate try: self.post_validate(form, stop_validation) except ValueError as e: self.errors.append(e.args[0]) return len(self.errors) == 0