打卡工具——数据库部分

Posted 2021-03-24 cts2710

背景

?上一章介绍了打卡工具的ui部分的代码，今天来接着介绍数据库部分的。其实这个工具涉及到数据库的内容也很简单，只有查询表，创建表，初始化表，读取数据表，写入数据表，删除表这些操作。

技术概览

1. sqlalchemy
2. pymysql

?这里解释下为什么要用到两个数据库的模块，因为我读取数据表的时候想让读取的数据是DataFrame类型的，所以我用了pd.read_sql()方法（而且也很简单方便啊！），用sqlalchemy做连接读取内容是没什么问题，但是在查表的时候一直报错（现在想想可能是我方法用错了），所以我所以我又用到pymysql（稍微更熟悉一些），但是to_sql函数只支持两类mysql引擎一个是sqlalchemy，另一个是sqlliet3，pymysql是不能用的。

同步数据库

?这里的同步分为自动和手动两种，打开工具的时候会自动从默认数据表读取内容显示出来，在工具上打卡完再手动同步记录到数据库。一开始打卡工具在写手动同步数据库的时候也想说要怎么实现才好，写了很多判断条件，比如判断数据库里的记录是否合法（日期唯一，格式正确），数据库和当前ui的表格记录有没有重复等等，但是这样写就不实用了（可能就是懒？），后来考虑到自己使用方便，就以ui表格记录为标准，每次手工同步数据库就删除表，重新把ui表格记录写进去，简单粗暴！！！

代码

 def auto_sync_mysql(self):
        '''
        自动同步数据库
        :return:
        '''
        try:
        #获取QLineEdit上的数据库用户名和密码
            user = self.user_edit.text()
            psw = self.psw_edit.text()
            #创建数据库连接
            cli = 'mysql+pymysql://' + user + ':' + psw + '@localhost:3306/' + self.db
            self.engine = create_engine(cli)
            connect = pymysql.connect(
                user=user,
                password=psw,
                db=self.db,
                host='127.0.0.1',
                port=3306,
                charset='utf8'
            )
            con = connect.cursor()
            sql = "show tables;"
            con.execute(sql)
            tables = [con.fetchall()]
            table_list = re.findall('('.*?')', str(tables))
            table_list = [re.sub("'", '', each) for each in table_list]
            #判断有没有打卡记录表，没有则新建
            if self.table not in table_list:
                self.init_db()
            else:
                self.sql = 'select * from ' + self.table + ';'
                #读表内容用DataFrame类型保存
                df = pd.read_sql(sql=self.sql, con=self.engine)
                #表不为空的情况下
                if not df.empty:
                    #检查数据库中是否有重复的日期
                    if df.date.duplicated().any():
                    #弹出提示框
                        self.errorTipIns.setupUi(tip='导入数据库失败', logInfo='请检查数据库中日期的唯一性！')
                        self.errorTipIns.show()
                        return
                    else:
                        for i in range(len(df)):
                            #检查格式是否正确
                            result1 = re.match('dddd-dd-dd$', str(df.date[i]))
                            result2 = re.match('[0-6]$', str(df.week[i]))
                            result3 = re.match('d{2}:dd:dd$', str(df.work_time[i]).split()[-1])
                            result_list = [result1, result2, result3]
                            if None in result_list:
                                self.errorTipIns.setupUi(tip='导入失败', logInfo='日期/星期/时间的格式不正确')
                                self.errorTipIns.show()
                                return
                            else:
                            #增加table的行数，写入记录
                                self.tableWidget.setRowCount(self.row + 1)
                                self.tableWidget.setItem(self.row, 0, QTableWidgetItem(str(df.date[i])))
                                self.tableWidget.setItem(self.row, 1, QTableWidgetItem(str(df.week[i])))
                                work_time = str(df.work_time[i]).split()[-1]
                                self.tableWidget.setItem(self.row, 2, QTableWidgetItem(work_time))
                                self.tableWidget.sortItems(0, Qt.AscendingOrder)
                                self.row += 1
        except Exception as e:
            print(e)
            self.errorTipIns.setupUi(tip='自动导入数据失败', logInfo='请检查数据库的用户名密码是否正确！')
            self.errorTipIns.show()
        finally:
            con.close()
            connect.close()

    def click_mysql_button(self):
        '''
        点击同步数据库
        :return:
        '''
        try:
            if self.row == 0:
                self.errorTipIns.setupUi(tip='同步失败', logInfo='当前表格没有记录需要同步到数据库！')
                self.errorTipIns.show()
                return
            else:
            #先删表再建表
                self.drop_db()
                self.init_db()
                #用到列表解析式获取table当前的记录
                date_list = [self.tableWidget.item(row, 0).text() for row in range(self.tableWidget.rowCount())]
                week_list = [self.tableWidget.item(row, 1).text() for row in range(self.tableWidget.rowCount())]
                time_list = [self.tableWidget.item(row, 2).text() for row in range(self.tableWidget.rowCount())]
                all_df = pd.DataFrame({'date':date_list,'week':week_list,'work_time':time_list})
                #写入数据库的表
                all_df.to_sql(self.table, con=self.engine, index=False, if_exists='replace')
                self.successTipIns.setupUi(tip='同步完成', logInfo='数据库和当前表格同步记录成功！')
                self.successTipIns.show()
        except Exception as e:
            print(e)
            self.errorTipIns.setupUi(tip='导入失败', logInfo='请检查用户名密码是否正确！')
            self.errorTipIns.show()

    def init_db(self): #初始化表
        Base.metadata.create_all(self.engine)
        print('创建数据库表')

    def drop_db(self):  #删除表
        Base.metadata.drop_all(self.engine)
        print('删除数据库表')

Base = declarative_base()  #创建Base
class clockinRecords(Base):            #创建表（一个类一张表）
#先获取数据库要创建的表名称
    iniString = ''
    try:
        with open('./config.ini', 'rb') as lines:
            for line in lines:
                iniString += str(line)
        table = re.findall('table:(w+);', iniString)[-1]

        __tablename__ = table   #表名
        date = Column(Date,nullable= False,primary_key = True) #日期型，不为空，主键
        week = Column(Integer,nullable= False) #整型，不为空
        work_time = Column(String(32),nullable= False)   #字符串（32），不为空

        __table_args__ = (
            UniqueConstraint('date',name = 'date'),    #唯一索引
        )
    except Exception as err:
        print('clockinRecords: error is ', err)

总结

?以上就是我打卡工具数据库部分的介绍，是不是很简单哈哈。如果有不对或者更好的方法，希望大家指正，谢谢~