1、准备Excel数据:
2、读取Excel数据 :read_excel.py
import openpyxl from log import * def excel_to_list(file_name,sheet_name): data_list=[] #打开Excel logging.info("打开Excel") wb=openpyxl.load_workbook(file_name) #获取工作簿 logging.info("打开worksheet") sh=wb[sheet_name] #获取第一行既标题行的值 logging.info("读取标题行") row1=(list(sh.rows))[0] header=[i.value for i in row1] #获取每一行的值 max_row=len(list(sh.rows)) for i in range(1,max_row): row_i_value=[j.value for j in (list(sh.rows))[i]] #将标题和每行数据组装成字典 d=dict(zip(header,row_i_value)) data_list.append(d) logging.info("成功读取EXCEL所有数据") return data_list#每个元素是一个字典 if __name__=="__main__": file_name="my_db.xlsx" sheet_name="sh" data_list=excel_to_list(file_name,sheet_name) print(data_list)
3、构造日志模块:log.py
import logging logging.basicConfig(level=logging.DEBUG, # log level format=‘[%(asctime)s] %(levelname)s [%(funcName)s: %(filename)s, %(lineno)d] %(message)s‘, # log格式 datefmt=‘%Y-%m-%d %H:%M:%S‘, # 日期格式 filename=‘log.txt‘, # 日志输出文件 filemode=‘a‘) # 追加模式 if __name__ == ‘__main__‘: logging.info("info") logging.debug("debug") logging.error("error") logging.warning("warning") logging.critical("critical")
4、读取mysql配置文件:read_db_config.py
import configparser import os class ReadConfig: """定义一个读取配置文件的类""" def __init__(self, filepath=None): if filepath: configpath = filepath else: base_path = os.path.dirname(__file__) configpath = base_path + "/db_config.ini" self.cf = configparser.ConfigParser() self.cf.read(configpath) def get_db(self, param): value = self.cf.get("mysqlconf", param) return value if __name__ == ‘__main__‘: test = ReadConfig() t = test.get_db("host") print(t)
5、Python操作数据库:my_sql.py
from pymysql import connect, cursors from pymysql.err import OperationalError from read_db_config import * from log import * from read_excel import * logging.info("读取数据库配置文件") config=ReadConfig() host=config.get_db("host") logging.info("读取host_ip :{}".format(host)) port=config.get_db("port") logging.info("读取host_port :{}".format(port)) db=config.get_db("db_name") logging.info("读取db_name :{}".format(db)) user=config.get_db("user") logging.info("读取user_name :{}".format(user)) passwd=config.get_db("password") logging.info("读取password :{}".format(passwd)) class DB: def __init__(self): try: self.connection = connect(host=host, port=int(port), db = db, user = user, password = passwd, charset = ‘utf8‘, cursorclass=cursors.DictCursor) except OperationalError as e: self.connection.rollback() logging.error("Mysql Error %d: %s" % (e.args[0], e.args[1])) self.cur = self.connection.cursor() def insert(self, table_name, data): logging.info("start to insert data") for key in data: data[key] = "‘"+str(data[key])+"‘" key = ‘,‘.join(data.keys()) value = ‘,‘.join(data.values()) sql = "INSERT INTO " + table_name + " (" + key + ") VALUES (" + value + ")" logging.debug(sql) self.cur.execute(sql) self.connection.commit() def clear(self,table_name): logging.info("start to clear data") sql = "delete from " + table_name + ";" self.cur.execute("SET FOREIGN_KEY_CHECKS=0;") logging.debug(sql) self.cur.execute(sql) self.connection.commit() def query(self,sql): logging.info("start to query data") self.cur.execute(sql) return self.cur.fetchall() def check_data(self,table_name,name): logging.info("start to check data") sql="select * from {} where name=‘{}‘".format(table_name,name) logging.debug(sql) try: res=DB.query(self,sql) return True if res else False except Exception as e: logging.error(str(e)) def delete(self,tabel_name,name): logging.info("start to delete data") sql="delete from {} where name=‘{}‘".format(table_name,name) logging.debug(sql) self.cur.execute(sql) def close(self): logging.info("close db") self.connection.close() if __name__=="__main__": db=DB() table_name="my_tbl" name="aa" print(db.check_data(table_name,name))
6、test_case: test_excel_to_db.py
import unittest from read_excel import * from my_sql import DB from log import * class Test(unittest.TestCase): @classmethod def setUpClass(cls): file_name="my_db.xlsx" sheet_name="sh" cls.data_list=excel_to_list(file_name,sheet_name) #构造name列表 cls.s=[] for i in cls.data_list: for k in i.keys(): if k=="name": cls.s.append({k:i[k]}) #数据库插入数据 db=DB() table_name="my_tbl" db.clear(table_name) for d in cls.data_list: db.insert(table_name,d) def test_check_insert(self): params=self.s for names in params: name = names["name"] db=DB() res=db.check_data("my_tbl",name) logging.info("name :{}查询结果{}".format(name,res)) self.assertTrue(res) @classmethod def tearDownClass(cls): db=DB() db.close() if __name__=="__main__": unittest.main(verbosity=2)