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)
日志结果如下：
image.png
