python操作mysql數(shù)據(jù)庫類

2019-11-08 01:01:10

字體：大中小

供稿：網(wǎng)友

#-*-coding:utf-8-*-############################ 功能：python操作MySQL 數(shù)據(jù)庫類，# 包括查詢、插入、修改、刪除記錄等# PRograme:pyMysql# programming environment：python3.4.4 && MySQL_5.6.15# by adengou@Foxmail.com 2017-02-20###########################import osimport mysql.connector#引入mysql連接驅(qū)動模塊import jsonimport time,datetime#-------------pyMysql數(shù)據(jù)模塊--------------class MySql_Model:####初始化變量 def __init__(self,username, passWord,hostname,database,port=3306,charset='utf8'): #初始化MySQL數(shù)據(jù)庫連接 self.hostname = hostname #服務(wù)器地址 self.username =username #數(shù)據(jù)庫用戶名 self.password = password #數(shù)據(jù)庫密碼 self.database = database #數(shù)據(jù)庫名稱 self.port =port #服務(wù)器端口 self.charset =charset#接收編碼 self.connector =mysql.connector#選擇mysql數(shù)據(jù)庫連接驅(qū)動 self.conn =None#初始化連接 self.cursor=None#初始化游標(biāo) self.bConnection =False#判數(shù)連接成功初始化####析構(gòu)函數(shù) def __del__(self): self.mySql_db_close() print("/r/n/t調(diào)用了MySql_Model析構(gòu)函數(shù)/r/n/t")####建立數(shù)據(jù)庫連接 def mySql_connect(self): try: #建立數(shù)據(jù)庫連接 self.conn = self.connector.connect(user=self.username, passwd=self.password,host=self.hostname,database=self.database,port=self.port,charset=self.charset) #設(shè)置游標(biāo) #self.cursor = self.conn.cursor(buffered=True,dictionary=True) self.cursor = self.conn.cursor() self.bConnection = True except mysql.connector.Error as err: if err.errno == errorcode.ER_access_DENIED_ERROR: print("Something is wrong with your user name or password") self.bConnection = False elif err.errno == errorcode.ER_BAD_DB_ERROR: print("Database does not exist") self.bConnection = False else: print(err) self.bConnection = False####關(guān)閉數(shù)據(jù)庫 def mySql_db_close(self,func=""): if(self.cursor != None and self.conn != None): self.cursor.close()#關(guān)閉游標(biāo) self.conn.close()#釋放數(shù)據(jù)庫資源 if(func!=""): print("/r/n/t"+func+"關(guān)閉了數(shù)據(jù)庫連接"+"/r/n/t") else: print("/r/n/t未建立數(shù)據(jù)庫連接/r/n") ####顯示MYSQL版本 def mySql_version(self): self.mySql_connect()#建立連接 self.cursor.execute ("SELECT VERSION()") row =self.cursor.fetchone () print ("MySQL server version:", row[0]) self.mySql_db_close("mySql_version") ####獲取數(shù)據(jù)庫所有記錄集 def mySql_db_getall(self,tablename,condition=u""): arrData =[] if(condition ==""or condition ==None ): sql ="select * from "+tablename+""; else: sql ="select * from "+tablename+" where 1=1 and "+condition ## self.mySql_connect()#建立連接 ## if(self.bConnection == True): self.cursor =self.conn.cursor(buffered=True,dictionary=True) #以緩存方式讀取，返回?cái)?shù)據(jù)為字典形式 try: self.cursor.execute(sql) arrData = self.cursor.fetchall() except(TypeError,ValueError) as e: print("error") print(str(e)) ## #print(arrData) self.mySql_db_close("mySql_db_getall")#關(guān)閉數(shù)據(jù)連接，釋放資源 return arrData ####獲取數(shù)據(jù)庫一條記錄集 def mySql_db_getone(self,tablename,condition=u""): arrData =[] if(condition ==""or condition ==None ): sql ="select * from "+tablename+""; else: sql ="select * from "+tablename+" where 1=1 and "+condition ## self.mySql_connect()#建立連接 ## if(self.bConnection == True): self.cursor =self.conn.cursor(buffered=True,dictionary=True) #以緩存方式讀取，返回?cái)?shù)據(jù)為字典形式 try: self.cursor.execute(sql) arrData = self.cursor.fetchone() ''' while arrData is not None: print(arrData) arrData = self.cursor.fetchone() ''' except(TypeError,ValueError) as e: print(str(e)) except( mysql.connector.errors.InternalError) as e: print("mySql_db_getone_error: "+str(e)) except: print("mySql_db_getone_error: shomethin is wrong ") ## #print(arrData) self.mySql_db_close("mySql_db_getone")#關(guān)閉數(shù)據(jù)連接，釋放資源 return arrData#插入無條件 def mySql_db_insert(self,tablename,dataArray): bInsert=False field = "" valueCode = "" if(isinstance(dataArray,list) or len(dataArray)<=0): print('沒有要插入的數(shù)據(jù)') bInsert=False for key in dataArray: field = field +key+"," valueCode = valueCode+"'"+dataArray[key]+"'," field = field[0:-1] valueCode = valueCode[0:-1] # self.mySql_connect()#建立連接 ## if(self.bConnection == True): self.cursor =self.conn.cursor(buffered=True,dictionary=True) #以緩存方式讀取，返回?cái)?shù)據(jù)為字典形式 try: sql = "insert into "+tablename+"("+field+") values("+valueCode+")" #self.cursor.execute("set names 'gbk'")#這步很重要，防止中文出現(xiàn)亂碼，數(shù)據(jù)庫編碼是中文 self.cursor.execute("SET NAMES 'UTF8'")#數(shù)據(jù)庫編碼是UTF-8 self.cursor.execute(sql) self.conn.commit() bInsert=True except( mysql.connector.errors.InternalError) as e: print("mySql_db_insert_error: "+str(e)) bInsert = False self.mySql_db_close("mySql_db_insert")#關(guān)閉數(shù)據(jù)連接，釋放資源 return bInsert #####有條件插入 def mysSql_db_insert_condiction(self,tablename,dictData,fieldname,strInput): # insertKey ='' insertValue ='' # bInsert = False #判斷是否插入成功 i=1 for key in dictData.keys(): if(i<len(dictData.keys())): insertValue =(insertValue +'"%s"'+',')%(dictData[key]) insertKey =(insertKey +"%s"+',')%(key) else: insertValue =(insertValue +'"%s"')%(dictData[key]) insertKey =(insertKey +"%s")%(key) # i=i+1 ## #print("insertKey:"+insertKey) #print("insertValue :"+insertValue ) # self.mySql_connect()#建立連接 ## if(self.bConnection == True): self.cursor =self.conn.cursor(buffered=True,dictionary=True) #以緩存方式讀取，返回?cái)?shù)據(jù)為字典形式 try: sql="INSERT INTO "+tablename+"("+insertKey+") SELECT "+insertValue +" FROM dual WHERE not exists (select * from "+tablename+" where instr("+fieldname+",'"+strInput+"')>0)" #self.cursor.execute("set names 'gbk'")#這步很重要，防止中文出現(xiàn)亂碼，數(shù)據(jù)庫編碼是中文 self.cursor.execute("SET NAMES 'UTF8'")#數(shù)據(jù)庫編碼是UTF-8 self.cursor.execute(sql) self.conn.commit() bInsert = True except( mysql.connector.errors.InternalError) as e: print("mysSql_db_insert_condiction_error: "+str(e)) bInsert = False self.mySql_db_close("mysSql_db_insert_condiction")#關(guān)閉數(shù)據(jù)連接，釋放資源 return bInsert ####增加記錄 def mySql_db_add(self,tablename,dictData): #方法同mySql_db_insert函數(shù) insertKey ='' insertValue ='' # bInsert = False #判斷是否插入成功 i=1 for key in dictData.keys(): if(i<len(dictData.keys())): insertValue =(insertValue +'"%s"'+',')%(dictData[key]) insertKey =(insertKey +"%s"+',')%(key) else: insertValue =(insertValue +'"%s"')%(dictData[key]) insertKey =(insertKey +"%s")%(key) # i=i+1 ## #print("insertKey:"+insertKey) #print("insertValue :"+insertValue ) # self.mySql_connect()#建立連接 ## if(self.bConnection == True): self.cursor =self.conn.cursor(buffered=True,dictionary=True) #以緩存方式讀取，返回?cái)?shù)據(jù)為字典形式 try: sql ="insert into "+tablename+"("+insertKey+") values("+insertValue +")" #self.cursor.execute("set names 'gbk'")#這步很重要，防止中文出現(xiàn)亂碼，數(shù)據(jù)庫編碼是中文 self.cursor.execute("SET NAMES 'UTF8'")#數(shù)據(jù)庫編碼是UTF-8 self.cursor.execute(sql) self.conn.commit() bInsert = True except( mysql.connector.errors.InternalError) as e: print("mySql_db_add_error: "+str(e)) bInsert = False self.mySql_db_close("mySql_db_add")#關(guān)閉數(shù)據(jù)連接，釋放資源 return bInsert####批量增加記錄 def mySql_db_add2(self,tablename,str_arrFieldname,json_arrValue): bInsert = False #判斷是否插入成功 arrKey =[] valueCode ="" arrKey = str_arrFieldname.split(",") j=1 for key in arrKey: if(j<len(arrKey)): valueCode =valueCode+"%s," else: valueCode =valueCode+"%s" j=j+1 # self.mySql_connect()#建立連接 ## if(self.bConnection == True): self.cursor =self.conn.cursor(buffered=True,dictionary=True) #以緩存方式讀取，返回?cái)?shù)據(jù)為字典形式 try: sql ="insert into "+tablename+"("+str_arrFieldname+") values("+valueCode +")" #self.cursor.execute("set names 'gbk'")#這步很重要，防止中文出現(xiàn)亂碼，數(shù)據(jù)庫編碼是中文 self.cursor.execute("SET NAMES 'UTF8'")#數(shù)據(jù)庫編碼是UTF-8 self.cursor.executemany(sql,json_arrValue) self.conn.commit() bInsert = True except( mysql.connector.errors.InternalError) as e: print("mySql_db_add2_error: "+str(e)) bInsert = False self.mySql_db_close("mySql_db_add2")#關(guān)閉數(shù)據(jù)連接，釋放資源 return bInsert#############################修改記錄 def mySql_db_update(self,tablename,dictData,condition=""): bUpdate = False #判斷更新是否成功 if(condition==""):return bUpdate # setStr="" i=1 for key in dictData.keys(): if(i<len(dictData.keys())): setStr =setStr+key+"='"+dictData[key]+"'," else: setStr =setStr+key+"='"+dictData[key]+"'" # i=i+1 # self.mySql_connect()#建立連接 ## if(self.bConnection == True): self.cursor =self.conn.cursor(buffered=True,dictionary=True) #以緩存方式讀取，返回?cái)?shù)據(jù)為字典形式 try: sql ="update "+tablename+ " set "+setStr+" where 1=1 and "+condition #self.cursor.execute("set names 'gbk'")#這步很重要，防止中文出現(xiàn)亂碼，數(shù)據(jù)庫編碼是中文 self.cursor.execute("SET NAMES 'UTF8'")#數(shù)據(jù)庫編碼是UTF-8 self.cursor.execute(sql) self.conn.commit() bUpdate = True except( mysql.connector.errors.InternalError) as e: self.conn.rollback()#更新不成功回滾 print("mySql_db_update_error: "+str(e)) bUpdate = False self.mySql_db_close("mySql_db_update")#關(guān)閉數(shù)據(jù)連接，釋放資源 return bUpdate####刪除記錄 def mySql_db_del(self,tablename,condition=""): bDel= False #判斷刪除是否成功 if(condition==""):return bDel # self.mySql_connect()#建立連接 ## if(self.bConnection == True): self.cursor =self.conn.cursor(buffered=True,dictionary=True) #以緩存方式讀取，返回?cái)?shù)據(jù)為字典形式 try: sql ="DELETE from "+tablename+ " where 1=1 and "+condition self.cursor.execute(sql) self.conn.commit() bDel = True except( mysql.connector.errors.InternalError) as e: print("mySql_db_del_error: "+str(e)) bDel = False self.mySql_db_close("mySql_db_del")#關(guān)閉數(shù)據(jù)連接，釋放資源 return bDel ####獲取數(shù)據(jù)庫記錄總數(shù) def mySql_db_recordcount(self,tablename,condition=u""): count =0 if(condition ==""or condition ==None ): sql ="select * from "+tablename+""; else: sql ="select * from "+tablename+" where 1=1 and "+condition ## self.mySql_connect()#建立連接 # if(self.bConnection == True): self.cursor =self.conn.cursor(buffered=True,dictionary=True) try: self.cursor.execute(sql) arrData = self.cursor.fetchall() count =len(arrData) except(TypeError,ValueError) as e: print("mySql_db_recordcount_error:"+str(e)) self.mySql_db_close("mySql_db_recordcount")#關(guān)閉數(shù)據(jù)連接，釋放資源 return count####重置ID字段，重新從1開始排序 def mySql_db_resetid(self,tablename): bId=False#判斷id字段是否存在 ## self.mySql_connect()#建立連接 # if(self.bConnection == True): self.cursor =self.conn.cursor(buffered=True,dictionary=True) try: sql ="select * from "+tablename self.cursor.execute(sql) self.cursor.fetchone() for fieldname in self.cursor.description: if(fieldname[0]=='id'): bId=True break if(bId==True): #1，刪除原有主鍵： sql="ALTER TABLE "+tablename+" DROP id" self.cursor.execute(sql) self.conn.commit() #2，添加新主鍵字段： sql="ALTER TABLE "+tablename+" ADD id int(11) NOT NULL FIRST" self.cursor.execute(sql) self.conn.commit() #3，設(shè)置新主鍵： sql="ALTER TABLE "+tablename+" MODIFY COLUMN id int(11) NOT NULL AUTO_INCREMENT,ADD PRIMARY KEY(id)" self.cursor.execute(sql) self.conn.commit() except(TypeError,ValueError) as e: bId=False print("mySql_db_resetid_error:"+str(e)) self.mySql_db_close("mySql_db_resetid")#關(guān)閉數(shù)據(jù)連接，釋放資源 return bId ####獲取數(shù)據(jù)庫最后一條id def mySql_db_insertId(self,tablename): insert_id =0 ## self.mySql_connect()#建立連接 # if(self.bConnection == True): self.cursor =self.conn.cursor(buffered=True,dictionary=True) try: sql ="select * from "+tablename+" order by id desc" self.cursor.execute(sql) arrData = self.cursor.fetchone() insert_id =arrData["id"] except(TypeError,ValueError) as e: print("error") return insert_id ####示例def Forexample():#測試 try: mysql=MySql_Model('root','','127.0.0.1','mydataBase',3306,'utf8') mysql.mySql_version() data=mysql.mySql_db_getone("cb_countbook","cbcreater='lhl' order by id desc") print(data) # dictData ={'cbdate': '2017-02-25', 'cbitem': '月供/房租','cbexpense': '123456.30', 'cbcreater': 'lhl', 'cbaddtime': '2017-2-12', 'cbmoditime': '2017-2-12'} bInsert=mysql.mySql_db_insert("cb_countbook",dictData) print("插入:"+str(bInsert)) # # dictData ={'cbdate': '2017-02-25', 'cbitem': '月供/房租','cbexpense': '123456.30', 'cbcreater': 'lhl', 'cbaddtime': '2017-2-12', 'cbmoditime': '2017-2-12'} bInsert=mysql.mySql_db_add("cb_countbook",dictData) print("插入:"+str(bInsert)) # # dictDataCondiction ={'cbdate': '2017-02-25', 'cbitem': '月供/房租','cbexpense': '123456.30', 'cbcreater': 'odg', 'cbaddtime': '2017-2-12', 'cbmoditime': '2017-2-12'} bInsert=mysql.mysSql_db_insert_condiction("cb_countbook",dictDataCondiction,'cbcreater',dictDataCondiction['cbcreater']) print("'條件插入:"+str(bInsert)) # arrFiedname ="cbitem,cbexpense,cbcreater" arrValues =[('月供/房租','545896','odghlfdfdf')] bInsert=mysql.mySql_db_add2("cb_countbook",arrFiedname,arrValues) print("/t/r/n插入:"+str(bInsert)) # dictData ={'cbdate': '2017-02-25', 'cbitem': '月供/房租','cbexpense': '99999.30', 'cbcreater': 'fuck'} condiction ="cbcreater='odghlfdfdf'" bUpdate = mysql.mySql_db_update("cb_countbook",dictData,condiction) print("/t/r/n更新:"+str(bUpdate)) # condiction ="cbcreater='fuck'" bDel =mysql.mySql_db_del("cb_countbook",condiction ) print("/t/r/n刪除:"+str(bDel)) # count =mysql.mySql_db_recordcount("cb_countbook") print("/t/r/n記錄數(shù):"+str(count)) # mysql.mySql_db_resetid("cb_countbook") insert_id =mysql.mySql_db_insertId("cb_countbook") print("/t/r/n最后ID:"+str(insert_id)) # mysql =None except (TypeError,ValueError) as e: #將異常對象輸出 print("error:"+str(e)) except: print("未知錯(cuò)誤") #主程序入口if __name__=='__main__': Forexample()

上一篇：python裝飾器、with和contextmanager

下一篇：理解Python中列表，元組，字典，集合的區(qū)別