mysql_pool.py 30 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679
  1. # -*- coding: utf-8 -*-
  2. # Author : Charley
  3. # Python : 3.12.10
  4. # Date : 2025/3/25 14:14
  5. import re
  6. import time
  7. import pymysql
  8. import YamlLoader
  9. from loguru import logger
  10. from dbutils.pooled_db import PooledDB
  11. # 获取yaml配置
  12. yaml = YamlLoader.readYaml()
  13. mysqlYaml = yaml.get("mysql")
  14. sql_host = mysqlYaml.getValueAsString("host")
  15. sql_port = mysqlYaml.getValueAsInt("port")
  16. sql_user = mysqlYaml.getValueAsString("username")
  17. sql_password = mysqlYaml.getValueAsString("password")
  18. sql_db = mysqlYaml.getValueAsString("db")
  19. class MySQLConnectionPool:
  20. """
  21. MySQL连接池
  22. """
  23. def __init__(self, mincached=1, maxcached=2, maxconnections=3, log=None):
  24. """
  25. 初始化连接池
  26. :param mincached: 初始化时,链接池中至少创建的链接,0表示不创建
  27. :param maxcached: 池中空闲连接的最大数目(0 或 None 表示池大小不受限制)
  28. :param maxconnections: 允许的最大连接数(0 或 None 表示任意数量的连接)
  29. :param log: 自定义日志记录器
  30. """
  31. # 使用 loguru 的 logger,如果传入了其他 logger,则使用传入的 logger
  32. self.log = log or logger
  33. self.pool = PooledDB(
  34. creator=pymysql,
  35. mincached=mincached,
  36. maxcached=maxcached,
  37. maxconnections=maxconnections,
  38. blocking=True, # 连接池中如果没有可用连接后,是否阻塞等待。True,等待;False,不等待然后报错
  39. host=sql_host,
  40. port=sql_port,
  41. user=sql_user,
  42. password=sql_password,
  43. database=sql_db,
  44. ping=2, # 每次执行前检查连接有效性,防止使用已断开的连接
  45. connect_timeout=5, # 连接超时时间(秒)
  46. # read_timeout=30, # 读取超时时间(秒)
  47. write_timeout=30 # 写入超时时间(秒)
  48. )
  49. # def _execute(self, query, args=None, commit=False):
  50. # """
  51. # 执行SQL
  52. # :param query: SQL语句
  53. # :param args: SQL参数
  54. # :param commit: 是否提交事务
  55. # :return: 查询结果
  56. # """
  57. # try:
  58. # with self.pool.connection() as conn:
  59. # with conn.cursor() as cursor:
  60. # cursor.execute(query, args)
  61. # if commit:
  62. # conn.commit()
  63. # self.log.debug(f"sql _execute, Query: {query}, Rows: {cursor.rowcount}")
  64. # return cursor
  65. # except Exception as e:
  66. # if commit and conn:
  67. # conn.rollback()
  68. # self.log.exception(f"Error executing query: {e}, Query: {query}, Args: {args}")
  69. # raise e
  70. def _execute(self, query, args=None, commit=False):
  71. """
  72. 执行SQL(带断连重试,指数退避,覆盖分钟级网络波动)。
  73. Args:
  74. query (str): SQL语句。
  75. args: SQL参数,可以是元组、列表或 None。
  76. commit (bool, optional): 是否提交事务。Defaults to False。
  77. Returns:
  78. cursor: pymysql cursor,调用方可读取 rowcount / lastrowid / fetchone 等。
  79. Raises:
  80. pymysql.err.IntegrityError: 唯一键冲突等完整性错误。
  81. pymysql.err.OperationalError: 重试耗尽后仍无法连接数据库。
  82. """
  83. for attempt in range(15): # 指数退避:5s→10s→20s…最多 300s,覆盖约 10 分钟波动
  84. try:
  85. with self.pool.connection() as conn:
  86. with conn.cursor() as cursor:
  87. cursor.execute(query, args)
  88. if commit:
  89. conn.commit()
  90. self.log.debug(f"sql _execute, Query: {query}, Rows: {cursor.rowcount}")
  91. return cursor
  92. except (pymysql.err.InterfaceError, pymysql.err.OperationalError) as e:
  93. errno = e.args[0] if e.args else None
  94. if errno in (2006, 2013): # MySQL server gone / lost connection during query
  95. delay = min(5 * (2 ** attempt), 300)
  96. self.log.warning(
  97. f"数据库连接断开 (errno={errno}),{delay}s 后重试 "
  98. f"(attempt {attempt + 1}/15): {e}"
  99. )
  100. time.sleep(delay)
  101. continue
  102. self.log.error(f"Error executing query: {e}, Query: {query}")
  103. raise
  104. except pymysql.err.IntegrityError:
  105. raise
  106. except Exception as e:
  107. self.log.exception(f"Error executing query: {e}, Query: {query}, Args: {args}")
  108. raise
  109. raise pymysql.err.OperationalError(2013, f"重试 15 次后仍无法连接数据库,Query: {query[:100]}")
  110. def _execute_many(self, query, args_list, commit=True):
  111. """
  112. executemany 的断连重试包装,指数退避,覆盖分钟级网络波动。
  113. Args:
  114. query (str): SQL语句。
  115. args_list (list): 参数列表。
  116. commit (bool, optional): 是否提交事务。Defaults to True。
  117. Returns:
  118. int: 影响行数(cursor.rowcount)。
  119. Raises:
  120. pymysql.err.IntegrityError: 唯一键冲突等完整性错误。
  121. pymysql.err.OperationalError: 重试耗尽后仍无法连接数据库。
  122. """
  123. for attempt in range(15):
  124. try:
  125. with self.pool.connection() as conn:
  126. with conn.cursor() as cursor:
  127. cursor.executemany(query, args_list)
  128. if commit:
  129. conn.commit()
  130. return cursor.rowcount
  131. except (pymysql.err.InterfaceError, pymysql.err.OperationalError) as e:
  132. errno = e.args[0] if e.args else None
  133. if errno in (2006, 2013):
  134. delay = min(5 * (2 ** attempt), 300)
  135. self.log.warning(
  136. f"数据库连接断开 (errno={errno}),{delay}s 后重试 "
  137. f"(attempt {attempt + 1}/15): {e}"
  138. )
  139. time.sleep(delay)
  140. continue
  141. if commit:
  142. try:
  143. conn.rollback()
  144. except Exception:
  145. pass
  146. raise
  147. except pymysql.err.IntegrityError:
  148. if commit:
  149. try:
  150. conn.rollback()
  151. except Exception:
  152. pass
  153. raise
  154. except Exception as e:
  155. if commit:
  156. try:
  157. conn.rollback()
  158. except Exception:
  159. pass
  160. self.log.exception(f"Error executing many: {e}, Query: {query}")
  161. raise
  162. raise pymysql.err.OperationalError(2013, f"重试 15 次后仍无法连接数据库,Query: {query[:100]}")
  163. def select_one(self, query, args=None):
  164. """
  165. 执行查询,返回单个结果
  166. :param query: 查询语句
  167. :param args: 查询参数
  168. :return: 查询结果
  169. """
  170. cursor = self._execute(query, args)
  171. return cursor.fetchone()
  172. def select_all(self, query, args=None):
  173. """
  174. 执行查询,返回所有结果
  175. :param query: 查询语句
  176. :param args: 查询参数
  177. :return: 查询结果
  178. """
  179. cursor = self._execute(query, args)
  180. return cursor.fetchall()
  181. def insert_one(self, query, args):
  182. """
  183. 执行单条插入语句
  184. :param query: 插入语句
  185. :param args: 插入参数
  186. """
  187. self.log.info('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>data insert_one 入库中>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
  188. cursor = self._execute(query, args, commit=True)
  189. return cursor.lastrowid # 返回插入的ID
  190. def insert_all(self, query, args_list):
  191. """
  192. 执行批量插入语句,如果失败则逐条插入
  193. :param query: 插入语句
  194. :param args_list: 插入参数列表
  195. """
  196. conn = None
  197. cursor = None
  198. try:
  199. conn = self.pool.connection()
  200. cursor = conn.cursor()
  201. cursor.executemany(query, args_list)
  202. conn.commit()
  203. self.log.debug(f"sql insert_all, SQL: {query[:100]}..., Rows: {cursor.rowcount}")
  204. self.log.info('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>data insert_all 入库中>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
  205. except pymysql.err.IntegrityError as e:
  206. if "Duplicate entry" in str(e):
  207. conn.rollback()
  208. self.log.warning(f"批量插入遇到重复,开始逐条插入。错误: {e}")
  209. rowcount = 0
  210. for args in args_list:
  211. try:
  212. self.insert_one(query, args)
  213. rowcount += 1
  214. except pymysql.err.IntegrityError as e2:
  215. if "Duplicate entry" in str(e2):
  216. self.log.debug(f"跳过重复条目: {e2}")
  217. else:
  218. self.log.error(f"插入失败: {e2}")
  219. except Exception as e2:
  220. self.log.error(f"插入失败: {e2}")
  221. self.log.info(f"逐条插入完成: {rowcount}/{len(args_list)}条")
  222. else:
  223. conn.rollback()
  224. self.log.exception(f"数据库完整性错误: {e}")
  225. raise e
  226. except Exception as e:
  227. conn.rollback()
  228. self.log.exception(f"批量插入失败: {e}")
  229. raise e
  230. finally:
  231. if cursor:
  232. cursor.close()
  233. if conn:
  234. conn.close()
  235. def insert_one_or_dict(self, table=None, data=None, query=None, args=None, commit=True, ignore=False):
  236. """
  237. 单条插入(支持字典或原始SQL)
  238. :param table: 表名(字典插入时必需)
  239. :param data: 字典数据 {列名: 值}
  240. :param query: 直接SQL语句(与data二选一)
  241. :param args: SQL参数(query使用时必需)
  242. :param commit: 是否自动提交
  243. :param ignore: 是否使用ignore
  244. :return: 最后插入ID
  245. """
  246. if data is not None:
  247. if not isinstance(data, dict):
  248. raise ValueError("Data must be a dictionary")
  249. keys = ', '.join([self._safe_identifier(k) for k in data.keys()])
  250. values = ', '.join(['%s'] * len(data))
  251. # 构建 INSERT IGNORE 语句
  252. ignore_clause = "IGNORE" if ignore else ""
  253. query = f"INSERT {ignore_clause} INTO {self._safe_identifier(table)} ({keys}) VALUES ({values})"
  254. args = tuple(data.values())
  255. elif query is None:
  256. raise ValueError("Either data or query must be provided")
  257. try:
  258. cursor = self._execute(query, args, commit)
  259. self.log.info(f"sql insert_one_or_dict, Table: {table}, Rows: {cursor.rowcount}")
  260. self.log.info('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>data insert_one_or_dict 入库中>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
  261. return cursor.lastrowid
  262. except pymysql.err.IntegrityError as e:
  263. if "Duplicate entry" in str(e):
  264. # 重复条目用 warning 简短输出,不打印堆栈
  265. self.log.warning(f"插入跳过-重复条目 Table: {table}, {e.args[1] if len(e.args) > 1 else e}")
  266. return -1 # 返回 -1 表示重复条目被跳过
  267. else:
  268. self.log.error(f"数据库完整性错误 Table: {table}, Error: {e}")
  269. raise
  270. except Exception as e:
  271. self.log.error(f"insert_one_or_dict 失败 Table: {table}, Error: {e}")
  272. raise
  273. def insert_many(self, table=None, data_list=None, query=None, args_list=None, batch_size=1000, commit=True,
  274. ignore=False):
  275. """
  276. 批量插入(支持字典列表或原始SQL)
  277. :param table: 表名(字典插入时必需)
  278. :param data_list: 字典列表 [{列名: 值}]
  279. :param query: 直接SQL语句(与data_list二选一)
  280. :param args_list: SQL参数列表(query使用时必需)
  281. :param batch_size: 分批大小
  282. :param commit: 是否自动提交
  283. :param ignore: 是否使用ignore
  284. :return: 影响行数
  285. """
  286. if data_list is not None:
  287. if not data_list or not isinstance(data_list[0], dict):
  288. raise ValueError("Data_list must be a non-empty list of dictionaries")
  289. keys = ', '.join([self._safe_identifier(k) for k in data_list[0].keys()])
  290. values = ', '.join(['%s'] * len(data_list[0]))
  291. # 构建 INSERT IGNORE 语句
  292. ignore_clause = "IGNORE" if ignore else ""
  293. query = f"INSERT {ignore_clause} INTO {self._safe_identifier(table)} ({keys}) VALUES ({values})"
  294. args_list = [tuple(d.values()) for d in data_list]
  295. elif query is None:
  296. raise ValueError("Either data_list or query must be provided")
  297. total = 0
  298. for i in range(0, len(args_list), batch_size):
  299. batch = args_list[i:i + batch_size]
  300. try:
  301. total += self._execute_many(query, batch, commit=commit)
  302. except pymysql.err.IntegrityError as e:
  303. # 处理唯一索引冲突
  304. if "Duplicate entry" in str(e):
  305. if ignore:
  306. self.log.warning(f"批量插入遇到重复条目(ignore模式): {e}")
  307. else:
  308. # 没有使用 IGNORE,降级为逐条插入
  309. self.log.warning(f"批量插入遇到重复条目,开始逐条插入。错误: {e}")
  310. rowcount = 0
  311. for j, args in enumerate(batch):
  312. try:
  313. if data_list:
  314. self.insert_one_or_dict(
  315. table=table,
  316. data=dict(zip(data_list[0].keys(), args)),
  317. commit=commit,
  318. ignore=False
  319. )
  320. else:
  321. self.insert_one(query, args)
  322. rowcount += 1
  323. except pymysql.err.IntegrityError as e2:
  324. if "Duplicate entry" in str(e2):
  325. self.log.debug(f"跳过重复条目[{i+j+1}]: {e2}")
  326. else:
  327. self.log.error(f"插入失败[{i+j+1}]: {e2}")
  328. except Exception as e2:
  329. self.log.error(f"插入失败[{i+j+1}]: {e2}")
  330. total += rowcount
  331. self.log.info(f"批次逐条插入完成: 成功{rowcount}/{len(batch)}条")
  332. else:
  333. self.log.exception(f"数据库完整性错误: {e}")
  334. raise e
  335. except Exception as e:
  336. self.log.exception(f"批量插入失败: {e}")
  337. raise e
  338. if table:
  339. self.log.info(f"sql insert_many, Table: {table}, Total Rows: {total}")
  340. else:
  341. self.log.info(f"sql insert_many, Query: {query}, Total Rows: {total}")
  342. return total
  343. def insert_many_two(self, table=None, data_list=None, query=None, args_list=None, batch_size=1000, commit=True,
  344. ignore=False):
  345. """
  346. 批量插入(支持字典列表或原始SQL) - 备用方法
  347. :param table: 表名(字典插入时必需)
  348. :param data_list: 字典列表 [{列名: 值}]
  349. :param query: 直接SQL语句(与data_list二选一)
  350. :param args_list: SQL参数列表(query使用时必需)
  351. :param batch_size: 分批大小
  352. :param commit: 是否自动提交
  353. :param ignore: 是否使用INSERT IGNORE
  354. :return: 影响行数
  355. """
  356. if data_list is not None:
  357. if not data_list or not isinstance(data_list[0], dict):
  358. raise ValueError("Data_list must be a non-empty list of dictionaries")
  359. keys = ', '.join([self._safe_identifier(k) for k in data_list[0].keys()])
  360. values = ', '.join(['%s'] * len(data_list[0]))
  361. ignore_clause = "IGNORE" if ignore else ""
  362. query = f"INSERT {ignore_clause} INTO {self._safe_identifier(table)} ({keys}) VALUES ({values})"
  363. args_list = [tuple(d.values()) for d in data_list]
  364. elif query is None:
  365. raise ValueError("Either data_list or query must be provided")
  366. total = 0
  367. for i in range(0, len(args_list), batch_size):
  368. batch = args_list[i:i + batch_size]
  369. try:
  370. total += self._execute_many(query, batch, commit=commit)
  371. except pymysql.err.IntegrityError as e:
  372. if "Duplicate entry" in str(e) and not ignore:
  373. self.log.warning(f"批量插入遇到重复,降级为逐条插入: {e}")
  374. rowcount = 0
  375. for args in batch:
  376. try:
  377. self.insert_one(query, args)
  378. rowcount += 1
  379. except pymysql.err.IntegrityError as e2:
  380. if "Duplicate entry" in str(e2):
  381. self.log.debug(f"跳过重复条目: {e2}")
  382. else:
  383. self.log.error(f"插入失败: {e2}")
  384. except Exception as e2:
  385. self.log.error(f"插入失败: {e2}")
  386. total += rowcount
  387. else:
  388. self.log.exception(f"数据库完整性错误: {e}")
  389. raise e
  390. except Exception as e:
  391. self.log.exception(f"批量插入失败: {e}")
  392. raise e
  393. self.log.info(f"sql insert_many_two, Table: {table}, Total Rows: {total}")
  394. return total
  395. def insert_too_many(self, query, args_list, batch_size=1000):
  396. """
  397. 执行批量插入语句,分片提交, 单次插入大于十万+时可用, 如果失败则降级为逐条插入
  398. :param query: 插入语句
  399. :param args_list: 插入参数列表
  400. :param batch_size: 每次插入的条数
  401. """
  402. self.log.info(f"sql insert_too_many, Query: {query}, Total Rows: {len(args_list)}")
  403. for i in range(0, len(args_list), batch_size):
  404. batch = args_list[i:i + batch_size]
  405. try:
  406. self._execute_many(query, batch, commit=True)
  407. self.log.debug(f"insert_too_many -> Total Rows: {len(batch)}")
  408. except Exception as e:
  409. self.log.error(f"insert_too_many error. Trying single insert. Error: {e}")
  410. # 当前批次降级为单条插入
  411. for args in batch:
  412. self.insert_one(query, args)
  413. def update_one(self, query, args):
  414. """
  415. 执行单条更新语句
  416. :param query: 更新语句
  417. :param args: 更新参数
  418. """
  419. self.log.info('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>data update_one 更新中>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
  420. return self._execute(query, args, commit=True)
  421. def update_all(self, query, args_list):
  422. """
  423. 执行批量更新语句,如果失败则逐条更新
  424. :param query: 更新语句
  425. :param args_list: 更新参数列表
  426. """
  427. try:
  428. self._execute_many(query, args_list, commit=True)
  429. self.log.debug(f"sql update_all, SQL: {query}, Rows: {len(args_list)}")
  430. self.log.info('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>data update_all 更新中>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
  431. except Exception as e:
  432. self.log.error(f"Error executing query: {e}")
  433. # 如果批量更新失败,则逐条更新
  434. rowcount = 0
  435. for args in args_list:
  436. self.update_one(query, args)
  437. rowcount += 1
  438. self.log.debug(f'Batch update failed. Updated {rowcount} rows individually.')
  439. def update_one_or_dict(self, table=None, data=None, condition=None, query=None, args=None, commit=True):
  440. """
  441. 单条更新(支持字典或原始SQL)
  442. :param table: 表名(字典模式必需)
  443. :param data: 字典数据 {列名: 值}(与 query 二选一)
  444. :param condition: 更新条件,支持以下格式:
  445. - 字典: {"id": 1} → "WHERE id = %s"
  446. - 字符串: "id = 1" → "WHERE id = 1"(需自行确保安全)
  447. - 元组: ("id = %s", [1]) → "WHERE id = %s"(参数化查询)
  448. :param query: 直接SQL语句(与 data 二选一)
  449. :param args: SQL参数(query 模式下必需)
  450. :param commit: 是否自动提交
  451. :return: 影响行数
  452. :raises: ValueError 参数校验失败时抛出
  453. """
  454. # 参数校验
  455. if data is not None:
  456. if not isinstance(data, dict):
  457. raise ValueError("Data must be a dictionary")
  458. if table is None:
  459. raise ValueError("Table name is required for dictionary update")
  460. if condition is None:
  461. raise ValueError("Condition is required for dictionary update")
  462. # 构建 SET 子句
  463. set_clause = ", ".join([f"{self._safe_identifier(k)} = %s" for k in data.keys()])
  464. set_values = list(data.values())
  465. # 解析条件
  466. condition_clause, condition_args = self._parse_condition(condition)
  467. query = f"UPDATE {self._safe_identifier(table)} SET {set_clause} WHERE {condition_clause}"
  468. args = set_values + condition_args
  469. elif query is None:
  470. raise ValueError("Either data or query must be provided")
  471. # 执行更新
  472. cursor = self._execute(query, args, commit)
  473. # self.log.debug(
  474. # f"Updated table={table}, rows={cursor.rowcount}, query={query[:100]}...",
  475. # extra={"table": table, "rows": cursor.rowcount}
  476. # )
  477. return cursor.rowcount
  478. def _parse_condition(self, condition):
  479. """
  480. 解析条件为 (clause, args) 格式
  481. :param condition: 字典/字符串/元组
  482. :return: (str, list) SQL 子句和参数列表
  483. """
  484. if isinstance(condition, dict):
  485. clause = " AND ".join([f"{self._safe_identifier(k)} = %s" for k in condition.keys()])
  486. args = list(condition.values())
  487. elif isinstance(condition, str):
  488. clause = condition # 注意:需调用方确保安全
  489. args = []
  490. elif isinstance(condition, (tuple, list)) and len(condition) == 2:
  491. clause, args = condition[0], condition[1]
  492. if not isinstance(args, (list, tuple)):
  493. args = [args]
  494. else:
  495. raise ValueError("Condition must be dict/str/(clause, args)")
  496. return clause, args
  497. def update_many(self, table=None, data_list=None, condition_list=None, query=None, args_list=None, batch_size=500,
  498. commit=True):
  499. """
  500. 批量更新(支持字典列表或原始SQL)
  501. :param table: 表名(字典插入时必需)
  502. :param data_list: 字典列表 [{列名: 值}]
  503. :param condition_list: 条件列表(必须为字典,与data_list等长)
  504. :param query: 直接SQL语句(与data_list二选一)
  505. :param args_list: SQL参数列表(query使用时必需)
  506. :param batch_size: 分批大小
  507. :param commit: 是否自动提交
  508. :return: 影响行数
  509. """
  510. if data_list is not None:
  511. if not data_list or not isinstance(data_list[0], dict):
  512. raise ValueError("Data_list must be a non-empty list of dictionaries")
  513. if condition_list is None or len(data_list) != len(condition_list):
  514. raise ValueError("Condition_list must be provided and match the length of data_list")
  515. if not all(isinstance(cond, dict) for cond in condition_list):
  516. raise ValueError("All elements in condition_list must be dictionaries")
  517. # 获取第一个数据项和条件项的键
  518. first_data_keys = set(data_list[0].keys())
  519. first_cond_keys = set(condition_list[0].keys())
  520. # 构造基础SQL
  521. set_clause = ', '.join([self._safe_identifier(k) + ' = %s' for k in data_list[0].keys()])
  522. condition_clause = ' AND '.join([self._safe_identifier(k) + ' = %s' for k in condition_list[0].keys()])
  523. base_query = f"UPDATE {self._safe_identifier(table)} SET {set_clause} WHERE {condition_clause}"
  524. total = 0
  525. # 分批次处理
  526. for i in range(0, len(data_list), batch_size):
  527. batch_data = data_list[i:i + batch_size]
  528. batch_conds = condition_list[i:i + batch_size]
  529. batch_args = []
  530. # 检查当前批次的结构是否一致
  531. can_batch = True
  532. for data, cond in zip(batch_data, batch_conds):
  533. data_keys = set(data.keys())
  534. cond_keys = set(cond.keys())
  535. if data_keys != first_data_keys or cond_keys != first_cond_keys:
  536. can_batch = False
  537. break
  538. batch_args.append(tuple(data.values()) + tuple(cond.values()))
  539. if not can_batch:
  540. # 结构不一致,转为单条更新
  541. for data, cond in zip(batch_data, batch_conds):
  542. self.update_one_or_dict(table=table, data=data, condition=cond, commit=commit)
  543. total += 1
  544. continue
  545. # 执行批量更新
  546. try:
  547. rowcount = self._execute_many(base_query, batch_args, commit=commit)
  548. total += rowcount
  549. self.log.debug(f"Batch update succeeded. Rows: {rowcount}")
  550. except Exception as e:
  551. if commit:
  552. conn.rollback()
  553. self.log.error(f"Batch update failed: {e}")
  554. # 降级为单条更新
  555. for args, data, cond in zip(batch_args, batch_data, batch_conds):
  556. try:
  557. self._execute(base_query, args, commit=commit)
  558. total += 1
  559. except Exception as e2:
  560. self.log.error(f"Single update failed: {e2}, Data: {data}, Condition: {cond}")
  561. self.log.info(f"Total updated rows: {total}")
  562. return total
  563. elif query is not None:
  564. # 处理原始SQL和参数列表
  565. if args_list is None:
  566. raise ValueError("args_list must be provided when using query")
  567. total = 0
  568. for i in range(0, len(args_list), batch_size):
  569. batch_args = args_list[i:i + batch_size]
  570. try:
  571. rowcount = self._execute_many(query, batch_args, commit=commit)
  572. total += rowcount
  573. self.log.debug(f"Batch update succeeded. Rows: {rowcount}")
  574. except Exception as e:
  575. self.log.error(f"Batch update failed: {e}")
  576. # 降级为单条更新
  577. for args in batch_args:
  578. try:
  579. self._execute(query, args, commit=commit)
  580. total += 1
  581. except Exception as e2:
  582. self.log.error(f"Single update failed: {e2}, Args: {args}")
  583. self.log.info(f"Total updated rows: {total}")
  584. return total
  585. else:
  586. raise ValueError("Either data_list or query must be provided")
  587. def check_pool_health(self):
  588. """
  589. 检查连接池中有效连接数
  590. # 使用示例
  591. # 配置 MySQL 连接池
  592. sql_pool = MySQLConnectionPool(log=log)
  593. if not sql_pool.check_pool_health():
  594. log.error("数据库连接池异常")
  595. raise RuntimeError("数据库连接池异常")
  596. """
  597. try:
  598. with self.pool.connection() as conn:
  599. conn.ping(reconnect=True)
  600. return True
  601. except Exception as e:
  602. self.log.error(f"Connection pool health check failed: {e}")
  603. return False
  604. def close(self):
  605. """
  606. 关闭连接池,释放所有连接
  607. """
  608. try:
  609. if hasattr(self, 'pool') and self.pool:
  610. self.pool.close()
  611. self.log.info("数据库连接池已关闭")
  612. except Exception as e:
  613. self.log.error(f"关闭连接池失败: {e}")
  614. @staticmethod
  615. def _safe_identifier(name):
  616. """SQL标识符安全校验"""
  617. if not re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', name):
  618. raise ValueError(f"Invalid SQL identifier: {name}")
  619. return name
  620. if __name__ == '__main__':
  621. sql_pool = MySQLConnectionPool()
  622. data_dic = {'card_type_id': 111, 'card_type_name': '补充包 继承的意志【OPC-13】', 'card_type_position': 964,
  623. 'card_id': 5284, 'card_name': '蒙奇·D·路飞', 'card_number': 'OP13-001', 'card_rarity': 'L',
  624. 'card_img': 'https://source.windoent.com/OnePiecePc/Picture/1757929283612OP13-001.png',
  625. 'card_life': '4', 'card_attribute': '打', 'card_power': '5000', 'card_attack': '-',
  626. 'card_color': '红/绿', 'subscript': 4, 'card_features': '超新星/草帽一伙',
  627. 'card_text_desc': '【咚!!×1】【对方的攻击时】我方处于活跃状态的咚!!不多于5张的场合,可以将我方任意张数的咚!!转为休息状态。每有1张转为休息状态的咚!!,本次战斗中,此领袖或我方最多1张拥有《草帽一伙》特征的角色力量+2000。',
  628. 'card_offer_type': '补充包 继承的意志【OPC-13】', 'crawler_language': '简中'}
  629. sql_pool.insert_one_or_dict(table="one_piece_record", data=data_dic)