Flask第十四篇- Flask-Session組件、WTForms組件、數據庫鏈接池（POOL）

本篇閱讀目錄

• 1、Flask-Session
• 2、WTForms組件
• 3、數據庫鏈接池（POOL）

 

1、Flask-Session

　　咱們使用過flask內置的session，知道它是把session存放在瀏覽器，即客戶端。今天要學習的flask-session是flask的第三方組件，看一下它和flask內置的session有什麼不一樣以及它的使用方法。

　　flask-session是flask框架的session組件，flask內置session使用簽名cookie保存，而該組件則將支持session保存到多個地方，如：

　　　　- redis

　　　　- memcached

　　　　- filesystem

　　　　- mongodb

　　　　- sqlalchmey

一、安裝flask-session

1	pip3 install flask - session

二、回顧flask自帶的session的使用方法

　　from flask import Flask, session

　　app = Flask(__name__)
　　app.secret_key = 'afhaslhg'    # 必定要有這句

　　@app.route('/')
　　def index():
    　　session['user'] = 'value'
    　　return 'hello'

　　if __name__ == '__main__':
　　    app.run(debug=True)

　　啓動程序，使用瀏覽器訪問http://127.0.0.1:5000時，會看到以下session：

三、flask-session的使用（以保存到redis中爲例）

　　from flask import Flask, session
　　from flask_session import Session   # 導入flask-session中的Session類
　　from redis import Redis

　　app = Flask(__name__)
　　# 對實例進行配置
　　app.config["SESSION_TYPE"] = "redis"
　　app.config["SESSION_REDIS"] = Redis(host="127.0.0.1",port=6379,db=6)
　　
　　Session(app)   # 把原來app中的 session 進行替換

　　@app.route('/')
　　def index():
　　    session['user'] = 'value'
　　    return 'hello'

　　if __name__ == '__main__':
　　    app.run(debug=True)

　　啓動程序（redis服務端要），瀏覽器訪問http://127.0.0.1:5000時，瀏覽器session以下圖：

　　打開redis客戶端，進行以下操做：

 

2、WTForms組件

　　WTForms是flask的組件，相似於django的modelform組件。

一、安裝

1	pip3 install wtforms

二、使用（以登錄和註冊爲例）

　　wtf.py文件：

from flask import Flask, render_template, request
from wtforms.fields import simple, core
from wtforms import validators
from wtforms import Form

app = Flask(__name__)

# 定義註冊類
class RegForm(Form):
    username = simple.StringField(
        label="用戶名",
        validators=[
            validators.DataRequired(message='該字段不能爲空'),
            validators.Length(min=3, max=10, message='用戶名必須3-10個字符')
        ],
        id="user_id",
        render_kw={"class": "user_name"}
    )
    password = simple.PasswordField(
        label="密碼",
        validators=[
            validators.DataRequired(message='該字段不能爲空'),
            validators.Length(min=6, max=12, message='用戶名必須6-12個字符')
        ],
        id="pwd",
        render_kw={"class": "pwd"}
    )
    repassword = simple.PasswordField(
        label="確認密碼",
        validators=[
            validators.EqualTo(fieldname='password', message='兩次密碼不一致')
        ],
        id="re_pwd",
        render_kw={"class": "re_pwd"}
    )
    email = simple.StringField(
        label="郵箱",
        validators=[
            validators.DataRequired(message='該字段不能爲空'),
            validators.Email(message='必須符合郵箱格式')
        ],
        id="email",
        render_kw={"class": "email"}
    )
    gender = core.RadioField(
        label='性別',
        coerce=int,   # 提交的數據類型，即1或者2的數據類型
        choices=(
            (1, '女'),  # 元組第一個元素是value，第二個元素是顯示的值
            (2, '男')
        ),
        default=1   # 默認值爲1
    )
    hobby = core.SelectMultipleField(
        label='愛好',
        validators=[validators.Length(min=1, max=3, message='愛好可爲1-3個')],
        coerce=str,  # 注意，類型爲str時，下面choices中每一個元組第一個值必須帶引號
        choices=(
            ('1', '足球'),
            ('2', '籃球'),
            ('3', '唱歌'),
            ('4', '跳舞')
        ),
        default=(1,3)  # 默認選中兩個
    )
    # button = simple.SubmitField()  # 渲染提交按鈕


# 定義登錄類
class LoginForm(Form):
    username = simple.StringField(
        label="用戶名", # lable標籤標記內容
        validators=[
            validators.DataRequired(message='該字段不能爲空'),
            validators.Length(min=3, max=10, message='用戶名必須3-10個字符')
        ], # 校驗條件,可迭代條件,由於可能校驗多個條件
        description='this is a description',  # 描述標記
        id="user_id",  # 標籤id
        widget=None,  # 默認組件(好比input type="text") 在StringField中已經被實例化了
        render_kw={"class":"my_login"}  # 添加屬性和值
    )
    password = simple.PasswordField(
        label="密碼",
        validators=[
            validators.DataRequired(message='該字段不能爲空'),
            validators.Length(min=6, max=12, message='用戶名必須6-12個字符')
        ],
        description='this is a description',
        id="pwd",
        default=None,
        render_kw={"class": "pwd"}
    )


@app.route("/reg", methods=["GET", "POST"])
def reg():
    if request.method == "GET":
        rf = RegForm()
        return render_template('reg.html', wtf=rf)
    else:
        rf = RegForm(request.form)
        if rf.validate():
            return rf.data.get('username')
        else:
            print(rf.data)
            print(rf.errors)
            return render_template('reg.html', wtf=rf)


@app.route("/login", methods=["GET", "POST"])
def login():
    if request.method == "GET":
        lf = LoginForm()  # 實例化登陸類
        return render_template('index.html', wtf=lf)
    else:
        lf = LoginForm(request.form)  # 將用戶提交數據傳入登錄類
        if lf.validate():  # 校驗用戶提交的數據
            return lf.data.get('username')  # 正確的在lf.data中
        else:   # 錯誤的在lf.errors中
            return render_template('index.html', wtf=lf)


app.run(debug=True)

　　reg.html文件：

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <title>註冊</title>
</head>
<body>
<form action="" method="post" novalidate>
  {% for field in wtf %}
    <p>
    {{ field.label }}
    {{ field }}
    {{ field.errors.0 }}
    </p>
  {% endfor %}
  <input type="submit" value="註冊">
</form>
</body>
</html>

　　login.html文件：

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <title>登陸</title>
</head>
<body>
<form action="" method="post" novalidate>
  <p>
    {{ wtf.username.label }}
    {{ wtf.username }}{{ wtf.username.errors.0 }}
  </p>
  <p>
    {{ wtf.password.label }}
    {{ wtf.password }}{{ wtf.password.errors.0 }}
  </p>
  <input type="submit" value="登陸">
</form>
</body>
</html>

3、數據庫鏈接池（POOL）

一、回顧pymysql（python操做數據庫的模塊）的使用

　　參考博客：http://www.javashuo.com/article/p-kbijeanu-d.html

二、DBUtils - python數據庫鏈接池

　　1）安裝DBUtils

1	pip3 install DBUtils

　　2）建立並使用鏈接池

　　　　dbpool.py文件：

import pymysql
from DBUtils.PooledDB import PooledDB

POOL = PooledDB(
    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,
    # 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="",
    charset="utf8",
    db="s15"
)

　　　　sqlhelper.py文件：

from dbpool import POOL
import pymysql

def create_conn():
    conn = POOL.connection()
    cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)

    return conn,cursor


def close_conn(conn,cursor):
    cursor.close()
    conn.close()


def insert(sql,args):
    conn,cursor = create_conn()
    res = cursor.execute(sql,args)
    conn.commit()
    close_conn(conn,cursor)
    return res

def fetch_one(sql,args):
    conn,cursor = create_conn()
    cursor.execute(sql,args)
    res = cursor.fetchone()
    close_conn(conn,cursor)
    return res

def fetch_all(sql,args):
    conn,cursor = create_conn()
    cursor.execute(sql,args)
    res = cursor.fetchall()
    close_conn(conn,cursor)
    return res


sql = "insert into users(name,age) VALUES (%s, %s)"
insert(sql,("mjj",9))

sql = "select * from users where name=%s and age=%s"
print(fetch_one(sql,("mjj",9)))