mysql_utils_pro.py 4.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161
  1. #!/usr/bin/env python
  2. # -*- encoding: utf-8 -*-
  3. """
  4. Author renyupeng
  5. coding=utf-8
  6. @Time : 2023/2/7 下午
  7. @Site :
  8. @File : mysql_utils.py
  9. @Software: PyCharm
  10. @contact: renyupeng@c-top.com.cn
  11. @Tel 1501435553
  12. encoding=utf-8
  13. """
  14. import pymysql
  15. # 导入所有Mysql配置常量,请自行指定文件
  16. from constant.ConfConstant import ConfConstant
  17. from utils.send_feishu_msg import SendFeiShuMsg
  18. class MysqlProUtils(object):
  19. """
  20. mysql操作类,对mysql数据库进行增删改查
  21. """
  22. def __init__(self):
  23. config = dict(
  24. host=ConfConstant.TIDB_PRO_HOST,
  25. user=ConfConstant.TIDB_PRO_USER,
  26. password=ConfConstant.TIDB_PRO_PASSWORD,
  27. port=ConfConstant.TIDB_PRO_PORT,
  28. charset='utf8mb4', # 编码要加上,否则可能出现中文乱码问题
  29. use_unicode=False,
  30. )
  31. # Connect to the database
  32. self.connection = pymysql.connect(**config)
  33. self.connection.autocommit(True)
  34. self.cursor = self.connection.cursor()
  35. def QueryAll(self, sql):
  36. """
  37. 查询所有数据
  38. :param sql:
  39. :return:
  40. """
  41. # 数据库若断开即重连
  42. self.reConnect()
  43. self.cursor.execute(sql)
  44. return self.cursor.fetchall()
  45. def QueryMany(self, sql, n):
  46. """
  47. 查询某几条数据数据
  48. :param sql:
  49. :return:
  50. """
  51. # 数据库若断开即重连
  52. self.reConnect()
  53. self.cursor.execute(sql)
  54. return self.cursor.fetchmany(n)
  55. def QueryOne(self, sql):
  56. """
  57. 查询某几条数据数据
  58. :param sql:
  59. :return:
  60. """
  61. # 数据库若断开即重连
  62. self.reConnect()
  63. self.cursor.execute(sql)
  64. return self.cursor.fetchone()
  65. # return self.cursor.fetchone()
  66. def reConnect(self):
  67. """
  68. 重连机制
  69. :return:
  70. """
  71. try:
  72. self.connection.ping()
  73. except:
  74. self.connection()
  75. def Operate(self, sql, params=None, DML=True):
  76. """
  77. 数据库操作:增删改查
  78. DML: insert / update / delete
  79. DDL: CREATE TABLE/VIEW/INDEX/SYN/CLUSTER
  80. """
  81. try:
  82. # 数据库若断开即重连
  83. self.reConnect()
  84. with self.connection.cursor() as cursor:
  85. cursor.execute(sql, params)
  86. self.connection.commit()
  87. except Exception as e:
  88. if DML:
  89. # 涉及DML操作时,若抛异常需要回滚
  90. self.connection.rollback()
  91. print(e)
  92. def __del__(self):
  93. """
  94. MysqlConnection实例对象被释放时调用此方法,用于关闭cursor和connection连接
  95. """
  96. self.cursor.close()
  97. self.connection.close()
  98. @staticmethod
  99. def batch_insert(item_list):
  100. con = pymysql.connect(host=ConfConstant.TIDB_PRO_HOST, user=ConfConstant.TIDB_PRO_USER,
  101. passwd=ConfConstant.TIDB_PRO_PASSWORD, charset='utf8', port=3390)
  102. # 对数据库进行插入操作,并不需要commit,twisted会自动commit
  103. placeholders = ', '.join(['%s'] * len(tuple(item_list[0])))
  104. print('====placeholders====', placeholders)
  105. columns = ', '.join(item_list[0].keys())
  106. print('====columns====', columns)
  107. insert_sql = "REPLACE INTO ruixuan.promoter_info ( %s ) VALUES ( %s )" % (
  108. columns, placeholders)
  109. cur = con.cursor()
  110. data = []
  111. try:
  112. for item in item_list:
  113. value = tuple(item.values())
  114. data.append(value)
  115. cur.executemany(insert_sql, data)
  116. con.commit()
  117. except Exception as e:
  118. SendFeiShuMsg.send_robot_msg('数据批量插入错误{e}'.format(e=e))
  119. con.rollback()
  120. con.close()
  121. @staticmethod
  122. def insert(item):
  123. print(item,'---------')
  124. con = pymysql.connect(host=ConfConstant.TIDB_PRO_HOST, user=ConfConstant.TIDB_PRO_USER,
  125. passwd=ConfConstant.TIDB_PRO_PASSWORD, charset='utf8', port=3390)
  126. async_item = tuple(item.values())
  127. # 对数据库进行插入操作,并不需要commit,twisted会自动commit
  128. placeholders = ', '.join(['%s'] * len(item))
  129. columns = ', '.join(item.keys())
  130. insert_sql = "REPLACE INTO ruixuan.promoter_info ( %s ) VALUES ( %s )" % (
  131. columns, placeholders)
  132. print(insert_sql)
  133. cur = con.cursor()
  134. try:
  135. cur.execute(insert_sql, async_item)
  136. con.commit()
  137. except Exception as e:
  138. SendFeiShuMsg.send_robot_msg('{table_name}数据插入错误{e}'.format(table_name=table_name, e=e))
  139. con.rollback()
  140. con.close()