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

Posted 2021-02-01 neozheng

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)(?=.*[[email protected]$!%*?&])[A-Za-z[email protected]$!%*?&]{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__ 的方法来解决该问题：保证每次实例化时都要从数据库取一次
        # （类中的静态字段只会在程序启动时加载一次）
        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=‘123456‘,
    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>

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