Flask（4）：wtforms組件 & 數據庫鏈接池 DBUtils

時間 2019-12-02

標籤 flask wtforms 組件數據庫鏈接 dbutils 欄目 Python 简体版

原文原文鏈接

wtforms 組件的做用：
　　--- 生成 HTML 標籤
　　--- form 表單驗證html

示例代碼：前端

app.pyhtml5

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.pymysql

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.htmlsql

<!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>

注：要積累本身的模板庫（如：後臺管理的前端模板頁面）flask

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 是指校驗一個字段遇到不合法時，再也不校驗該字段其它地方是否合法