本篇文章為大家展示了怎么在Flask框架中使用DBUtils模塊連接數(shù)據(jù)庫,內(nèi)容簡明扼要并且容易理解,絕對能使你眼前一亮,通過這篇文章的詳細(xì)介紹希望你能有所收獲。
成都創(chuàng)新互聯(lián)公司是一家專注于網(wǎng)站設(shè)計(jì)、成都網(wǎng)站制作與策劃設(shè)計(jì),伍家崗網(wǎng)站建設(shè)哪家好?成都創(chuàng)新互聯(lián)公司做網(wǎng)站,專注于網(wǎng)站建設(shè)十余年,網(wǎng)設(shè)計(jì)領(lǐng)域的專業(yè)建站公司;建站業(yè)務(wù)涵蓋:伍家崗等地區(qū)。伍家崗做網(wǎng)站價(jià)格咨詢:18982081108Flask連接數(shù)據(jù)庫
數(shù)據(jù)庫連接池:
Django使用:django ORM(pymysql/MySqldb)
Flask/其他使用:
-原生SQL
-pymysql(支持python2/3)
-MySqldb(支持python2)
-SQLAchemy(ORM)
原生SQL
需要解決的問題:
-不能為每個(gè)用戶創(chuàng)建一個(gè)連接
-創(chuàng)建一定數(shù)量的連接池,如果有人來
使用DBUtils模塊
兩種使用模式:
1 為每個(gè)線程創(chuàng)建一個(gè)連接,連接不可控,需要控制線程數(shù)
2 創(chuàng)建指定數(shù)量的連接在連接池,當(dāng)線程訪問的時(shí)候去取,如果不夠了線程排隊(duì),直到有人釋放。平時(shí)建議使用這種?。?!
模式一:
import pymysql from DBUtils.PersistentDB import PersistentDB POOL = PersistentDB( creator=pymysql, # 使用鏈接數(shù)據(jù)庫的模塊 maxusage=None, # 一個(gè)鏈接最多被重復(fù)使用的次數(shù),None表示無限制 setsession=[], # 開始會話前執(zhí)行的命令列表。如:["set datestyle to ...", "set time zone ..."] ping=0, # ping MySQL服務(wù)端,檢查是否服務(wù)可用。# 如:0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always closeable=False, # 建議為False,如果為False時(shí), conn.close() 實(shí)際上被忽略,供下次使用,再線程關(guān)閉時(shí),才會自動關(guān)閉鏈接。如果為True時(shí), conn.close()則關(guān)閉鏈接,那么再次調(diào)用pool.connection時(shí)就會報(bào)錯,因?yàn)橐呀?jīng)真的關(guān)閉了連接(pool.steady_connection()可以獲取一個(gè)新的鏈接) threadlocal=None, # 本線程獨(dú)享值得對象,用于保存鏈接對象,如果鏈接對象被重置 host='127.0.0.1', port=3306, user='root', password='123', database='pooldb', charset='utf8' ) def func(): conn = POOL.connection(shareable=False) cursor = conn.cursor() cursor.execute('select * from tb1') result = cursor.fetchall() cursor.close() conn.close() func()
模式二(推薦):
import time import pymysql import threading from DBUtils.PooledDB import PooledDB, SharedDBConnection POOL = PooledDB( creator=pymysql, # 使用鏈接數(shù)據(jù)庫的模塊 maxconnections=6, # 連接池允許的大連接數(shù),0和None表示不限制連接數(shù) mincached=2, # 初始化時(shí),鏈接池中至少創(chuàng)建的空閑的鏈接,0表示不創(chuàng)建 maxcached=5, # 鏈接池中最多閑置的鏈接,0和None不限制 maxshared=3, # 鏈接池中最多共享的鏈接數(shù)量,0和None表示全部共享。PS: 無用,因?yàn)閜ymysql和MySQLdb等模塊的 threadsafety都為1,所有值無論設(shè)置為多少,_maxcached永遠(yuǎn)為0,所以永遠(yuǎn)是所有鏈接都共享。 blocking=True, # 連接池中如果沒有可用連接后,是否阻塞等待。True,等待;False,不等待然后報(bào)錯 maxusage=None, # 一個(gè)鏈接最多被重復(fù)使用的次數(shù),None表示無限制 setsession=[], # 開始會話前執(zhí)行的命令列表。如:["set datestyle to ...", "set time zone ..."] ping=0, # ping MySQL服務(wù)端,檢查是否服務(wù)可用。# 如: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='123', database='pooldb', charset='utf8' ) def func(): # 檢測當(dāng)前正在運(yùn)行連接數(shù)的是否小于大鏈接數(shù),如果不小于則:等待或報(bào)raise TooManyConnections異常 # 否則 # 則優(yōu)先去初始化時(shí)創(chuàng)建的鏈接中獲取鏈接 SteadyDBConnection。 # 然后將SteadyDBConnection對象封裝到PooledDedicatedDBConnection中并返回。 # 如果最開始創(chuàng)建的鏈接沒有鏈接,則去創(chuàng)建一個(gè)SteadyDBConnection對象,再封裝到PooledDedicatedDBConnection中并返回。 # 一旦關(guān)閉鏈接后,連接就返回到連接池讓后續(xù)線程繼續(xù)使用。 conn = POOL.connection() # print(th, '鏈接被拿走了', conn1._con) # print(th, '池子里目前有', pool._idle_cache, '\r\n') cursor = conn.cursor() cursor.execute('select * from tb1') result = cursor.fetchall() conn.close() func()
具體寫法:
通過導(dǎo)入的方式
app.py
from flask import Flask from db_helper import SQLHelper app = Flask(__name__) @app.route("/") def hello(): result = SQLHelper.fetch_one('select * from xxx',[]) print(result) return "Hello World" if __name__ == '__main__': app.run()
DBUTILs
以下為兩種寫法:
第一種是用靜態(tài)方法裝飾器,通過直接執(zhí)行類的方法來連接使用數(shù)據(jù)庫
第二種是通過實(shí)例化對象,通過對象來調(diào)用方法執(zhí)行語句
建議使用第一種,更方便,第一種還可以在修改優(yōu)化為,將一些公共語句在摘出來使用。
import time import pymysql from DBUtils.PooledDB import PooledDB POOL = PooledDB( creator=pymysql, # 使用鏈接數(shù)據(jù)庫的模塊 maxconnections=6, # 連接池允許的大連接數(shù),0和None表示不限制連接數(shù) mincached=2, # 初始化時(shí),鏈接池中至少創(chuàng)建的空閑的鏈接,0表示不創(chuàng)建 maxcached=5, # 鏈接池中最多閑置的鏈接,0和None不限制 maxshared=3, # 鏈接池中最多共享的鏈接數(shù)量,0和None表示全部共享。PS: 無用,因?yàn)閜ymysql和MySQLdb等模塊的 threadsafety都為1,所有值無論設(shè)置為多少,_maxcached永遠(yuǎn)為0,所以永遠(yuǎn)是所有鏈接都共享。 blocking=True, # 連接池中如果沒有可用連接后,是否阻塞等待。True,等待;False,不等待然后報(bào)錯 maxusage=None, # 一個(gè)鏈接最多被重復(fù)使用的次數(shù),None表示無限制 setsession=[], # 開始會話前執(zhí)行的命令列表。如:["set datestyle to ...", "set time zone ..."] ping=0, # ping MySQL服務(wù)端,檢查是否服務(wù)可用。# 如: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='123', database='pooldb', charset='utf8' ) """ class SQLHelper(object): @staticmethod def fetch_one(sql,args): conn = POOL.connection() cursor = conn.cursor() cursor.execute(sql, args) result = cursor.fetchone() conn.close() return result @staticmethod def fetch_all(self,sql,args): conn = POOL.connection() cursor = conn.cursor() cursor.execute(sql, args) result = cursor.fetchone() conn.close() return result # 調(diào)用方式: result = SQLHelper.fetch_one('select * from xxx',[]) print(result) """ """ #第二種: class SQLHelper(object): def __init__(self): self.conn = POOL.connection() self.cursor = self.conn.cursor() def close(self): self.cursor.close() self.conn.close() def fetch_one(self,sql, args): self.cursor.execute(sql, args) result = self.cursor.fetchone() self.close() return result def fetch_all(self, sql, args): self.cursor.execute(sql, args) result = self.cursor.fetchall() self.close() return result obj = SQLHelper() obj.fetch_one() """
上述內(nèi)容就是怎么在Flask框架中使用DBUtils模塊連接數(shù)據(jù)庫,你們學(xué)到知識或技能了嗎?如果還想學(xué)到更多技能或者豐富自己的知識儲備,歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道。
分享標(biāo)題:怎么在Flask框架中使用DBUtils模塊連接數(shù)據(jù)庫-創(chuàng)新互聯(lián)
文章出自:http://www.muchs.cn/article38/dhjdpp.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供移動網(wǎng)站建設(shè)、Google、關(guān)鍵詞優(yōu)化、電子商務(wù)、網(wǎng)頁設(shè)計(jì)公司、面包屑導(dǎo)航
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請盡快告知,我們將會在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場,如需處理請聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時(shí)需注明來源: 創(chuàng)新互聯(lián)
猜你還喜歡下面的內(nèi)容