python操作mysql数据库类

2019-11-08 01:01:10

字体：大中小

来源：转载

供稿：网友

#-*-coding:utf-8-*-############################ 功能：python操作MySQL 数据库类，# 包括查询、插入、修改、删除记录等# PRograme:pyMysql# programming environment：python3.4.4 && MySQL_5.6.15# by adengou@Foxmail.com 2017-02-20###########################import osimport mysql.connector#引入mysql连接驱动模块import jsonimport time,datetime#-------------pyMysql数据模块--------------class MySql_Model:####初始化变量 def __init__(self,username, passWord,hostname,database,port=3306,charset='utf8'): #初始化MySQL数据库连接 self.hostname = hostname #服务器地址 self.username =username #数据库用户名 self.password = password #数据库密码 self.database = database #数据库名称 self.port =port #服务器端口 self.charset =charset#接收编码 self.connector =mysql.connector#选择mysql数据库连接驱动 self.conn =None#初始化连接 self.cursor=None#初始化游标 self.bConnection =False#判数连接成功初始化####析构函数 def __del__(self): self.mySql_db_close() print("/r/n/t调用了MySql_Model析构函数/r/n/t")####建立数据库连接 def mySql_connect(self): try: #建立数据库连接 self.conn = self.connector.connect(user=self.username, passwd=self.password,host=self.hostname,database=self.database,port=self.port,charset=self.charset) #设置游标 #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####关闭数据库 def mySql_db_close(self,func=""): if(self.cursor != None and self.conn != None): self.cursor.close()#关闭游标 self.conn.close()#释放数据库资源 if(func!=""): print("/r/n/t"+func+"关闭了数据库连接"+"/r/n/t") else: print("/r/n/t未建立数据库连接/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") ####获取数据库所有记录集 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) #以缓存方式读取，返回数据为字典形式 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")#关闭数据连接，释放资源 return arrData ####获取数据库一条记录集 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) #以缓存方式读取，返回数据为字典形式 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")#关闭数据连接，释放资源 return arrData#插入无条件 def mySql_db_insert(self,tablename,dataArray): bInsert=False field = "" valueCode = "" if(isinstance(dataArray,list) or len(dataArray)<=0): print('没有要插入的数据') 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) #以缓存方式读取，返回数据为字典形式 try: sql = "insert into "+tablename+"("+field+") values("+valueCode+")" #self.cursor.execute("set names 'gbk'")#这步很重要，防止中文出现乱码，数据库编码是中文 self.cursor.execute("SET NAMES 'UTF8'")#数据库编码是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")#关闭数据连接，释放资源 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) #以缓存方式读取，返回数据为字典形式 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'")#这步很重要，防止中文出现乱码，数据库编码是中文 self.cursor.execute("SET NAMES 'UTF8'")#数据库编码是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")#关闭数据连接，释放资源 return bInsert ####增加记录 def mySql_db_add(self,tablename,dictData): #方法同mySql_db_insert函数 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) #以缓存方式读取，返回数据为字典形式 try: sql ="insert into "+tablename+"("+insertKey+") values("+insertValue +")" #self.cursor.execute("set names 'gbk'")#这步很重要，防止中文出现乱码，数据库编码是中文 self.cursor.execute("SET NAMES 'UTF8'")#数据库编码是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")#关闭数据连接，释放资源 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) #以缓存方式读取，返回数据为字典形式 try: sql ="insert into "+tablename+"("+str_arrFieldname+") values("+valueCode +")" #self.cursor.execute("set names 'gbk'")#这步很重要，防止中文出现乱码，数据库编码是中文 self.cursor.execute("SET NAMES 'UTF8'")#数据库编码是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")#关闭数据连接，释放资源 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) #以缓存方式读取，返回数据为字典形式 try: sql ="update "+tablename+ " set "+setStr+" where 1=1 and "+condition #self.cursor.execute("set names 'gbk'")#这步很重要，防止中文出现乱码，数据库编码是中文 self.cursor.execute("SET NAMES 'UTF8'")#数据库编码是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")#关闭数据连接，释放资源 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) #以缓存方式读取，返回数据为字典形式 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")#关闭数据连接，释放资源 return bDel ####获取数据库记录总数 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")#关闭数据连接，释放资源 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，设置新主键： 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")#关闭数据连接，释放资源 return bId ####获取数据库最后一条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记录数:"+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("未知错误") #主程序入口if __name__=='__main__': Forexample()

上一篇：python装饰器、with和contextmanager

下一篇：理解Python中列表，元组，字典，集合的区别