TidbConn.py 2.5 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091
  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. import pymysql
  11. from Apietl.constant.ConfConstant import ConfConstant
  12. class TidbConn:
  13. """
  14. Tidb连接管理
  15. """
  16. @staticmethod
  17. def get_connection():
  18. """
  19. 获取Tidb连接
  20. :return: 连接
  21. """
  22. return pymysql.connect(host=ConfConstant.TIDB_HOST, user=ConfConstant.TIDB_USER,
  23. passwd=ConfConstant.TIDB_PASSWORD,
  24. port=ConfConstant.TIDB_PORT, charset="utf8")
  25. @classmethod
  26. def _reConn(cls, num=28800, stime=3): # 重试连接总次数为1天,这里根据实际情况自己设置,如果服务器宕机1天都没发现就......
  27. _number = 0
  28. _status = True
  29. while _status and _number <= num:
  30. try:
  31. TidbConn.get_connection().ping() # cping 校验连接是否异常
  32. _status = False
  33. except:
  34. if TidbConn.get_connection(): # 重新连接,成功退出
  35. _status = False
  36. break
  37. _number += 1
  38. time.sleep(stime) # 连接不成功,休眠3秒钟,继续循环,知道成功或重试次数结束
  39. @classmethod
  40. def quary(cls, sql):
  41. TidbConn._reConn(num=28800, stime=3)
  42. mysql_conn = TidbConn.get_connection()
  43. cur = mysql_conn.cursor()
  44. cur.execute(sql)
  45. result = cur.fetchall()
  46. return result
  47. @staticmethod
  48. def conn_close(conn: pymysql.connections.Connection):
  49. """
  50. 关闭资源
  51. """
  52. if conn:
  53. try:
  54. conn.close()
  55. except pymysql.err.Error:
  56. pass
  57. @staticmethod
  58. def exec_sql(conn, sql_buff):
  59. sql_list = sql_buff.split(";")
  60. cur = conn.cursor()
  61. for sql in sql_list:
  62. if sql.strip():
  63. cur.execute(sql)
  64. return cur
  65. @staticmethod
  66. def upsert(sql):
  67. TidbConn._reConn(num=28800, stime=3)
  68. try:
  69. TidbConn.exec_sql(TidbConn.get_connection(), sql)
  70. except Exception as e:
  71. logging.error('失败详细信息:', repr(e))
  72. TidbConn.get_connection().commit()
  73. if __name__ == '__main__':
  74. db = TidbConn.get_connection()
  75. cursor = db.cursor()
  76. # 执行sql语句
  77. sql = "show databases"
  78. cursor.execute(sql)
  79. result = cursor.fetchall()
  80. print("result====", result)