工具类

import pymysql
from logaid import log
import timeclass MySQLDB:def __init__(self, host, port, user, passwd: str, db, charset: str):self.host = hostself.port = portself.user = userself.passwd = passwdself.db = dbself.charset = charsetself.cursor = Noneself.connection = Nonedef connect(self):"""连接到 MySQL 数据库"""try:self.connection = pymysql.Connect(host=self.host,port=self.port,user=self.user,passwd=self.passwd,db=self.db,charset=self.charset)self.cursor = self.connection.cursor()return Trueexcept Exception as e:time.sleep(6)log.error(f'Connect mysql is failed. error{e}')return Falsedef close(self):"""关闭数据库连接"""if self.cursor:self.cursor.close()if self.connection:self.connection.close()def execute(self, query, params=None) -> bool:"""原生 sql 执行语句"""self.connect()try:if self.cursor:self.cursor.execute(query, params)self.connection.commit()return Trueexcept Exception as e:self.connection.rollback()log.error(f'Execute is failed. error:{e}. query:{query}')finally:self.close()return Falsedef query(self, sql):"""查询，返回结果:param sql: SQL 查询语句:return: 查询结果"""if self.execute(sql):return self.cursor.fetchall()return Falsedef insert(self, sql, value):"""插入，返回记录 ID:param sql: SQL 插入语句:param value: 插入语句的参数:return: bool|int"""if self.execute(sql, value):insert_id = self.cursor.lastrowidreturn insert_idreturn Falsedef update(self, sql, value=None) -> bool:"""更新:param sql: SQL 更新语句:param value: 更新语句的参数:return: bool"""return self.execute(sql, value)def exist(self, sql) -> bool:"""是否存在:param sql: sql 查询语句:return: bool"""if self.execute(sql):return self.cursor.fetchone()[0] > 0return Trueif __name__ == '__main__':a = MySQLDB('127.0.0.1',3306,'admin','admin','abc','utf8mb4')sql = f"select count(*) from login where source='{1}';"sql_result = a.exist(sql)print(sql_result)