#!/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('数据插入错误{e}'.format(e=e)) con.rollback() con.close() @staticmethod def MgsInsert(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 miaogousi.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('数据插入错误{e}'.format(e=e)) con.rollback() con.close() @staticmethod def RocketInsert(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 rocket.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('数据插入错误{e}'.format(e=e)) con.rollback() con.close()