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)(?=.*[[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>

注:要打造自己的 模板库(如:后台管理的前端模板页面)

原文地址:https://www.cnblogs.com/neozheng/p/10230959.html

时间: 2024-07-30 02:57:44

Flask(4):wtforms组件 & 数据库连接池 DBUtils的相关文章

Python 数据库连接池DButils

常规的数据库链接存在的问题: 场景一: 缺点:每次请求反复创建数据库连接,连接数太多 import pymysql def index(): conn = pymysql.connect() cursor = conn.cursor() cursor.execute('select * from tb where id > %s',[5,]) result = cursor.fetchall() cursor.close() conn.close() print(result) def upda

Python数据库连接池DButils

DButils是python的一个实现数据库连接池的模块 两种模式: 1.为每一个线程创建一个链接,即使线程即使调用了close()方法,也不会关闭,只是把线程放到连接池,供自己再次使用,当连接关闭时,线程连接自动关闭. 1 from DBUtils.PersistentDB import PersistentDB 2 import pymysql 3 PooL = PersistentDB( 4 creator = pymysql, #使用链接数据库的模块 5 maxusage = None,

Python的数据库连接池DBUtils

DBUtils是Python的一个用于实现数据库连接池的模块. 此连接池有两种连接模式: 模式一:为每个线程创建一个连接,线程即使调用了close方法,也不会关闭,只是把连接重新放到连接池,供自己线程再次使用.当线程终止时,连接自动关闭. POOL = PersistentDB( creator=pymysql, # 使用链接数据库的模块 maxusage=None, # 一个链接最多被重复使用的次数,None表示无限制 setsession=[], # 开始会话前执行的命令列表.如:["set

数据库连接池(DBUtils)

DButils是python的一个实现数据库连接池的模块 两种模式: 1.为每一个线程创建一个链接,即使线程即使调用了close()方法,也不会关闭,只是把线程放到连接池,供自己再次使用,当连接关闭时,线程连接自动关闭. from DBUtils.PersistentDB import PersistentDB import pymysql PooL = PersistentDB( creator = pymysql, #使用链接数据库的模块 maxusage = None, #一个链接最多被使

11.2_数据库连接池&amp;DBUtils

一.数据库连接池 1. 数据库的连接对象创建工作,比较消耗性能. 2.一开始现在内存中开辟一块空间(集合) , 一开先往池子里面放置 多个连接对象. 后面需要连接的话,直接从池子里面去.不要去自己创建连接了. 使用完毕, 要记得归还连接.确保连接对象能循环利用. 二.开源连接池 1.DBCP 1. 导入jar文件 2. 不使用配置文件: public void testDBCP01(){ Connection conn = null; PreparedStatement ps = null; t

数据库连接池 DBUtils:

import pymysqlfrom DBUtils.PooledDB import PooledDB, SharedDBConnectionPOOL = PooledDB ( creator=pymysql, # 使用链接数据库的模块 maxconnections=10, # 连接池允许的最大连接数,0和None表示不限制连接数 mincached=2, # 初始化时,链接池中至少创建的空闲的链接,0表示不创建 maxcached=5, # 链接池中最多闲置的链接,0和None不限制 maxs

Java Web(五) -- 事务 &amp; 数据库连接池 &amp; DBUtiles

#事务&数据库连接池&DBUtils ##事务 > Transaction 其实指的一组操作,里面包含许多个单一的逻辑.只要有一个逻辑没有执行成功,那么都算失败. 所有的数据都回归到最初的状态(回滚) * 为什么要有事务? > 为了确保逻辑的成功. 例子: 银行的转账. ###使用命令行方式演示事务. * 开启事务 start transaction; * 提交或者回滚事务 commit; 提交事务, 数据将会写到磁盘上的数据库 rollback ; 数据回滚,回到最初的状态.

13-事务&amp;数据库连接池&amp;DBUtiles

事务&数据库连接池&DBUtils 事务 Transaction  其实指的一组操作,里面包含许多个单一的逻辑.只要有一个逻辑没有执行成功,那么都算失败. 所有的数据都回归到最初的状态(回滚) 为什么要有事务? 为了确保逻辑的成功. 例子: 银行的转账. 使用命令行方式演示事务. 开启事务 start transaction; 提交或者回滚事务 commit; 提交事务, 数据将会写到磁盘上的数据库 rollback ;  数据回滚,回到最初的状态. 1. 关闭自动提交功能. 2. 演示事

十三 .Flask wtforms组件和选择框动态数据实时更新

一 . wtforms组件     选择框动态数据实时更新 https://www.cnblogs.com/lovershowtime/p/11384494.html 1.wtforms组件使用(登录) 1. 用户登录 当用户登录时候,需要对用户提交的用户名和密码进行多种格式校验.如: 用户不能为空:用户长度必须大于6: 密码不能为空:密码长度必须大于12:密码必须包含 字母.数字.特殊字符等(自定义正则): login.html <!DOCTYPE html> <html lang=&