Python MySQL 操作类（有使用例子）

# -*- coding:utf-8 -*- import MySQLdb import time ''' · MySQL 操作类 · V1.0 ''' class MySQLClass(object): def __init__(self, host, user, password, charset="utf8"): super(MySQLClass, self).__init__() self.host = host self.user = user self.password = password self.port = 3307 self.charset = charset try: self.conn = MySQLdb.connect(host=self.host, user=self.user, passwd=self.password, port=self.port) self.conn.set_character_set(self.charset) self.cur = self.conn.cursor() except MySQLdb.Error as e: self.printError(e, '') def getTime(self): return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time())) def printError(self, e, sql): print("{0} MySQl Error {1}: {2}".format(self.getTime(), e.args[0], e.args[1])) print(sql) def setdb(self, db): try: self.conn.select_db(db) except MySQLdb.Error as e: self.printError(e, '') # 新增 def insert(self, tablename, DicData): try: arrParam = [] arrValue = [] for key in DicData: arrParam.append('%s') arrValue.append(DicData[key]) strfield = '`{0}`'.format('`,`'.join(DicData.keys())) strParam = 'values({0})'.format(','.join(arrParam)) sql = 'insert into {0}({1}) {2};'.format(tablename, strfield, strParam) self.cur.execute(sql, tuple(arrValue)) # 单调新增 self.conn.commit() except MySQLdb.Error as e: self.printError(e, sql) # self.conn.close() # 批量新增 def insertBatch(self, tablename, arrFiled, arrTupleValue): try: strfield = '`{0}`'.format('`,`'.join(arrFiled)) arrParam = [] for x in arrFiled: arrParam.append('%s') strParam = 'values({0})'.format(','.join(arrParam)) sql = 'insert into {0}({1}) {2};'.format(tablename, strfield, strParam) self.cur.executemany(sql, tuple(arrTupleValue)) # many 多条新增 self.conn.commit() except MySQLdb.Error as e: self.printError(e, '') # self.conn.close() # 更新语句 def update(self, sql): try: self.cur.execute(sql) self.conn.commit() except MySQLdb.Error as e: self.conn.rollback() self.printError(e, sql) # self.conn.close() # 查询单个字段 返回List def executeList(self, sql): try: result = [] self.cur.execute(sql) ds = self.cur.fetchall() for i, row in enumerate(ds): result.append(str(row[0])) return result except MySQLdb.Error as e: self.printError(e, sql) def executeDataRow(self, sql): try: self.cur.execute(sql) ds = self.cur.fetchall() return ds except MySQLdb.Error as e: self.printError(e, sql) # db=MySQLClass('127.0.0.1','root','123456') # db.setdb('test') # arr=db.executeList('SELECT DISTINCT skuid from commodity') # for x in arr: # print(x) # table=db.insert('commodity',{'title':'10'}) # db.conn.commit() # arrTr={'title':'1'} # if 'titl1e' in arrTr: # print (1) # else: # print (0) # arrFiled = ['name', 'age'] # arrValue=['1','2'] # arrValueS=[] # arrValueS.append(tuple([1,'1']) ) # arrValueS.append(tuple(['2',2]) ) # arrValueS.append(tuple([3,3]) ) # arrValueS.append(arrValue) # tuple1=(('2','3'),) # db.insertBatch('test',arrFiled,arrValueS)