Flask（4）：wtforms组件 & 数据库连接池 DBUtils

wtforms 组件的作用：
　　--- 生成 HTML 标签
　　--- form 表单验证

示例代码：

app.py

from flask import Flask, render_template, request

from wtforms import Form

from wtforms.fields import simple

from wtforms.fields import core

from wtforms.fields import html5

from wtforms import widgets

from wtforms import validators

app = Flask(__name__)

class LoginForm(Form):

    name = simple.StringField(

        label='用户名',

        validators=[  # 校验器

            validators.DataRequired(message='用户名不能为空.'),  # 不能为空

            validators.Length(min=6, max=18, message='用户名长度必须大于%(min)d且小于%(max)d')  # 长度限制

        ],

        render_kw={"placeholder": "请输入用户名", "class": "username"}

    )

    # render_kw = {} 中是自定义 input 标签的属性

    psw = simple.PasswordField(  # simple.PasswordField 定义了 input 标签的 type 属性

        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个特殊字符')

        ],

        render_kw={"placeholder": "请输入密码"}

    )

    postscript = simple.StringField(

        widget=widgets.TextArea(),  # 将输入框变成了文本域（textarea）；默认的是 widgets.TextInput

    )

@app.route('/login', methods=["GET", "POST"])

def login():

    if request.method == "GET":

        form = LoginForm()

        print(form.name)

        print(type(form.name))

        # 打印结果：

        # <input id="name" name="name" type="text" value="">

        # <class 'wtforms.fields.core.StringField'>   # 所以 form.name 是 StringField 一个对象；StringField().__str__

        return render_template("login.html", form=form)

    form = LoginForm(formdata=request.form)  # 传入 POST 请求的数据 进行 LoginForm 的实例化

    if form.validate():  # form 进行 校验

        print("验证成功")

        print(form.data)  # 校验后的“干净”数据

        return "登陆成功"

    else:

        print(form.errors)  # 校验的错误信息；全部的错误信息 （在template中取错误信息时不要用这种写法）

        print(form.name.errors)  # name字段中的错误信息

        print(form.psw.errors[0])  # name字段中的错误信息； form.psw.errors[0] 这种写法即使 psw字段中没有错误也不会报错

        return render_template("login.html", form=form)

# 用户注册：

# 注册页面需要让用户输入：用户名、密码、密码重复、性别、爱好等。

class RegisterForm(Form):

    name = simple.StringField(

        label='用户名',

        validators=[

            validators.DataRequired()

        ],

        widget=widgets.TextInput(),

        render_kw={'class': 'form-control'},

        default='neo'

    )

    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="两次密码输入不一致")  # validators.EqualTo(字段)  用于判断两个字段的值是否相等

        ],

        widget=widgets.PasswordInput(),

        render_kw={'class': 'form-control'}

    )

    email = html5.EmailField(  # EmailField 在 wtforms.fields 中的 html5 中

        label='邮箱',

        validators=[

            validators.DataRequired(message='邮箱不能为空.'),

            validators.Email(message='邮箱格式错误')  # 要求是邮箱格式

        ],

        widget=widgets.TextInput(input_type='email'),

        render_kw={'class': 'form-control'}

    )

    gender = core.RadioField(  # core.RadioField 是单选框； <input type="radio">

        label='性别',

        choices=(

            (1, '男'),

            (2, '女'),

        ),

        coerce=int  # 作用： int("1") ，即把前端传过来的字符串自动转成 int 类型

    )

    city = core.SelectField(  # core.SelectField ：下拉列表（单选）；<select></select>

        label='城市',

        choices=(

            ('bj', '北京'),

            ('sh', '上海'),

        )

    )

    hobby = core.SelectMultipleField(  # core.SelectMultipleField ：（多选）；<select multiple="">

        label='爱好',

        choices=(

            (1, '篮球'),

            (2, '足球'),

        ),

        coerce=int

    )

    favor = core.SelectMultipleField(  #

        label='喜好',

        choices=(

            (1, '篮球'),

            (2, '足球'),

        ),

        widget=widgets.ListWidget(prefix_label=False),

        # wtforms.widgets.ListWidget(html_tag='ul', prefix_label=True) ： Renders a list of fields as a ul or ol list.

        option_widget=widgets.CheckboxInput(),

        # core.SelectMultipleField 和 option_widget=widgets.CheckboxInput() ：多选框；<input type="checkbox">

        coerce=int,

        default=[1, 2]  # 默认值（此处为默认选中的）

    )

@app.route('/register', methods=['GET', 'POST'])

def register():

    if request.method == 'GET':

        form = RegisterForm(data={'gender': 1})

        return render_template('register.html', form=form)

    form = RegisterForm(formdata=request.form)

    if form.validate():

        print('用户提交数据通过格式验证，提交的值为：', form.data)

    else:

        print(form.errors)

    return render_template('register.html', form=form)

import helper

class UserForm(Form):

    name = simple.StringField(label='姓名')

    city = core.SelectField(

        label='城市',

        choices=(),  # choices初始为一个空元组

        # choices 需要从数据库中动态调取，所以不能在此处写死了；

        # 如果写成：choices = helper.fetch_all("select id,city_name from tb1",[],type=None ) 会出现一个问题：

        # choices为面向对象的静态字段，所以上述代码只有在程序启动的时候才去数据库执行 fetch_all() 的操作，因此在程序运行时，当你在 tb1 中插入新的数据（id,city_name）或删除数据后,只要程序不重启，那么浏览器将不能加载出来你刚新插入的 id 和 city_name

        # 但生产环境下不能总是重启服务器程序，所以可以利用下面的 重构 __init__ 的方法来解决该问题：保证每次实例化时都要从数据库取一次

        # （类中的静态字段只会在程序启动时加载一次； Django的Form组件也有这个问题，解决办法也是重写 构造方法 __init__ ）

        coerce=int

    )

    def __init__(self, *args, **kwargs):

        """

        每次类实例化的时候，都要从数据库取一次

        :param args:

        :param kwargs:

        """

        super(UserForm, self).__init__(*args, **kwargs)

        self.city.choices = helper.fetch_all('select id,name from tb1', [], type=None)  # 每次实例化都会从数据库取数据

@app.route('/user')

def user():

    if request.method == "GET":

        # form = UserForm(data={'name':'neo','city':3})

        form = UserForm()

        return render_template('user.html', form=form)

if __name__ == '__main__':

    app.run()

helper.py

import pymysql

from DBUtils.PooledDB import PooledDB, SharedDBConnection  # DBUtils 是第三方插件，需要事先安装

POOL = PooledDB(  # POOL 这个配置也可写在 配置文件中（最好写在配置文件中）

    creator=pymysql,  # 使用链接数据库的模块

    maxconnections=6,  # 连接池允许的最大连接数，0和None表示不限制连接数

    mincached=2,  # 初始化时，链接池中至少创建的空闲的链接，0表示不创建

    maxcached=5,  # 链接池中最多闲置的链接，0和None不限制

    maxshared=3,

    # 链接池中最多共享的链接数量，0和None表示全部共享。PS: 无用，因为pymysql和MySQLdb等模块的 threadsafety都为1，所有值无论设置为多少，_maxcached永远为0，所以永远是所有链接都共享。

    blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错

    maxusage=None,  # 一个链接最多被重复使用的次数，None表示无限制

    setsession=[],  # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]

    ping=0,  # 0 或 4 用的比较多

    # ping MySQL服务端，检查是否服务可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always

    host='127.0.0.1',

    port=3306,

    user='root',

    password='',

    database='dbwtforms',

    charset='utf8'

)

def connect(type):

    """

    连接数据库连接池

    :param type: 查询结果的表示形式是字典还是元组

    :return: conn,cursor （连接和游标）

    """

    conn = POOL.connection()  # POOL.connection() ：表示连接 DBUtils 的数据库连接池 （DBUtils的语法）；如果 POOL写在了配置文件中，则是：conn = Config.POOL.connection()

    cursor = conn.cursor(

        cursor=type)  # cursor 参数用于表示 从数据库获取到结果 是否表示为 字典的形式；如：type=pymysql.cursors.DictCursor 为字典形式， type=None 为元组形式

    return conn, cursor

def connect_close(conn, cursor):

    """

    关闭 连接 和 游标

    :param conn:

    :param cursor:

    :return:

    """

    cursor.close()

    conn.close()

def fetch_all(sql, args, type=pymysql.cursors.DictCursor):

    """

    查询所有

    :param sql:

    :param args:

    :param type:

    :return:

    """

    conn, cursor = connect(type)

    cursor.execute(sql, args)  # sql 为 sql 语句； args 为 sql语句中的占位符（如果sql 中不需要占位符，args就传入一个空列表 [] 或者 空元组 ()）

    record_list = cursor.fetchall()

    connect_close(conn, cursor)

    return record_list

def fetch_one(sql, args,type=None):

    """

    查询单个

    :param sql:

    :param args:

    :param type:

    :return:

    """

    conn, cursor = connect(type)

    cursor.execute(sql, args)

    result = cursor.fetchone()

    connect_close(conn, cursor)

    return result

def insert(sql, args, type=None):

    """

    插入数据

    :param sql:

    :param args:

    :param type:

    :return:

    """

    conn, cursor = connect(type)

    row = cursor.execute(sql, args)

    conn.commit()  # 插入数据之后要提交 conn.commit()

    connect_close(conn, cursor)

    return row

"""

注意：1. 写 SQL语句一定要用 数据库连接池

      2. 封装自己的SQL方法（可以封装成一个类）

"""

templates/login.html

<!DOCTYPE html>

<html lang="en">

<head>

    <meta charset="UTF-8">

    <title>Title</title>

</head>

<body>

    <form action="" method="post" novalidate>

        <p>{{ form.name }}{{ form.name.errors[0] }}</p>

        <p>{{ form.psw.label }}{{ form.psw }}</p>

        <p>{{ form.postscript }}</p>

        <p><input type="submit" value="提交"></p>

    </form>

</body>

</html>

templates/register.html

<!DOCTYPE html>

<html lang="en">

<head>

    <meta charset="UTF-8">

    <title>Title</title>

</head>

<body>

    <h2>用户注册</h2>

    <form action="" method="post">

        {# form 对象能够直接 for 遍历；按照字段在类中的定义顺序来渲染相应的标签 #}

        {% for field in form %}

            <p>{{ field.label}}：{{ field }} <span>{{ field.errors[0] }}</span></p>

        {% endfor %}

        <p><input type="submit" value="提交"></p>

    </form>

</body>

</html>

注：要积累自己的模板库（如：后台管理的前端模板页面）

wtforms的钩子函数：

class RegisterForm(Form):

    name = simple.StringField(

        label='用户名',

        validators=[

            validators.DataRequired()

        ],

        widget=widgets.TextInput(),

        render_kw={'class': 'form-control'},

        default='neo'

    )

    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="两次密码输入不一致")  # validators.EqualTo(字段)  用于判断两个字段的值是否相等

        ],

        widget=widgets.PasswordInput(),

        render_kw={'class': 'form-control'}

    )

    # 钩子函数

    def validate_pwd_confirm(self,field):

        """

        自定义 pwd_confirm 字段检验规则，例如：与 pwd 字段是否一致；语法： validate_字段

        :param field:

        :return:

        """

        print("field.data",field.data)  # field.data 表示 所要 validate_字段 的值

        print("self.data",self.data)  # self.data 表示 所有 “干净”的数据（字典的形式）

        if field.data != self.data['pwd']:

            # raise validators.ValidationError("密码不一致") # 继续后续验证

            raise validators.StopValidation("密码不一致")  # 不再继续后续验证

            # validate_字段 有两种 Validation ：ValidationError 和 StopValidation；ValidationError表示遇到检验不合法时，还会该检验该字段其它地方是否合法； StopValidation 是指校验一个字段遇到不合法时，不再校验该字段其它地方是否合法

巴特西

Flask（4）：wtforms组件 & 数据库连接池 DBUtils

wtforms的钩子函数：

最新文章

热门文章