本篇文章為大家展示了怎么在Flask框架中使用DBUtils模塊連接數(shù)據(jù)庫，內(nèi)容簡明扼要并且容易理解，絕對能使你眼前一亮，通過這篇文章的詳細介紹希望你能有所收獲。

成都創(chuàng)新互聯(lián)公司是一家專注于網(wǎng)站設(shè)計、成都網(wǎng)站制作與策劃設(shè)計,伍家崗網(wǎng)站建設(shè)哪家好?成都創(chuàng)新互聯(lián)公司做網(wǎng)站,專注于網(wǎng)站建設(shè)十余年,網(wǎng)設(shè)計領(lǐng)域的專業(yè)建站公司;建站業(yè)務(wù)涵蓋:伍家崗等地區(qū)。伍家崗做網(wǎng)站價格咨詢:18982081108

Flask連接數(shù)據(jù)庫

數(shù)據(jù)庫連接池：

Django使用:django ORM(pymysql/MySqldb)

Flask/其他使用：

    -原生SQL
        -pymysql（支持python2/3）
        -MySqldb（支持python2）
    -SQLAchemy（ORM）

原生SQL

需要解決的問題：

        -不能為每個用戶創(chuàng)建一個連接
        -創(chuàng)建一定數(shù)量的連接池，如果有人來

使用DBUtils模塊

兩種使用模式：

    1 為每個線程創(chuàng)建一個連接，連接不可控，需要控制線程數(shù)
    2 創(chuàng)建指定數(shù)量的連接在連接池，當(dāng)線程訪問的時候去取，如果不夠了線程排隊，直到有人釋放。平時建議使用這種?。?！

模式一：

import pymysql
from DBUtils.PersistentDB import PersistentDB
POOL = PersistentDB(
  creator=pymysql, # 使用鏈接數(shù)據(jù)庫的模塊
  maxusage=None, # 一個鏈接最多被重復(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時， conn.close() 實際上被忽略，供下次使用，再線程關(guān)閉時，才會自動關(guān)閉鏈接。如果為True時， conn.close()則關(guān)閉鏈接，那么再次調(diào)用pool.connection時就會報錯，因為已經(jīng)真的關(guān)閉了連接（pool.steady_connection()可以獲取一個新的鏈接）
  threadlocal=None, # 本線程獨享值得對象，用于保存鏈接對象，如果鏈接對象被重置
  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, # 初始化時，鏈接池中至少創(chuàng)建的空閑的鏈接，0表示不創(chuàng)建
  maxcached=5, # 鏈接池中最多閑置的鏈接，0和None不限制
  maxshared=3, # 鏈接池中最多共享的鏈接數(shù)量，0和None表示全部共享。PS: 無用，因為pymysql和MySQLdb等模塊的 threadsafety都為1，所有值無論設(shè)置為多少，_maxcached永遠為0，所以永遠是所有鏈接都共享。
  blocking=True, # 連接池中如果沒有可用連接后，是否阻塞等待。True，等待；False，不等待然后報錯
  maxusage=None, # 一個鏈接最多被重復(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)前正在運行連接數(shù)的是否小于大鏈接數(shù)，如果不小于則：等待或報raise TooManyConnections異常
  # 否則
  # 則優(yōu)先去初始化時創(chuàng)建的鏈接中獲取鏈接 SteadyDBConnection。
  # 然后將SteadyDBConnection對象封裝到PooledDedicatedDBConnection中并返回。
  # 如果最開始創(chuàng)建的鏈接沒有鏈接，則去創(chuàng)建一個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ù)庫

第二種是通過實例化對象，通過對象來調(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, # 初始化時，鏈接池中至少創(chuàng)建的空閑的鏈接，0表示不創(chuàng)建
  maxcached=5, # 鏈接池中最多閑置的鏈接，0和None不限制
  maxshared=3, # 鏈接池中最多共享的鏈接數(shù)量，0和None表示全部共享。PS: 無用，因為pymysql和MySQLdb等模塊的 threadsafety都為1，所有值無論設(shè)置為多少，_maxcached永遠為0，所以永遠是所有鏈接都共享。
  blocking=True, # 連接池中如果沒有可用連接后，是否阻塞等待。True，等待；False，不等待然后報錯
  maxusage=None, # 一個鏈接最多被重復(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è)資訊頻道。

網(wǎng)頁標(biāo)題：怎么在Flask框架中使用DBUtils模塊連接數(shù)據(jù)庫-創(chuàng)新互聯(lián)
本文來源：http://weahome.cn/article/dhjdpp.html