使用python进行数据迁移重组之mysql工具类

àì夳堔傛蜴生んèń 2022-07-14 15:32 186阅读 0赞

#coding:utf-8
    '''
    这里暂时以mysql为主(可以对此类进行任意扩展)
    
    使用方式:
    print db.queryResults('表名',field=['id','字段二','字段三'])
    insert = {'字段二':字段值,'字段三':字段三值}
    print db.update('表名',insert,where='id=8')
    print db.update('表名',insert,where='id=8',addfield='num=num+1')
    
    '''
    import MySQLdb
    import logging
    import warnings
    warnings.filterwarnings("ignore")
    
    CLIENT_LONG_PASSWORD = 1
    CLIENT_FOUND_ROWS = 1 << 1
    CLIENT_LONG_FLAG = 1 << 2
    CLIENT_CONNECT_WITH_DB = 1 << 3
    CLIENT_NO_SCHEMA = 1 << 4
    CLIENT_COMPRESS = 1 << 5
    CLIENT_ODBC = 1 << 6
    CLIENT_LOCAL_FILES = 1 << 7
    CLIENT_IGNORE_SPACE = 1 << 8
    CLIENT_PROTOCOL_41 = 1 << 9
    CLIENT_INTERACTIVE = 1 << 10
    CLIENT_SSL = 1 << 11
    CLIENT_IGNORE_SIGPIPE = 1 << 12
    CLIENT_TRANSACTIONS  = 1 << 13
    CLIENT_SECURE_CONNECTION = 1 << 15
    CLIENT_MULTI_STATEMENTS = 1 << 16
    CLIENT_MULTI_RESULTS = 1 << 17
    CLIENT_CAPABILITIES = CLIENT_LONG_PASSWORD|CLIENT_LONG_FLAG|CLIENT_TRANSACTIONS| \
                            CLIENT_PROTOCOL_41|CLIENT_SECURE_CONNECTION
    
    
    logger = logging.getLogger('django')
    
    class mydb:
        cur = None
        conn = None
    
        def _init_(self,**agrs):
            conv = MySQLdb.converters.conversions.copy()
            conv[12] = str #convert datetime to string in MySQLdb.converters.constans.FIELD_TYPE
            self.conn = MySQLdb.connect(host=agrs['host'],port=agrs['port'],db=agrs['db'],\
                                  user=agrs['user'],passwd=agrs['pwd'],\
                                  charset='utf8',conv=conv,\
                                  client_flag=CLIENT_MULTI_STATEMENTS|CLIENT_MULTI_RESULTS
                                  )
            self.cur = self.conn.cursor()
            self.cur.execute('SET NAMES utf8')
    
        def __init__(self,**agrs):
            #conv = MySQLdb.converters.conversions.copy()
            #conv[12] = str #convert datetime to string in MySQLdb.converters.constans.FIELD_TYPE
            self.conn = MySQLdb.connect(host=agrs['host'],port=agrs['port'],db=agrs['db'],\
                                  user=agrs['user'],passwd=agrs['pwd'],\
                                  charset='utf8',\
                                  client_flag=CLIENT_MULTI_STATEMENTS|CLIENT_MULTI_RESULTS
                                  )
            self.cur = self.conn.cursor()
            self.cur.execute('SET NAMES utf8')
            
    
        #普通完整查寻
        def query(self,sql):
    #        print("query: %s "%sql.encode('utf-8'))
            res = self.cur.execute(sql)
            res = self.cur.fetchall()
            return res
    
        #获取一个结果
        def fetchone(self,sql):
            res = self.cur.execute(sql)
            res = self.cur.fetchone()
            return res
        
        #获取查寻结果，如果没有指定field会出错，因此如果是获取所有字段就必须显式的全部指定
        def queryResults(self,table,**agrs):           
            field = ','.join(['%s'%(x) for x in agrs['field']]) if 'field' in agrs else '*'
            where = 'where %s'%(agrs['where']) if 'where' in agrs else ''
            groupby = 'group by %s'%(agrs['groupby']) if 'groupby' in agrs else ''
            order = 'order by %s'%(agrs['order']) if 'order' in agrs else ''
            limit = 'limit %s'%(agrs['limit']) if 'limit' in agrs else ''
            sql = 'select %s from %s %s %s %s %s'%(field,table,where,groupby,order,limit)
            #logger.debug("queryResults:%s", %sql.encode('utf-8'))
    #        print "queryResults:%s"%(sql)
    
            self.cur.execute(sql)
            res = []
            #这里处理成字典形式，方便直接用字段访问值
            '''
             实际上这里还可以这样(以字段形式返回)：
                    cur = self.conn.cursor(MySQLdb.cursors.DictCursor)
                    cur.execute(sql)
                    print cur.fetchall()  
            '''
            for line in self.cur.fetchall():
                getline = {}
                for i,v in enumerate(line):
                    getline[agrs['field'][i]] = v
                res.append(getline)
    
            return res
    
        #插入
        '''
            data为一个字典对象:{'xxx':'xx','xxx':'xxx'}
        '''
        def insert(self,table,data,**args):
            field = ','.join(['`%s`'%(x) for x in data.keys()])
            #print field
            values = []
            for x in data.values():
                if x==None:
                    y='NULL'
                    values.append(y)
                else:
                    if isinstance(x,str) or isinstance(x,unicode):
                        x="'"+x+"'"
                    else:
                        x="'"+str(x)+"'"
                    values.append(x)
            values = ','.join(["%s"%(x) for x in values]).replace('\\','')
            #print values
            #values = ','.join(["'%s'"%(x) for x in data.values()])
            sql='insert into `%s`(%s) values(%s)'%(table,field,values)
            #print sql
            #print("insert: %s "%sql.encode('utf-8'))
            self.cur.execute(sql)
            
            willcommit = args['commit'] if 'commit' in args else True
            if willcommit:  
                self.conn.commit()
            ids = int(self.getInsertID())
            return  ids
    
        #更新操作
        def update(self,table,data,**agrs):
            '''
                data为字典对象，更新的字段是非数据库增加值，
                如果有 field=field+1此类字段(addfield)，请写在关键字参数里
                addfield:field=field+1
            '''
            field = ','.join(["`%s`='%s'"%(k,data[k]) for k in data])
            field += ',%s'%(agrs['addfield']) if 'addfield' in agrs else ''
            where = 'where %s'%(agrs['where']) if 'where' in agrs else ''
            limit = 'limit %s'%(agrs['limit']) if 'limit' in agrs else 'limit 1'
            sql='update `%s` set %s %s %s'%(table,field,where,limit)
            #print sql
            logger.debug("update:%s", sql)
            res = self.cur.execute(sql)
            self.conn.commit()
            return  res
        
         #更新操作
        def updateBysql(self,sql):
            logger.debug("update:%s", sql)
            res = self.cur.execute(sql)
            self.conn.commit()
            return  res
    
        #获取刚刚插入的ID
        def getInsertID(self):
            sql = 'select last_insert_id() as id'
            res = self.fetchone(sql)[0]
            return res 
    
        #删除
        def delete(self,table,**agrs):
            where = 'where %s'%(agrs['where']) if 'where' in agrs else ''
            limit = 'limit %s'%(agrs['limit']) if 'limit' in agrs else ''
            willcommit = agrs['commit'] if 'commit' in agrs else True
            sql = 'delete from `%s` %s %s'%(table,where,limit)
            logger.debug("delete sql:%s, willcommit:%s", sql, willcommit)
            res = self.cur.execute(sql)
            if willcommit:
                self.conn.commit()
            return res
    
        #替换单引号
        def qutoSin(self,string):
            if string:
                return string.replace("'",'"')
            return string
    
        #获取数据库下面的所有表名
        def showTables(self):
            r = []
            res = []
            self.cur.execute("show tables")
            r = self.cur.fetchall()
            for (table_name,) in r:
                res.append(table_name)
            return res
    
        #获取查询到的总数
        def getCount(self,table,**agrs):
            where = 'where %s'%(agrs['where']) if 'where' in agrs else ''
            sql = 'select count(*) from %s %s'%(table,where)
            logger.debug("getCount sql:%s", sql)
            
            self.cur.execute(sql)
            count = self.cur.fetchone()[0]
            return count
    
        def commit(self):
            self.conn.commit()
            
        def rollback(self):
            self.conn.rollback()
            
    
        #释放资源
        def __del__(self):
            if self.cur:
                self.cur.close()
                self.cur = None
            if self.conn:
                self.conn.close()
                self.conn = None
    
        #获取查寻结果(何净)
        def querySql(self,sql,**agrs):
            logger.debug("queryResults:%s", sql)
    #        print("queryResults:%s"%sql)
            #self.cur.execute(sql)
            self.cur.execute(sql)
            res = []
            #这里处理成字典形式，方便直接用字段访问值
            '''
             实际上这里还可以这样(以字段形式返回)：
                    cur = self.conn.cursor(MySQLdb.cursors.DictCursor)
                    cur.execute(sql)
                    print cur.fetchall()  
            '''
           # for line in self.cur.fetchall():
            for line in self.cur.fetchall():
                getline = {}
                for i,v in enumerate(line):
                    getline[agrs['field'][i]] = v
                res.append(getline)
        
            return res
    
        def close(self):
            if self.cur:
                self.cur.close()
                self.cur = None
            if self.conn:
                self.conn.close()
                self.conn = None
    
    #实例化
    #db = db(host='xxxx',db='xxxx',user='xxxx',pwd='xxxx')