# Author renyupeng # coding=utf-8 # @Time : 2021/9/15 11:06 上午 # @Site : # @File : TidbProConn.py # @Software: PyCharm # @contact: renyupeng@c-top.com.cn import logging import time from sqlite3 import OperationalError import pymysql from Apietl.constant.ConfConstant import ConfConstant class 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)