elab_mvp/db/mysql_db.py

import pymysql as ps


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

    def __init__(self, db_name):
        self.db_name = db_name
        self.con = ps.connect(host='192.168.0.13', 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__':
    mysql_db = MysqlDB('marketing_db')
    # sql = 'select * from bq_question a left join bq_option b on a.id = b.question_id where a.status = 1 and b.status = 1'
    sql = 'select city from f_t_daren_score_2 group by city'
    result = mysql_db.select(sql)
    for rt in result:
        print(rt)