Flask - 使用MySQL數據池

helper.py

 1 import pymysql
 2 
 3 from settings import Config
 4 
 5 def connect():
 6     conn = Config.POOL.connection()
 7     cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)  # 以 字典的方式 顯示
 8     return conn,cursor
 9 
10 
11 def connect_close(conn,cursor):
12     cursor.close()
13     conn.close()
14 
15 def fetch_all(sql,args):
16     conn,cursor = connect()
17 
18     cursor.execute(sql, args)
19     record_list = cursor.fetchall()
20     connect_close(conn,cursor)
21 
22     return record_list
23 
24 
25 def fetch_one(sql, args):
26     conn, cursor = connect()
27     cursor.execute(sql, args)
28     result = cursor.fetchone()
29     connect_close(conn, cursor)
30 
31     return result
32 
33 
34 def insert(sql, args):
35     conn, cursor = connect()
36     row = cursor.execute(sql, args)
37     conn.commit()
38     connect_close(conn, cursor)
39     return row

View Code

 

settings.py

 1 from DBUtils.PooledDB import PooledDB, SharedDBConnection
 2 import pymysql
 3 # 下載  DBUtils : pip3 install DBUtils
 4 class Config(object):
 5     SALT = b"sdf1123df"
 6     SECRET_KEY = 'asdf123sdfsdfsdf'
 7     MAX_CONTENT_LENGTH = 1024 * 1024 * 7
 8 
 9 
10     POOL = PooledDB(
11         creator=pymysql,  # 使用連接數據庫的模塊
12         maxconnections=6,  # 鏈接池容許的最大鏈接數，0和None表示不限制鏈接數
13         mincached=2,  # 初始化時，連接池中至少建立的空閒的連接，0表示不建立
14         maxcached=5,  # 連接池中最多閒置的連接，0和None不限制
15         maxshared=3,
16         # 連接池中最多共享的連接數量，0和None表示所有共享。PS: 無用，由於pymysql和MySQLdb等模塊的 threadsafety都爲1，全部值不管設置爲多少，_maxcached永遠爲0，因此永遠是全部連接都共享。
17         blocking=True,  # 鏈接池中若是沒有可用鏈接後，是否阻塞等待。True，等待；False，不等待而後報錯
18         maxusage=None,  # 一個連接最多被重複使用的次數，None表示無限制
19         setsession=[],  # 開始會話前執行的命令列表。如：["set datestyle to ...", "set time zone ..."]
20         ping=0,
21         # 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
22         host='127.0.0.1',
23         port=3306,
24         user='root',
25         password='123456',
26         database='db2',
27         charset='utf8'
28     )

View Code

 

views.py

 1 @account.route('/login/', methods=['GET', 'POST'])
 2 def login():
 3     '''
 4     登錄
 5     :return:
 6     '''
 7     if request.method == 'GET':
 8         return render_template('login.html')
 9 
10     username = request.form.get('user')
11     password = request.form.get('pwd')
12 
13     pwd_md5 = md5(password)
14 
15     data = helper.fetch_one("select id,nickname from userinfo where user=%s and pwd =%s", (username, pwd_md5))
16 
17     if not data:
18         return render_template('login.html', error='用戶名密碼錯誤')
19 
20     # session['user_info'] = data
21     session['user_info'] = {'id': data['id'], 'nickname': data['nickname']}
22 
23     return redirect('/home/')

View Code