| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102 | # Author renyupeng# coding=utf-8# @Time    : 2021/9/15 11:06 上午# @Site    :# @File    : TidbProConn.py# @Software: PyCharm# @contact: renyupeng@c-top.com.cnimport loggingimport timefrom sqlite3 import OperationalErrorimport pymysqlfrom Apietl.constant.ConfConstant import ConfConstantclass TidbProConn:    """    Tidb连接管理    """    @staticmethod    def get_connection():        """        获取Tidb连接        :return: 连接        """        return pymysql.connect(host=ConfConstant.TIDB_PRO_HOST, user=ConfConstant.TIDB_PRO_USER,                               passwd=ConfConstant.TIDB_PRO_PASSWORD,                               port=ConfConstant.TIDB_PRO_PORT, charset="utf8")    @classmethod    def _reConn(cls, num=28800, stime=5):  # 重试连接总次数为1天,这里根据实际情况自己设置,如果服务器宕机1天都没发现就......        _number = 0        _status = True        while _status and _number <= num:            try:                TidbProConn.get_connection().ping()  # cping 校验连接是否异常                _status = False            except:                if TidbProConn.get_connection() == True:  # 重新连接,成功退出                    _status = False                    break                _number += 1                time.sleep(stime)  # 连接不成功,休眠3秒钟,继续循环,知道成功或重试次数结束    @classmethod    def quary(cls, conn, sql):        TidbProConn._reCon(conn)        cur = conn.cursor()        cur.execute(sql)        result = cur.fetchall()        return result    @staticmethod    def conn_close(conn: pymysql.connections.Connection):        """        关闭资源        """        if conn:            try:                conn.close()            except pymysql.err.Error:                pass    @staticmethod    def exec_sql(conn, sql_buff):        sql_list = sql_buff.split(";")        cur = conn.cursor()        for sql in sql_list:            if sql.strip():                cur.execute(sql)        return cur    @staticmethod    def upsert(conn, sql_buff):        TidbProConn._reCon(conn)        try:            TidbProConn.exec_sql(conn, sql_buff)        except Exception as e:            logging.error('失败详细信息:', repr(e))        TidbProConn.get_connection().commit()    @staticmethod    def _reCon(conn):        """ MySQLdb.OperationalError异常"""        # self.con.close()        while True:            try:                conn.ping()                break            except OperationalError:                conn.ping(True)if __name__ == '__main__':    db = TidbProConn.get_connection()    cursor = db.cursor()    # 执行sql语句    sql = "show databases"    cursor.execute(sql)    result = cursor.fetchall()    print("result====", result)
 |