TidbConn.py 3.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118
  1. # Author renyupeng
  2. # coding=utf-8
  3. # @Time : 2021/9/15 11:06 上午
  4. # @Site :
  5. # @File : TidbConn.py
  6. # @Software: PyCharm
  7. # @contact: renyupeng@c-top.com.cn
  8. import logging
  9. import time
  10. from sqlite3 import OperationalError
  11. import pymysql
  12. from Apietl.constant.ConfConstant import ConfConstant
  13. class TidbConn:
  14. """
  15. Tidb连接管理
  16. """
  17. @staticmethod
  18. def get_connection():
  19. """
  20. 获取Tidb连接
  21. :return: 连接
  22. """
  23. return pymysql.connect(host=ConfConstant.TIDB_HOST, user=ConfConstant.TIDB_USER,
  24. passwd=ConfConstant.TIDB_PASSWORD,
  25. port=ConfConstant.TIDB_PORT, charset="utf8")
  26. @staticmethod
  27. def get_pro_connection():
  28. """
  29. 获取Tidb连接
  30. :return: 连接
  31. """
  32. return pymysql.connect(host=ConfConstant.TIDB_PRO_HOST, user=ConfConstant.TIDB_PRO_USER,
  33. passwd=ConfConstant.TIDB_PRO_PASSWORD,
  34. port=ConfConstant.TIDB_PRO_PORT, charset="utf8")
  35. @classmethod
  36. def _reConn(cls, num=28800, stime=5): # 重试连接总次数为1天,这里根据实际情况自己设置,如果服务器宕机1天都没发现就......
  37. _number = 0
  38. _status = True
  39. while _status and _number <= num:
  40. try:
  41. TidbConn.get_connection().ping() # cping 校验连接是否异常
  42. _status = False
  43. except:
  44. if TidbConn.get_connection() == True: # 重新连接,成功退出
  45. _status = False
  46. break
  47. _number += 1
  48. time.sleep(stime) # 连接不成功,休眠3秒钟,继续循环,知道成功或重试次数结束
  49. @classmethod
  50. def quary(cls, conn, sql):
  51. # mysql_conn = TidbConn.get_connection()
  52. TidbConn._reCon(conn)
  53. cur = conn.cursor()
  54. cur.execute(sql)
  55. result = cur.fetchall()
  56. return result
  57. @staticmethod
  58. def conn_close(conn: pymysql.connections.Connection):
  59. """
  60. 关闭资源
  61. """
  62. if conn:
  63. try:
  64. conn.close()
  65. except pymysql.err.Error:
  66. pass
  67. @staticmethod
  68. def exec_sql(conn, sql_buff):
  69. sql_list = sql_buff.split(";")
  70. cur = conn.cursor()
  71. for sql in sql_list:
  72. if sql.strip():
  73. cur.execute(sql)
  74. return cur
  75. @staticmethod
  76. def upsert(conn, sql_buff):
  77. while True:
  78. try:
  79. conn.ping()
  80. break
  81. except OperationalError:
  82. conn.ping(True)
  83. try:
  84. TidbConn.exec_sql(conn, sql_buff)
  85. except Exception as e:
  86. logging.error('失败详细信息:', repr(e))
  87. conn.commit()
  88. @staticmethod
  89. def _reCon(conn):
  90. """ MySQLdb.OperationalError异常"""
  91. # self.con.close()
  92. while True:
  93. try:
  94. conn.ping()
  95. break
  96. except OperationalError:
  97. conn.ping(True)
  98. if __name__ == '__main__':
  99. db = TidbConn.get_connection()
  100. cursor = db.cursor()
  101. # 执行sql语句
  102. sql = "show databases"
  103. cursor.execute(sql)
  104. result = cursor.fetchall()
  105. print("result====", result)