在这里，我们把form表单验证的代码进行工具化了，以后稍微修改一下参数就可以拿来用了
先贴上代码
forms.py
from backend.form import fields class BaseForm: def __init__(self): self._value_dict = {} self._error_dict = {} self._valid_status = True def valid(self, handler): for field_name, field_obj in self.__dict__.items(): if field_name.startswith('_'): continue if type(field_obj) == fields.CheckBoxField: post_value = handler.get_arguments(field_name, None) elif type(field_obj) == fields.FileField: post_value = [] file_list = handler.request.files.get(field_name, []) for file_item in file_list: post_value.append(file_item['filename']) else: post_value = handler.get_argument(field_name, None) field_obj.match(field_name, post_value) if field_obj.is_valid: self._value_dict[field_name] = field_obj.value else: self._error_dict[field_name] = field_obj.error self._valid_status = False return self._valid_status
fields.py
import re import os class Field: def __init__(self): self.is_valid = False self.name = None self.value = None self.error = None def match(self, name, value): self.name = name if not self.required: self.is_valid = True self.value = value else: if not value: if self.custom_error_dict.get('required', None): self.error = self.custom_error_dict['required'] else: self.error = "%s is required" % name else: ret = re.match(self.REGULAR, value) if ret: self.is_valid = True self.value = value else: if self.custom_error_dict.get('valid', None): self.error = self.custom_error_dict['valid'] else: self.error = "%s is invalid" % name class StringField(Field): REGULAR = "^.*$" def __init__(self, custom_error_dict=None, required=True): self.custom_error_dict = {} # {'required': 'IP不能为空', 'valid': 'IP格式错误'} if custom_error_dict: self.custom_error_dict.update(custom_error_dict) self.required = required super(StringField, self).__init__() class IPField(Field): REGULAR = "^(25[0-5]|2[0-4]d|[0-1]?d?d)(.(25[0-5]|2[0-4]d|[0-1]?d?d)){3}$" def __init__(self, custom_error_dict=None, required=True): self.custom_error_dict = {} # {'required': 'IP不能为空', 'valid': 'IP格式错误'} if custom_error_dict: self.custom_error_dict.update(custom_error_dict) self.required = required super(IPField, self).__init__() class EmailField(Field): REGULAR = "^w+([-+.']w+)*@w+([-.]w+)*.w+([-.]w+)*$" def __init__(self, custom_error_dict=None, required=True): self.custom_error_dict = {} # {'required': 'IP不能为空', 'valid': 'IP格式错误'} if custom_error_dict: self.custom_error_dict.update(custom_error_dict) self.required = required super(EmailField, self).__init__() class IntegerField(Field): REGULAR = "^d+$" def __init__(self, custom_error_dict=None, required=True): self.custom_error_dict = {} # {'required': 'IP不能为空', 'valid': 'IP格式错误'} if custom_error_dict: self.custom_error_dict.update(custom_error_dict) self.required = required super(IntegerField, self).__init__() class CheckBoxField(Field): REGULAR = "^d+$" def __init__(self, custom_error_dict=None, required=True): self.custom_error_dict = {} # {'required': 'IP不能为空', 'valid': 'IP格式错误'} if custom_error_dict: self.custom_error_dict.update(custom_error_dict) self.required = required super(CheckBoxField, self).__init__() def match(self, name, value): self.name = name if not self.required: self.is_valid = True self.value = value else: if not value: if self.custom_error_dict.get('required', None): self.error = self.custom_error_dict['required'] else: self.error = "%s is required" % name else: if isinstance(name, list): self.is_valid = True self.value = value else: if self.custom_error_dict.get('valid', None): self.error = self.custom_error_dict['valid'] else: self.error = "%s is invalid" % name class FileField(Field): REGULAR = "^(w+.pdf)|(w+.mp3)|(w+.py)$" def __init__(self, custom_error_dict=None, required=True): self.custom_error_dict = {} # {'required': 'IP不能为空', 'valid': 'IP格式错误'} if custom_error_dict: self.custom_error_dict.update(custom_error_dict) self.required = required super(FileField, self).__init__() def match(self, name, file_name_list): flag = True self.name = name if not self.required: self.is_valid = True self.value = file_name_list else: if not file_name_list: if self.custom_error_dict.get('required', None): self.error = self.custom_error_dict['required'] else: self.error = "%s is required" % name flag = False else: for file_name in file_name_list: if not file_name or not file_name.strip(): if self.custom_error_dict.get('required', None): self.error = self.custom_error_dict['required'] else: self.error = "%s is required" % name flag = False break else: ret = re.match(self.REGULAR, file_name) if not ret: if self.custom_error_dict.get('valid', None): self.error = self.custom_error_dict['valid'] else: self.error = "%s is invalid" % name flag = False break self.is_valid = flag def save(self, request, upload_to=""): file_metas = request.files[self.name] for meta in file_metas: file_name = meta['filename'] file_path_name = os.path.join(upload_to, file_name) with open(file_path_name, 'wb') as up: up.write(meta['body']) upload_file_path_list = map(lambda path: os.path.join(upload_to, path), self.value) self.value = list(upload_file_path_list)
在form.py这个文件，做了一件什么事呢？代码就定义了父类，主要是判断要验证内容的类型，然后取值，然后又调用了在fields.py里类的match方法，最后把验证后结果信息返回（vaild方法）。
而在fields.py文件里，主要是对为空检测，合法性检测，并把检测结果返回给form.py的vaild方法里（match方法）
首先
-
form组件只做为空检测，合法性检测，并没做超时检测，内部可获取验证状态_valid_status--True/False，错误信息_error_dict，验证通过时的用户数据_value_dict
-
验证类型：字符串，IP，邮箱，数字，复选框，文件
再者，怎么用？
-
分析你的应用场景，需要对哪几个类型进行验证，定义一个类，把需要的验证类型写入到构造方法里，记得继承一下BaseForm类，并且继承一下父类的构造方法
-
在构造方法里，实例Field对象时，可以传入自定制错误类型信息custom_error_dict，required=False可为空设置
from backend.form.forms import BaseForm from backend.form.fields import StringField from backend.form.fields import IntegerField from backend.form.fields import EmailField class SendMsgForm(BaseForm): def __init__(self): self.email = EmailField(custom_error_dict={'required': '注册邮箱不能为空.', 'valid': '注册邮箱格式错误.'}) super(SendMsgForm, self).__init__() class RegisterForm(BaseForm): def __init__(self): self.username = StringField() self.email = EmailField() self.password = StringField() self.email_code = StringField() super(RegisterForm, self).__init__() class LoginForm(BaseForm): def __init__(self): self.user = StringField() self.pwd = StringField() self.code = StringField() super(LoginForm, self).__init__()
最后
-
在post方法里，调用一下form对象的vaild方法（把handler对象，也就是self传入），接下来只要根据form对象里检测完后的信息进行相应的操作
import io import datetime import json from backend.utils import check_code from backend.core.request_handler import BaseRequestHandler from forms import account from backend.utils.response import BaseResponse from backend import commons from models import chouti_orm as ORM from sqlalchemy import and_, or_ class CheckCodeHandler(BaseRequestHandler): def get(self, *args, **kwargs): stream = io.BytesIO() img, code = check_code.create_validate_code() img.save(stream, "png") self.session["CheckCode"] = code self.write(stream.getvalue()) class LoginHandler(BaseRequestHandler): def post(self, *args, **kwargs): #对象里有self.status=False，self.data=None，self.summary=None，self.message={} rep = BaseResponse() form = account.LoginForm() if form.valid(self): if form._value_dict['code'].lower() != self.session["CheckCode"].lower(): rep.message = {'code': '验证码错误'} self.write(json.dumps(rep.__dict__)) return conn = ORM.session() obj = conn.query(ORM.UserInfo).filter( or_( and_(ORM.UserInfo.email == form._value_dict['user'], ORM.UserInfo.password == form._value_dict['pwd']), and_(ORM.UserInfo.username == form._value_dict['user'], ORM.UserInfo.password == form._value_dict['pwd']) )).first() if not obj: rep.message = {'user': '用户名邮箱或密码错误'} self.write(json.dumps(rep.__dict__)) return self.session['is_login'] = True self.session['user_info'] = obj.__dict__ rep.status = True else: rep.message = form._error_dict self.write(json.dumps(rep.__dict__)) class RegisterHandler(BaseRequestHandler): def post(self, *args, **kwargs): rep = BaseResponse() form = account.RegisterForm() if form.valid(self): current_date = datetime.datetime.now() limit_day = current_date - datetime.timedelta(minutes=1) conn = ORM.session() is_valid_code = conn.query(ORM.SendMsg).filter(ORM.SendMsg.email == form._value_dict['email'], ORM.SendMsg.code == form._value_dict['email_code'], ORM.SendMsg.ctime > limit_day).count() if not is_valid_code: rep.message['email_code'] = '邮箱验证码不正确或过期' self.write(json.dumps(rep.__dict__)) return has_exists_email = conn.query(ORM.UserInfo).filter(ORM.UserInfo.email == form._value_dict['email']).count() if has_exists_email: rep.message['email'] = '邮箱已经存在' self.write(json.dumps(rep.__dict__)) return has_exists_username = conn.query(ORM.UserInfo).filter( ORM.UserInfo.username == form._value_dict['username']).count() if has_exists_username: rep.message['email'] = '用户名已经存在' self.write(json.dumps(rep.__dict__)) return form._value_dict['ctime'] = current_date form._value_dict.pop('email_code') obj = ORM.UserInfo(**form._value_dict) conn.add(obj) conn.query(ORM.SendMsg).filter_by(email=form._value_dict['email']).delete() conn.commit() self.session['is_login'] = True self.session['user_info'] = obj.__dict__ rep.status = True else: rep.message = form._error_dict self.write(json.dumps(rep.__dict__)) class SendMsgHandler(BaseRequestHandler): def post(self, *args, **kwargs): rep = BaseResponse() form = account.SendMsgForm() if form.valid(self): email = form._value_dict['email'] conn = ORM.session() has_exists_email = conn.query(ORM.UserInfo).filter(ORM.UserInfo.email == form._value_dict['email']).count() if has_exists_email: rep.summary = "此邮箱已经被注册" self.write(json.dumps(rep.__dict__)) return current_date = datetime.datetime.now() code = commons.random_code() count = conn.query(ORM.SendMsg).filter_by(**form._value_dict).count() if not count: insert = ORM.SendMsg(code=code, email=email, ctime=current_date) conn.add(insert) conn.commit() rep.status = True else: limit_day = current_date - datetime.timedelta(hours=1) times = conn.query(ORM.SendMsg).filter(ORM.SendMsg.email == email, ORM.SendMsg.ctime > limit_day, ORM.SendMsg.times >= 10, ).count() if times: rep.summary = "'已经超过今日最大次数（1小时后重试）'" else: unfreeze = conn.query(ORM.SendMsg).filter(ORM.SendMsg.email == email, ORM.SendMsg.ctime < limit_day).count() if unfreeze: conn.query(ORM.SendMsg).filter_by(email=email).update({"times": 0}) conn.query(ORM.SendMsg).filter_by(email=email).update({"times": ORM.SendMsg.times + 1, "code": code, "ctime": current_date}, synchronize_session="evaluate") conn.commit() rep.status = True else: rep.summary = form._error_dict['email'] self.write(json.dumps(rep.__dict__))