python数据库处理

1、测试python对数据库增删改查并打印日志，使用本地服务器，数据库使用mysql。

代码如下处理数据库模块：

class OperateSql:

    def connect_db(self):
        logger.info('===================开始连接数据库！===================')
        db = pymysql.connect(
            host='127.0.0.1',
            user='python',
            password='ab123456',
            db='python'
        )
        logger.info('===================连接数据库成功！===================')
        cursor = db.cursor()
        return cursor, db

    def create_table(self):
        cursor, db = self.connect_db()
        drop_sql = 'drop table if exists xxxa;'
        create_sql = """create table xxxa(
            id int auto_increment primary key,
            sno varchar(50),
            name varchar(50),
            sex bit(1),
            city varchar(100),
            time datetime);"""
        try:
            logger.info('===================判断表是否存在===================')
            cursor.execute(drop_sql)
            logger.info('===================开始创建表===================')
            cursor.execute(create_sql)
            logger.info('===================创建表成功===================')
        except Exception as e:
            logger.info(f'建表失败，失败原因为：{e}')
            raise

    def insert_sql(self, num, sno):
        cursor, db = self.connect_db()
        city = ['China', 'USA', 'Japan', 'Canada', 'France']
        date_l = [datetime.strftime(x, '%Y-%m-%d') for x in list(pd.date_range(start='11/1/2018', end='11/8/2018'))]
        try:
            for i in range(1, num + 1):
                v_sno = str(sno) + str(i)
                v_name = 'hk_' + str(rdn(1, 100))
                v_sex = str(rdn(0, 1))
                v_city = str(choice(city))
                v_time = str(choice(date_l))
                tmp = "('" + v_sno + "','" + v_name + "'," + v_sex + ",'" + v_city + "','" + v_time + "');"
                print(tmp)
                insert_sql = """insert into xxxa(sno,name,sex,city,time) values""" + tmp
                logger.info(f'开始插入第{i}条数据: {insert_sql}')
                cursor.execute(insert_sql)
                db.commit()
                logger.info(f'===================插入第{i}条数据成功！===================')
            logger.info(f'===================插入数据成功！===================')
        except Exception as e:
            db.rollback()
            logger.info('数据插入失败！！！！！')

    def select_sql(self, n):
        cursor, db = self.connect_db()
        sql_select = """select city from xxxa where id = """ + str(n) + ";"
        print(sql_select)
        try:
            cursor.execute(sql_select)
            row = cursor.fetchone()
            print(row[0])
            logger.info(f'查询成功，第{n}条记录的学生city为：{row[0]}')
            return row[0]
        except Exception as e:
            logger.info(f'查询失败，未查到第{n}条记录的学生city！！！')

    def update_sql(self, n):
        cursor, db = self.connect_db()
        update_sql = """update xxxa set city = concat(city,'浦东') where id =""" + str(n) + ";"
        try:
            old_val = self.select_sql(n)
            logger.info(f'更新前第{n}条记录的学生city为:{old_val}')
            cursor.execute(update_sql)
            db.commit()
            new_val = self.select_sql(n)
            logger.info(f'更新后第{n}条记录的学生city为:{new_val}')
            assert new_val == old_val + '浦东'
            logger.info(f'第{n}条记录的学生city更新成功！city为:{new_val}')
        except Exception as e:
            db.rollback()
            logger.info(f'第{n}条记录的学生city更新失败！！！！')
            raise

    def delete_table(self, m):
        cursor, db = self.connect_db()
        delete_sql = """delete from xxxa where id = """ + str(m) + ";"
        try:
            old_val = self.select_sql(m)
            logger.info(f'删除前第{m}条记录的学生city为:{old_val}')
            cursor.execute(delete_sql)
            db.commit()
            new_val = self.select_sql(m)
            logger.info(f'删除后第{m}条记录的学生city为:{new_val}')
            assert new_val == None
            logger.info(f'删除第{m}条的学生信息成功！！！')
        except Exception as e:
            db.rollback()
            logger.info(f'删除第{m}条的学生信息失败！！！！！')
            raise
        finally:
            db.close()

日志打印模块如下：

class Log:
    def __init__(self, log_path):
        self.log_file = os.path.join(log_path, f'{time.strftime("%Y-%m-%d")}.log')

    def __printconsole(self, level, message):
        logger = logging.getLogger()
        logger.setLevel(logging.DEBUG)
        fh = logging.FileHandler(self.log_file, 'a+', encoding='utf-8')
        fh.setLevel(logging.DEBUG)
        ch = logging.StreamHandler()
        ch.setLevel(logging.DEBUG)
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        fh.setFormatter(formatter)
        ch.setFormatter(formatter)
        logger.addHandler(fh)
        logger.addHandler(ch)

        if level == 'info':
            logging.info(message)
        if level == 'debug':
            logging.info(message)
        if level == 'warning':
            logging.info(message)
        if level == 'error':
            logging.info(message)
        logger.removeHandler(ch)
        logger.removeHandler(fh)

        fh.close()
        return logger

    def debug(self, message):
        self.__printconsole('debug', message)

    def info(self, message):
        self.__printconsole('info', message)

    def warning(self, message):
        self.__printconsole('warning', message)

    def error(self, message):
        self.__printconsole('error', message)

日志结果如下：