Flask(4):wtforms組件 & 數據庫連接池 DBUtils
阿新 • • 發佈:2019-01-07
下拉列表 上海 sub multi html_ one regex ext 沒有
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__ 的方法來解決該問題:保證每次實例化時都要從數據庫取一次 # (類中的靜態字段只會在程序啟動時加載一次) 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>
註:要打造自己的 模板庫(如:後臺管理的前端模板頁面)
Flask(4):wtforms組件 & 數據庫連接池 DBUtils