123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118 |
- # Author renyupeng
- # coding=utf-8
- # @Time : 2021/9/15 11:06 上午
- # @Site :
- # @File : TidbConn.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 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")
- @staticmethod
- def get_pro_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:
- TidbConn.get_connection().ping() # cping 校验连接是否异常
- _status = False
- except:
- if TidbConn.get_connection() == True: # 重新连接,成功退出
- _status = False
- break
- _number += 1
- time.sleep(stime) # 连接不成功,休眠3秒钟,继续循环,知道成功或重试次数结束
- @classmethod
- def quary(cls, conn, sql):
- # mysql_conn = TidbConn.get_connection()
- TidbConn._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):
- while True:
- try:
- conn.ping()
- break
- except OperationalError:
- conn.ping(True)
- try:
- TidbConn.exec_sql(conn, sql_buff)
- except Exception as e:
- logging.error('失败详细信息:', repr(e))
- conn.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 = TidbConn.get_connection()
- cursor = db.cursor()
- # 执行sql语句
- sql = "show databases"
- cursor.execute(sql)
- result = cursor.fetchall()
- print("result====", result)
|