elab_mvp/mysql_db.py

import pymysql as ps


class MysqlDB:
    """
        mysql操作
    """
    con = None
    cursor = None

    def __init__(self, db_name, db_type=None):
        self.db_name = db_name
        if db_type is None:
            self.con = ps.connect(host='172.19.189.136', port=3306, user='bi_etl', password='XPtpswuU5lwGo4kx',
                                  db=self.db_name, charset='utf8')
        else:
            self.con = ps.connect(host='106.15.201.221', port=3306, user='root', password='elab@123'
                                  , db=self.db_name, charset='utf8')
        self.cursor = self.con.cursor()

    def show_tables(self):
        self.cursor.execute('show tables')
        for talbe in self.cursor.fetchall():
            print(talbe)

    def select(self, sql, params=None):
        if params:
            self.cursor.execute(sql, params)
        else:
            self.cursor.execute(sql)
        return self.cursor.fetchall()

    def add_some(self, sql, data):
        try:
            self.cursor.executemany(sql, data)
            self.con.commit()
        except:
            print('数据插入异常...')
            self.con.rollback()

    def add_one(self, sql, data):
        try:
            self.cursor.execute(sql, data)
            self.con.commit()
        except:
            self.con.rollback()

    def truncate(self, table_name):
        sql = 'truncate table {}'.format(table_name)
        self.cursor.execute(sql)
        self.con.commit()

    def close(self):
        self.cursor.close()
        self.con.close()


if __name__ == '__main__':
    sql = 'select *from bq_testcase limit 10'
    si = {
        1: 1
    }
    print(si.get('1'))