漫漫長夜，最近python2轉(zhuǎn)python3之后，發(fā)現(xiàn)自己還沒有手寫過mysql連接的工具類，發(fā)現(xiàn)稍微改改之前python2的版本在python3就可用直接使用。下面就廢話少說，直接上代碼。
使用pip3安裝pymysql
pip3 install pymysql
工具類代碼
# -*- coding: utf-8 -*-
import pymysql
import re
class MysqldbHelper(object): # 繼承object類所有方法
'''
構(gòu)造方法：
config = {
'host': '127.0.0.1',
'port': 3306,
'user': 'root',
'passwd': 'root',
'charset':'utf8',
'cursorclass':pymysql.cursors.DictCursor
}
conn = pymysql.connect(**config)
conn.autocommit(1)
cursor = conn.cursor()
'''
def __init__(self , config):
self.host = config['host']
self.username = config['user']
self.password = config['passwd']
self.port = config['port']
self.con = None
self.cur = None
try:
self.con = pymysql.connect(**config)
self.con.autocommit(1)
# 所有的查詢，都在連接 con 的一個模塊 cursor 上面運行的
self.cur = self.con.cursor()
except:
print("DataBase connect error,please check the db config.")
# 關(guān)閉數(shù)據(jù)庫連接
def close(self):
if not self.con:
self.con.close()
else:
print("DataBase doesn't connect,close connectiong error;please check the db config.")
# 創(chuàng)建數(shù)據(jù)庫
def createDataBase(self,DB_NAME):
# 創(chuàng)建數(shù)據(jù)庫
self.cur.execute('CREATE DATABASE IF NOT EXISTS %s DEFAULT CHARACTER SET utf8 COLLATE utf8_general_ci' % DB_NAME)
self.con.select_db(DB_NAME)
print('creatDatabase:' + DB_NAME)
# 選擇數(shù)據(jù)庫
def selectDataBase(self,DB_NAME):
self.con.select_db(DB_NAME)
# 獲取數(shù)據(jù)庫版本號
def getVersion(self):
self.cur.execute("SELECT VERSION()")
return self.getOneData()
# 獲取上個查詢的結(jié)果
def getOneData(self):
# 取得上個查詢的結(jié)果，是單個結(jié)果
data = self.cur.fetchone()
return data
# 創(chuàng)建數(shù)據(jù)庫表
def creatTable(self, tablename, attrdict, constraint):
"""創(chuàng)建數(shù)據(jù)庫表
args：
tablename ：表名字
attrdict ：屬性鍵值對,{'book_name':'varchar(200) NOT NULL'...}
constraint ：主外鍵約束,PRIMARY KEY(`id`)
"""
if self.isExistTable(tablename):
print("%s is exit" % tablename)
return
sql = ''
sql_mid = '`id` bigint(11) NOT NULL AUTO_INCREMENT,'
for attr,value in attrdict.items():
sql_mid = sql_mid + '`'+attr + '`'+' '+ value+','
sql = sql + 'CREATE TABLE IF NOT EXISTS %s ('%tablename
sql = sql + sql_mid
sql = sql + constraint
sql = sql + ') ENGINE=InnoDB DEFAULT CHARSET=utf8'
print('creatTable:'+sql)
self.executeCommit(sql)
def executeSql(self,sql=''):
"""執(zhí)行sql語句，針對讀操作返回結(jié)果集
args：
sql ：sql語句
"""
try:
self.cur.execute(sql)
records = self.cur.fetchall()
return records
except pymysql.Error as e:
error = 'MySQL execute failed! ERROR (%s): %s' %(e.args[0],e.args[1])
print(error)
def executeCommit(self,sql=''):
"""執(zhí)行數(shù)據(jù)庫sql語句，針對更新,刪除,事務(wù)等操作失敗時回滾
"""
try:
self.cur.execute(sql)
self.con.commit()
except pymysql.Error as e:
self.con.rollback()
error = 'MySQL execute failed! ERROR (%s): %s' %(e.args[0],e.args[1])
print("error:", error)
return error
def insert(self, tablename, params):
"""創(chuàng)建數(shù)據(jù)庫表
args：
tablename ：表名字
key ：屬性鍵
value ：屬性值
"""
key = []
value = []
for tmpkey, tmpvalue in params.items():
key.append(tmpkey)
if isinstance(tmpvalue, str):
value.append("\'" + tmpvalue + "\'")
else:
value.append(tmpvalue)
attrs_sql = '('+','.join(key)+')'
values_sql = ' values('+','.join(value)+')'
sql = 'insert into %s'%tablename
sql = sql + attrs_sql + values_sql
print('_insert:'+sql)
self.executeCommit(sql)
def select(self, tablename, cond_dict='', order='', fields='*'):
"""查詢數(shù)據(jù)
args：
tablename ：表名字
cond_dict ：查詢條件
order ：排序條件
example：
print mydb.select(table)
print mydb.select(table, fields=["name"])
print mydb.select(table, fields=["name", "age"])
print mydb.select(table, fields=["age", "name"])
"""
consql = ' '
if cond_dict!='':
for k, v in cond_dict.items():
consql = consql+'`'+k +'`'+ '=' + '"'+v + '"' + ' and'
consql = consql + ' 1=1 '
if fields == "*":
sql = 'select * from %s where ' % tablename
else:
if isinstance(fields, list):
fields = ",".join(fields)
sql = 'select %s from %s where ' % (fields, tablename)
else:
print("fields input error, please input list fields.")
sql = sql + consql + order
print('select:' + sql)
return self.executeSql(sql)
def insertMany(self,table, attrs, values):
"""插入多條數(shù)據(jù)
args：
tablename ：表名字
attrs ：屬性鍵
values ：屬性值
example：
table='test_mysqldb'
key = ["id" ,"name", "age"]
value = [[101, "liuqiao", "25"], [102,"liuqiao1", "26"], [103 ,"liuqiao2", "27"], [104 ,"liuqiao3", "28"]]
mydb.insertMany(table, key, value)
"""
values_sql = ['%s' for v in attrs]
attrs_sql = '('+','.join(attrs)+')'
values_sql = ' values('+','.join(values_sql)+')'
sql = 'insert into %s'% table
sql = sql + attrs_sql + values_sql
print('insertMany:'+sql)
try:
print(sql)
for i in range(0,len(values),20000):
self.cur.executemany(sql,values[i:i+20000])
self.con.commit()
except pymysql.Error as e:
self.con.rollback()
error = 'insertMany executemany failed! ERROR (%s): %s' %(e.args[0],e.args[1])
print(error)
def delete(self, tablename, cond_dict):
"""刪除數(shù)據(jù)
args：
tablename ：表名字
cond_dict ：刪除條件字典
example：
params = {"name" : "caixinglong", "age" : "38"}
mydb.delete(table, params)
"""
consql = ' '
if cond_dict!='':
for k, v in cond_dict.items():
if isinstance(v, str):
v = "\'" + v + "\'"
consql = consql + tablename + "." + k + '=' + v + ' and '
consql = consql + ' 1=1 '
sql = "DELETE FROM %s where%s" % (tablename, consql)
print(sql)
return self.executeCommit(sql)
def update(self, tablename, attrs_dict, cond_dict):
"""更新數(shù)據(jù)
args：
tablename ：表名字
attrs_dict ：更新屬性鍵值對字典
cond_dict ：更新條件字典
example：
params = {"name" : "caixinglong", "age" : "38"}
cond_dict = {"name" : "liuqiao", "age" : "18"}
mydb.update(table, params, cond_dict)
"""
attrs_list = []
consql = ' '
for tmpkey, tmpvalue in attrs_dict.items():
attrs_list.append("`" + tmpkey + "`" + "=" +"\'" + tmpvalue + "\'")
attrs_sql = ",".join(attrs_list)
print("attrs_sql:", attrs_sql)
if cond_dict!='':
for k, v in cond_dict.items():
if isinstance(v, str):
v = "\'" + v + "\'"
consql = consql + "`" + tablename +"`." + "`" + k + "`" + '=' + v + ' and '
consql = consql + ' 1=1 '
sql = "UPDATE %s SET %s where%s" % (tablename, attrs_sql, consql)
print(sql)
return self.executeCommit(sql)
def dropTable(self, tablename):
"""刪除數(shù)據(jù)庫表
args：
tablename ：表名字
"""
sql = "DROP TABLE %s" % tablename
self.executeCommit(sql)
def deleteTable(self, tablename):
"""清空數(shù)據(jù)庫表
args：
tablename ：表名字
"""
sql = "DELETE FROM %s" % tablename
print("sql=",sql)
self.executeCommit(sql)
def isExistTable(self, tablename):
"""判斷數(shù)據(jù)表是否存在
args：
tablename ：表名字
Return:
存在返回True，不存在返回False
"""
sql = "select * from %s" % tablename
result = self.executeCommit(sql)
if result is None:
return True
else:
if re.search("doesn't exist", result):
return False
else:
return True
if __name__ == "__main__":
# 定義數(shù)據(jù)庫訪問參數(shù)
config = {
'host': '127.0.0.1',
'port': 6806,
'user': 'root',
'passwd': 'tcltest',
'charset': 'utf8',
'cursorclass': pymysql.cursors.DictCursor
}
# 初始化打開數(shù)據(jù)庫連接
mydb = MysqldbHelper(config)
# 打印數(shù)據(jù)庫版本
print(mydb.getVersion())
# 創(chuàng)建數(shù)據(jù)庫
DB_NAME = 'test_db'
# mydb.createDataBase(DB_NAME)
# 選擇數(shù)據(jù)庫
print("========= 選擇數(shù)據(jù)庫%s ===========" % DB_NAME)
mydb.selectDataBase(DB_NAME)
#創(chuàng)建表
TABLE_NAME = 'test_user'
print("========= 選擇數(shù)據(jù)表%s ===========" % TABLE_NAME)
# CREATE TABLE %s(id int(11) primary key,name varchar(30))' %TABLE_NAME
attrdict = {'name':'varchar(30) NOT NULL'}
constraint = "PRIMARY KEY(`id`)"
# mydb.creatTable(TABLE_NAME,attrdict,constraint)
# 插入紀(jì)錄
print("========= 單條數(shù)據(jù)插入 ===========")
params = {}
for i in range(5):
params.update({"name":"testuser"+str(i)}) # 生成字典數(shù)據(jù)，循環(huán)插入
print(params)
mydb.insert(TABLE_NAME, params)
print("")
# 批量插入數(shù)據(jù)
print("========= 多條數(shù)據(jù)同時插入 ===========")
insert_values = []
for i in range(5):
# values.append((i,"testuser"+str(i)))
insert_values.append([u"測試用戶"+str(i)]) # 插入中文數(shù)據(jù)
print(insert_values)
insert_attrs = ["name"]
mydb.insertMany(TABLE_NAME,insert_attrs, insert_values)
# 數(shù)據(jù)查詢
print("========= 數(shù)據(jù)查詢 ===========")
print(mydb.select(TABLE_NAME, fields=["id", "name"]))
print(mydb.select(TABLE_NAME, cond_dict = {'name':'測試用戶2'},fields=["id", "name"]))
print(mydb.select(TABLE_NAME, cond_dict = {'name':'測試用戶2'},fields=["id", "name"],order="order by id desc"))
# 刪除數(shù)據(jù)
print("========= 刪除數(shù)據(jù) ===========")
delete_params = {"name": "測試用戶2"}
mydb.delete(TABLE_NAME, delete_params)
# 更新數(shù)據(jù)
print("========= 更新數(shù)據(jù) ===========")
update_params = {"name": "測試用戶99"} # 需要更新為什么值
update_cond_dict = {"name": "測試用戶3"} # 更新執(zhí)行的查詢條件
mydb.update(TABLE_NAME, update_params, update_cond_dict)
# 刪除表數(shù)據(jù)
print("========= 刪除表數(shù)據(jù) ===========")
mydb.deleteTable(TABLE_NAME)
# 刪除表
print("========= 刪除表 ===========")
mydb.dropTable(TABLE_NAME)
