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  # 我这里MySql端口是3307，默认3306
        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)


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([1,‘1‘])

db.insertBatch(‘test‘,arrFiled,arrValueS)