123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161 |
- #!/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()
|