|
@@ -0,0 +1,161 @@
|
|
|
+#!/usr/bin/env python
|
|
|
+# -*- encoding: utf-8 -*-
|
|
|
+"""
|
|
|
+Author renyupeng
|
|
|
+coding=utf-8
|
|
|
+@Time : 2023/2/7 下午
|
|
|
+@Site :
|
|
|
+@File : mysql_utils.py
|
|
|
+@Software: PyCharm
|
|
|
+@contact: renyupeng@c-top.com.cn
|
|
|
+@Tel 1501435553
|
|
|
+encoding=utf-8
|
|
|
+"""
|
|
|
+import pymysql
|
|
|
+# 导入所有Mysql配置常量,请自行指定文件
|
|
|
+from constant.ConfConstant import ConfConstant
|
|
|
+from utils.send_feishu_msg import SendFeiShuMsg
|
|
|
+
|
|
|
+
|
|
|
+class MysqlProUtils(object):
|
|
|
+ """
|
|
|
+ mysql操作类,对mysql数据库进行增删改查
|
|
|
+ """
|
|
|
+
|
|
|
+ def __init__(self):
|
|
|
+
|
|
|
+ config = dict(
|
|
|
+ host=ConfConstant.TIDB_PRO_HOST,
|
|
|
+ user=ConfConstant.TIDB_PRO_USER,
|
|
|
+ password=ConfConstant.TIDB_PRO_PASSWORD,
|
|
|
+ port=ConfConstant.TIDB_PRO_PORT,
|
|
|
+ charset='utf8mb4', # 编码要加上,否则可能出现中文乱码问题
|
|
|
+ use_unicode=False,
|
|
|
+ )
|
|
|
+ # Connect to the database
|
|
|
+ self.connection = pymysql.connect(**config)
|
|
|
+ self.connection.autocommit(True)
|
|
|
+ self.cursor = self.connection.cursor()
|
|
|
+
|
|
|
+ def QueryAll(self, sql):
|
|
|
+ """
|
|
|
+ 查询所有数据
|
|
|
+ :param sql:
|
|
|
+ :return:
|
|
|
+ """
|
|
|
+ # 数据库若断开即重连
|
|
|
+ self.reConnect()
|
|
|
+
|
|
|
+ self.cursor.execute(sql)
|
|
|
+ return self.cursor.fetchall()
|
|
|
+
|
|
|
+ def QueryMany(self, sql, n):
|
|
|
+ """
|
|
|
+ 查询某几条数据数据
|
|
|
+ :param sql:
|
|
|
+ :return:
|
|
|
+ """
|
|
|
+ # 数据库若断开即重连
|
|
|
+ self.reConnect()
|
|
|
+
|
|
|
+ self.cursor.execute(sql)
|
|
|
+ return self.cursor.fetchmany(n)
|
|
|
+
|
|
|
+ def QueryOne(self, sql):
|
|
|
+ """
|
|
|
+ 查询某几条数据数据
|
|
|
+ :param sql:
|
|
|
+ :return:
|
|
|
+ """
|
|
|
+ # 数据库若断开即重连
|
|
|
+ self.reConnect()
|
|
|
+
|
|
|
+ self.cursor.execute(sql)
|
|
|
+ return self.cursor.fetchone()
|
|
|
+
|
|
|
+ # return self.cursor.fetchone()
|
|
|
+
|
|
|
+ def reConnect(self):
|
|
|
+ """
|
|
|
+ 重连机制
|
|
|
+ :return:
|
|
|
+ """
|
|
|
+ try:
|
|
|
+ self.connection.ping()
|
|
|
+ except:
|
|
|
+ self.connection()
|
|
|
+
|
|
|
+ def Operate(self, sql, params=None, DML=True):
|
|
|
+ """
|
|
|
+ 数据库操作:增删改查
|
|
|
+ DML: insert / update / delete
|
|
|
+ DDL: CREATE TABLE/VIEW/INDEX/SYN/CLUSTER
|
|
|
+ """
|
|
|
+ try:
|
|
|
+ # 数据库若断开即重连
|
|
|
+ self.reConnect()
|
|
|
+
|
|
|
+ with self.connection.cursor() as cursor:
|
|
|
+ cursor.execute(sql, params)
|
|
|
+
|
|
|
+ self.connection.commit()
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ if DML:
|
|
|
+ # 涉及DML操作时,若抛异常需要回滚
|
|
|
+ self.connection.rollback()
|
|
|
+ print(e)
|
|
|
+
|
|
|
+ def __del__(self):
|
|
|
+ """
|
|
|
+ MysqlConnection实例对象被释放时调用此方法,用于关闭cursor和connection连接
|
|
|
+ """
|
|
|
+ self.cursor.close()
|
|
|
+ self.connection.close()
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def batch_insert(item_list):
|
|
|
+ con = pymysql.connect(host=ConfConstant.TIDB_PRO_HOST, user=ConfConstant.TIDB_PRO_USER,
|
|
|
+ passwd=ConfConstant.TIDB_PRO_PASSWORD, charset='utf8', port=3390)
|
|
|
+ # 对数据库进行插入操作,并不需要commit,twisted会自动commit
|
|
|
+ placeholders = ', '.join(['%s'] * len(tuple(item_list[0])))
|
|
|
+ print('====placeholders====', placeholders)
|
|
|
+
|
|
|
+ columns = ', '.join(item_list[0].keys())
|
|
|
+ print('====columns====', columns)
|
|
|
+ insert_sql = "REPLACE INTO ruixuan.promoter_info ( %s ) VALUES ( %s )" % (
|
|
|
+ columns, placeholders)
|
|
|
+ cur = con.cursor()
|
|
|
+ data = []
|
|
|
+ try:
|
|
|
+ for item in item_list:
|
|
|
+ value = tuple(item.values())
|
|
|
+ data.append(value)
|
|
|
+ cur.executemany(insert_sql, data)
|
|
|
+ con.commit()
|
|
|
+ except Exception as e:
|
|
|
+ SendFeiShuMsg.send_robot_msg('数据批量插入错误{e}'.format(e=e))
|
|
|
+ con.rollback()
|
|
|
+ con.close()
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def insert(item):
|
|
|
+ print(item,'---------')
|
|
|
+ con = pymysql.connect(host=ConfConstant.TIDB_PRO_HOST, user=ConfConstant.TIDB_PRO_USER,
|
|
|
+ passwd=ConfConstant.TIDB_PRO_PASSWORD, charset='utf8', port=3390)
|
|
|
+ async_item = tuple(item.values())
|
|
|
+ # 对数据库进行插入操作,并不需要commit,twisted会自动commit
|
|
|
+ placeholders = ', '.join(['%s'] * len(item))
|
|
|
+ columns = ', '.join(item.keys())
|
|
|
+ insert_sql = "REPLACE INTO ruixuan.promoter_info ( %s ) VALUES ( %s )" % (
|
|
|
+ columns, placeholders)
|
|
|
+ print(insert_sql)
|
|
|
+
|
|
|
+ cur = con.cursor()
|
|
|
+ try:
|
|
|
+ cur.execute(insert_sql, async_item)
|
|
|
+ con.commit()
|
|
|
+ except Exception as e:
|
|
|
+ SendFeiShuMsg.send_robot_msg('{table_name}数据插入错误{e}'.format(table_name=table_name, e=e))
|
|
|
+ con.rollback()
|
|
|
+ con.close()
|