| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091 | # Author renyupeng# coding=utf-8# @Time    : 2021/9/15 11:06 上午# @Site    : # @File    : TidbConn.py# @Software: PyCharm# @contact: renyupeng@c-top.com.cnimport loggingimport timeimport pymysqlfrom Apietl.constant.ConfConstant import ConfConstantclass TidbConn:    """    Tidb连接管理    """    @staticmethod    def get_connection():        """        获取Tidb连接        :return: 连接        """        return pymysql.connect(host=ConfConstant.TIDB_HOST, user=ConfConstant.TIDB_USER,                               passwd=ConfConstant.TIDB_PASSWORD,                               port=ConfConstant.TIDB_PORT, charset="utf8")    @classmethod    def _reConn(cls, num=28800, stime=3):  # 重试连接总次数为1天,这里根据实际情况自己设置,如果服务器宕机1天都没发现就......        _number = 0        _status = True        while _status and _number <= num:            try:                TidbConn.get_connection().ping()  # cping 校验连接是否异常                _status = False            except:                if TidbConn.get_connection():  # 重新连接,成功退出                    _status = False                    break                _number += 1                time.sleep(stime)  # 连接不成功,休眠3秒钟,继续循环,知道成功或重试次数结束    @classmethod    def quary(cls, sql):        TidbConn._reConn(num=28800, stime=3)        mysql_conn = TidbConn.get_connection()        cur = mysql_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(sql):        TidbConn._reConn(num=28800, stime=3)        try:            TidbConn.exec_sql(TidbConn.get_connection(), sql)        except Exception as e:            logging.error('失败详细信息:', repr(e))        TidbConn.get_connection().commit()if __name__ == '__main__':    db = TidbConn.get_connection()    cursor = db.cursor()    # 执行sql语句    sql = "show databases"    cursor.execute(sql)    result = cursor.fetchall()    print("result====", result)
 |