寫在前面
之前，以前分享過兩篇python操作mysql數(shù)據(jù)庫的文章：
《python操作mysql數(shù)據(jù)庫的精美實用模塊》
《python操作mysql數(shù)據(jù)庫的精美實用模塊（升級版）》
好了，今天繼續(xù)升級，二次升級。
為何又升級呢？緣起是，在實際項目開發(fā)中，當(dāng)前的模塊都是基于列表來操作，當(dāng)面對復(fù)雜表格時就會很不方便，嚴(yán)重影響開發(fā)效率和開發(fā)體驗。
所以，非常有必要，將當(dāng)前的模塊，從基于列表轉(zhuǎn)為基于字典，來進行操作。
然后，網(wǎng)上資料一頓查，終于成功升級了。來吧，不啰嗦，分享給大家！
python操作mysql數(shù)據(jù)庫的精美實用模塊—— DBUtils連接池版
import pymysql
from dbutils.pooled_db import PooledDB
#############################################################################################
# 連接mysql數(shù)據(jù)庫-----使用數(shù)據(jù)庫連接池技術(shù)----dbutils庫
#############################################################################################
db_pool = PooledDB(
creator=pymysql, # 使用鏈接數(shù)據(jù)庫的模塊
maxconnections=800, # 連接池允許的最大連接數(shù)，0和None表示不限制連接數(shù)
mincached=8, # 初始化時，鏈接池中至少創(chuàng)建的空閑的鏈接，0表示不創(chuàng)建
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='root',
database='mydbdemo',
charset='utf8'
)
class DBManager():
# 創(chuàng)建數(shù)據(jù)庫連接
def __init__(self):
self.db = db_pool.connection()
# 關(guān)閉數(shù)據(jù)庫連接
def closeDB(self):
self.db.close()
#連接數(shù)據(jù)庫，執(zhí)行插入SQl
def executeQueryID(self,sqlstring):
try:
cursor = self.db.cursor()
cursor.execute(sqlstring)
self.db.commit()
the_id = int(cursor.lastrowid)
#the_id = int(db.insert_id())
return the_id
except Exception:
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
# 連接數(shù)據(jù)庫，執(zhí)行SQl，無返回
def executeQueryNO(self, sqlstring):
try:
cursor = self.db.cursor()
cursor.execute(sqlstring)
self.db.commit()
return True
except Exception:
self.db.rollback()
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
#連接數(shù)據(jù)庫,執(zhí)行SQL，返回單條數(shù)據(jù)
def executeQueryone(self, sqlstring):
try:
cursor = self.db.cursor()
cursor.execute(sqlstring)
dataone = cursor.fetchone()
return dataone
except Exception:
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
#連接數(shù)據(jù)庫,執(zhí)行SQL，返回多條數(shù)據(jù)
def executeQueryall(self, sqlstring):
try:
cursor = self.db.cursor()
cursor.execute(sqlstring)
dataall = cursor.fetchall()
return dataall
except Exception:
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
#######二次升級版###############################################
# 查詢單條數(shù)據(jù)
def select_dataone_dict(self,sqlstring):
try:
cursor = self.db.cursor(cursor = DictCursor)
cursor.execute(sqlstring)
dataone = cursor.fetchone()
return dataone
except Exception:
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
# 查詢前n行數(shù)據(jù)
def select_datamany_dict(self,sqlstring,manyncount):
try:
cursor = self.db.cursor(cursor = DictCursor)
cursor.execute(sqlstring)
datamany = cursor.fetchmany(manyncount)
return datamany
except Exception:
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
# 查詢所有數(shù)據(jù)
def select_dataall_dict(self,sqlstring):
try:
cursor = self.db.cursor(cursor = DictCursor)
cursor.execute(sqlstring)
dataall = cursor.fetchall()
return dataall
except Exception:
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
# 插入單條數(shù)據(jù)
def insert_dataone_dict(self,table,dataone):
try:
sqlstring = self.generate_sqlstring_insert(table, dataone)
cursor = self.db.cursor()
cursor.execute(sqlstring,dataone)
self.db.commit()
the_id = int(cursor.lastrowid)
return the_id
except Exception:
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
# 插入多條數(shù)據(jù)
def insert_datamany_dict(self,table,datamany):
try:
sqlstring = self.generate_sqlstring_insert(table, datamany[0])
cursor = self.db.cursor()
cursor.executemany(sqlstring,datamany)
self.db.commit()
return True
except Exception:
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
# 更新數(shù)據(jù)
def update_data_dict(self,table,data,where):
try:
sqlstring = self.generate_sqlstring_update(table, data,where)
cursor = self.db.cursor()
cursor.execute(sqlstring,data)
self.db.commit()
return True
except Exception:
self.db.rollback()
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
# 刪除數(shù)據(jù)
def delete_data_dict(self,table,where):
try:
sqlstring = self.generate_sqlstring_delete(table, where)
cursor = self.db.cursor()
cursor.execute(sqlstring)
self.db.commit()
return True
except Exception:
self.db.rollback()
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
# 生成sql語句-----插入
def generate_sqlstring_insert(self, table, data):
cols_name = ", ".join(f'`{key}`' for key in data.keys())
cols_value = ', '.join(f'%({key})s' for key in data.keys())
sqlstring = f"INSERT INTO {table} ({cols_name}) VALUES ({cols_value})"
return sqlstring
# 生成sql語句----更新
def generate_sqlstring_update(self, table, data,where):
cols = ", ".join(f'`{key}`=%({key})s' for key in data.keys())
sqlstring = f"UPDATE {table} SET {cols} {where}"
return sqlstring
# 生成sql語句----刪除
def generate_sqlstring_delete(self, table ,where):
sqlstring = f"DELETE FROM {table} {where}"
return sqlstring
python操作mysql數(shù)據(jù)庫的精美實用模塊——非連接池版
##############################################################################################
# 連接mysql數(shù)據(jù)庫
#############################################################################################
class DBManager_Old():
# 創(chuàng)建數(shù)據(jù)庫連接
def __init__(self):
db_config = ConfigManager.get_db_config()
self.db = pymysql.Connect(
host='127.0.0.1',
port=3306,
user='root',
passwd='root',
db='mydbdemo',
charset='utf8')
# 關(guān)閉數(shù)據(jù)庫連接
def closeDB(self):
self.db.close()
#連接數(shù)據(jù)庫，執(zhí)行插入SQl
def executeQueryID(self,sqlstring):
try:
cursor = self.db.cursor()
cursor.execute(sqlstring)
self.db.commit()
the_id = int(cursor.lastrowid)
#the_id = int(db.insert_id())
return the_id
except Exception:
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
# 連接數(shù)據(jù)庫，執(zhí)行SQl，無返回
def executeQueryNO(self, sqlstring):
try:
cursor = self.db.cursor()
cursor.execute(sqlstring)
self.db.commit()
return True
except Exception:
self.db.rollback()
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
#連接數(shù)據(jù)庫,執(zhí)行SQL，返回單條數(shù)據(jù)
def executeQueryone(self, sqlstring):
try:
cursor = self.db.cursor()
cursor.execute(sqlstring)
dataone = cursor.fetchone()
return dataone
except Exception:
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
#連接數(shù)據(jù)庫,執(zhí)行SQL，返回多條數(shù)據(jù)
def executeQueryall(self, sqlstring):
try:
cursor = self.db.cursor()
cursor.execute(sqlstring)
dataall = cursor.fetchall()
return dataall
except Exception:
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
#######二次升級版###############################################
# 查詢單條數(shù)據(jù)
def select_dataone_dict(self,sqlstring):
try:
cursor = self.db.cursor(cursor = DictCursor)
cursor.execute(sqlstring)
dataone = cursor.fetchone()
return dataone
except Exception:
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
# 查詢前n行數(shù)據(jù)
def select_datamany_dict(self,sqlstring,manyncount):
try:
cursor = self.db.cursor(cursor = DictCursor)
cursor.execute(sqlstring)
datamany = cursor.fetchmany(manyncount)
return datamany
except Exception:
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
# 查詢所有數(shù)據(jù)
def select_dataall_dict(self,sqlstring):
try:
cursor = self.db.cursor(cursor = DictCursor)
cursor.execute(sqlstring)
dataall = cursor.fetchall()
return dataall
except Exception:
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
# 插入單條數(shù)據(jù)
def insert_dataone_dict(self,table,dataone):
try:
sqlstring = self.generate_sqlstring_insert(table, dataone)
cursor = self.db.cursor()
cursor.execute(sqlstring,dataone)
self.db.commit()
the_id = int(cursor.lastrowid)
return the_id
except Exception:
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
# 插入多條數(shù)據(jù)
def insert_datamany_dict(self,table,datamany):
try:
sqlstring = self.generate_sqlstring_insert(table, datamany[0])
cursor = self.db.cursor()
cursor.executemany(sqlstring,datamany)
self.db.commit()
return True
except Exception:
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
# 更新數(shù)據(jù)
def update_data_dict(self,table,data,where):
try:
sqlstring = self.generate_sqlstring_update(table, data,where)
cursor = self.db.cursor()
cursor.execute(sqlstring,data)
self.db.commit()
return True
except Exception:
self.db.rollback()
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
# 刪除數(shù)據(jù)
def delete_data_dict(self,table,where):
try:
sqlstring = self.generate_sqlstring_delete(table, where)
cursor = self.db.cursor()
cursor.execute(sqlstring)
self.db.commit()
return True
except Exception:
self.db.rollback()
return False
finally:
self.db.close() # 關(guān)閉數(shù)據(jù)庫連接
# 生成sql語句-----插入
def generate_sqlstring_insert(self, table, data):
cols_name = ", ".join(f'`{key}`' for key in data.keys())
cols_value = ', '.join(f'%({key})s' for key in data.keys())
sqlstring = f"INSERT INTO {table} ({cols_name}) VALUES ({cols_value})"
return sqlstring
# 生成sql語句----更新
def generate_sqlstring_update(self, table, data,where):
cols = ", ".join(f'`{key}`=%({key})s' for key in data.keys())
sqlstring = f"UPDATE {table} SET {cols} {where}"
return sqlstring
# 生成sql語句----刪除
def generate_sqlstring_delete(self, table ,where):
sqlstring = f"DELETE FROM {table} {where}"
return sqlstring
