|
|
#!/usr/bin/env python
|
|
|
# -*- encoding: utf-8 -*-
|
|
|
'''
|
|
|
@Date:2022/08/26 09:48:12
|
|
|
'''
|
|
|
import sys
|
|
|
sys.path.append('.')
|
|
|
import pymysql
|
|
|
from dbutils.pooled_db import PooledDB
|
|
|
from Common.log_utlis import logger
|
|
|
|
|
|
def list_base_str(data_list):
|
|
|
if data_list:
|
|
|
# data = []
|
|
|
resp_data = []
|
|
|
for i in data_list:
|
|
|
tup_data = []
|
|
|
key, value = "", ""
|
|
|
for k,v in i.items():
|
|
|
if k == "is_add":
|
|
|
v = 0
|
|
|
if v is None:
|
|
|
continue
|
|
|
key += f"{k},"
|
|
|
value += "%s,"
|
|
|
tup_data.append(v)
|
|
|
# data.append(tuple(tup_data))
|
|
|
if key.count(",") > 1:
|
|
|
key = key[:-1]
|
|
|
value = value[:-1]
|
|
|
resp_data.append(
|
|
|
{
|
|
|
"key":key,
|
|
|
"value":value,
|
|
|
"data":tuple(tup_data)
|
|
|
}
|
|
|
)
|
|
|
|
|
|
return resp_data
|
|
|
return None
|
|
|
|
|
|
|
|
|
class MysqlPool:
|
|
|
"""
|
|
|
MySQL 数据库连接池类 配置变量
|
|
|
"""
|
|
|
|
|
|
'''
|
|
|
:param
|
|
|
reset: how connections should be reset when returned to the pool
|
|
|
(False or None to rollback transcations started with begin(),
|
|
|
True to always issue a rollback for safety's sake)
|
|
|
|
|
|
:param
|
|
|
setsession: optional list of SQL commands that may serve to prepare
|
|
|
the session, e.g. ["set datestyle to ...", "set time zone ..."]
|
|
|
'''
|
|
|
|
|
|
'''
|
|
|
https://blog.51cto.com/abyss/1736844
|
|
|
其中的
|
|
|
setsession=['SET AUTOCOMMIT = 1']
|
|
|
就是用来设置线程池是否打开自动更新的配置，0为False，1为True
|
|
|
'''
|
|
|
|
|
|
# 初始化数据库连接池变量
|
|
|
__pool = None
|
|
|
|
|
|
# 创建连接池的最大数量
|
|
|
__MAX_CONNECTIONS = 20
|
|
|
# 连接池中空闲连接的初始数量
|
|
|
__MIN_CACHED = 5
|
|
|
# 连接池中空闲连接的最大数量
|
|
|
__MAX_CACHED = 5
|
|
|
# 共享连接的最大数量
|
|
|
__MAX_SHARED = 0
|
|
|
|
|
|
# 超过最大连接数量时候的表现，为True等待连接数量下降，为false直接报错处理
|
|
|
__BLOCK = True
|
|
|
# 单个连接的最大重复使用次数
|
|
|
__MAX_USAGE = 1
|
|
|
|
|
|
# 当返回到池时，连接应该如何重置
|
|
|
# (False或None回滚以begin()开始的事务，为了安全起见，总是发出回滚)
|
|
|
# 设置自动提交
|
|
|
__SET_SESSION = []
|
|
|
|
|
|
# 不能是 UTF-8
|
|
|
__CHARSET = 'UTF8'
|
|
|
|
|
|
def __init__(self, host, port, user, password, database):
|
|
|
"""
|
|
|
:param host: 数据库主机地址
|
|
|
:param port: 端口号
|
|
|
:param user: 用户名
|
|
|
:param password: 密码
|
|
|
:param database: 数据库名
|
|
|
"""
|
|
|
if not self.__pool:
|
|
|
# self代表当前类的实例，即为 MysqlPool（） 带小括号，执行后的数据。
|
|
|
# __class__，魔法函数，代表从当前类的实例中，获取当前类，即为 MysqlPool 不带小括号的类。
|
|
|
# __pool，这个代表的事类的变量，即为在类下面创建的初始化连接池，__pool
|
|
|
self.__class__.__pool = PooledDB(
|
|
|
creator=pymysql,
|
|
|
host=host,
|
|
|
port=port,
|
|
|
user=user,
|
|
|
password=password,
|
|
|
database=database,
|
|
|
maxconnections=self.__MAX_CONNECTIONS,
|
|
|
mincached=self.__MIN_CACHED,
|
|
|
maxcached=self.__MAX_CACHED,
|
|
|
maxshared=self.__MAX_SHARED,
|
|
|
blocking=self.__BLOCK,
|
|
|
maxusage=self.__MAX_USAGE,
|
|
|
setsession=self.__SET_SESSION,
|
|
|
charset=self.__CHARSET
|
|
|
)
|
|
|
|
|
|
def get_connect(self):
|
|
|
return self.__pool.connection()
|
|
|
|
|
|
|
|
|
class DbManager(object):
|
|
|
|
|
|
conn = None
|
|
|
cur = None
|
|
|
POOL = MysqlPool(host='127.0.0.1',port=3306,user='root',password='123456',database='test111')
|
|
|
|
|
|
|
|
|
def connectDatabase(self):
|
|
|
try:
|
|
|
self.conn = self.POOL.get_connect()
|
|
|
self.cur = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
|
|
|
return True
|
|
|
except:
|
|
|
logger.error("connectDatabase failed")
|
|
|
return False
|
|
|
|
|
|
# 关闭数据库
|
|
|
def close(self):
|
|
|
if self.conn and self.cur:
|
|
|
self.cur.close()
|
|
|
self.conn.close()
|
|
|
return True
|
|
|
|
|
|
# 基本的执行SQL方法，下面几乎都调用这个
|
|
|
def execute(self, sql, params=None, exe_many=False):
|
|
|
res = self.connectDatabase()
|
|
|
if not res:
|
|
|
return False
|
|
|
cnt = 0
|
|
|
try:
|
|
|
if self.conn and self.cur:
|
|
|
if exe_many:
|
|
|
cnt = self.cur.executemany(sql, params)
|
|
|
else:
|
|
|
cnt = self.cur.execute(sql)
|
|
|
self.conn.commit()
|
|
|
except Exception as e:
|
|
|
logger.error("execute failed: " + sql)
|
|
|
logger.error(str(e) + "\n\n")
|
|
|
return False
|
|
|
finally:
|
|
|
self.close()
|
|
|
return cnt
|
|
|
|
|
|
###############################################################
|
|
|
### 增删改查
|
|
|
###############################################################
|
|
|
|
|
|
# 存在不做操作，不存在则新增
|
|
|
def insert_ignore(self, **kwargs):
|
|
|
"""
|
|
|
table：必填，表名，如：table="test_table"
|
|
|
data ：必填，更新数据，字典类型，如：data=[{"aaa": "666'6", "bbb": "888"}]
|
|
|
|
|
|
"key":key,
|
|
|
"value":value,
|
|
|
"data":data
|
|
|
"""
|
|
|
table = kwargs["table"]
|
|
|
data = kwargs["data"]
|
|
|
local_data = kwargs.get("update_obj")
|
|
|
# table_obj = kwargs.get("table_obj")
|
|
|
data_dict = list_base_str(data)
|
|
|
if not data_dict:
|
|
|
return None
|
|
|
for i in data_dict:
|
|
|
sql = f"insert ignore {table}({i.get('key')}) values{i.get('data')};"
|
|
|
logger.info("sql：\n{}\n".format(sql))
|
|
|
try:
|
|
|
self.execute(sql)
|
|
|
if local_data:
|
|
|
local_data.executeNoParam(
|
|
|
f"update {table} set is_add=1 where {i.get('key').split(',')[0]}='{i.get('data')[0]}'")
|
|
|
except:
|
|
|
self.conn.rollback()
|
|
|
|
|
|
# 存在不做操作，不存在则新增
|
|
|
def insert_ignore_update(self, **kwargs):
|
|
|
"""
|
|
|
table：必填，表名，如：table="test_table"
|
|
|
data ：必填，更新数据，字典类型，如：data=[{"aaa": "666'6", "bbb": "888"}]
|
|
|
|
|
|
"key":key,
|
|
|
"value":value,
|
|
|
"data":data
|
|
|
"""
|
|
|
table = kwargs["table"]
|
|
|
data = kwargs["data"]
|
|
|
# local_data = kwargs.get("update_obj")
|
|
|
# table_obj = kwargs.get("table_obj")
|
|
|
data_dict = list_base_str(data)
|
|
|
if not data_dict:
|
|
|
return None
|
|
|
for i in data_dict:
|
|
|
"""
|
|
|
replace into 'subject' (subjectId,subjectName)
|
|
|
values (1,'离散');
|
|
|
"""
|
|
|
sql = f"replace into {table}({i.get('key')}) values{i.get('data')};"
|
|
|
# sql = f"insert ignore {table}({i.get('key')}) values{i.get('data')};"
|
|
|
logger.info("sql：\n{}\n".format(sql))
|
|
|
try:
|
|
|
self.execute(sql)
|
|
|
# if local_data:
|
|
|
# local_data.executeNoParam(
|
|
|
# f"update {table} set is_add=1 where {i.get('key').split(',')[0]}='{i.get('data')[0]}'")
|
|
|
except:
|
|
|
self.conn.rollback()
|
|
|
# def insert_ignore
|
|
|
|
|
|
|
|
|
|
|
|
# if __name__ == '__main__':
|
|
|
# from common.utlis import get_uuid
|
|
|
# db_obj = DbManager()
|
|
|
# # table：必填，表名，如：table = "test_table"
|
|
|
# # data ：必填，更新数据，字典类型，如：data=[{"aaa": "666'6", "bbb": "888"}]
|
|
|
# data_dic = {
|
|
|
# "table":"rms_user",
|
|
|
# "data":[
|
|
|
# {
|
|
|
# 'user_id': "f87cd129-2501-11ed-af7b-f47b094925e2",
|
|
|
# 'user_code': 10991,
|
|
|
# 'account': '测试11',
|
|
|
# 'real_name':" 测试123",
|
|
|
# 'sex': 1,
|
|
|
# 'mobile': "10111111111",
|
|
|
# 'is_enabled': 1,
|
|
|
# 'last_visit_date': '2022-07-15 08:30:47',
|
|
|
# 'is_add': 1
|
|
|
# }
|
|
|
# ]
|
|
|
# }
|
|
|
# db_obj.insert_ignore(**data_dic) |
