選択できるのは25トピックまでです。 トピックは、先頭が英数字で、英数字とダッシュ('-')を使用した35文字以内のものにしてください。

228 行
7.2KB

  1. # -*- coding: UTF-8 -*-
  2. import pymysql
  3. from loguru import logger
  4. from dbutils.pooled_db import PooledDB
  5. """
  6. @功能:创建数据库连接池
  7. """
  8. class MyConnectionPool(object):
  9. __pool = None
  10. def __init__(self, content):
  11. self.conn = self.__getConn(content)
  12. self.cursor = self.conn.cursor()
  13. # 创建数据库连接conn和游标cursor
  14. # def __enter__(self):
  15. # self.conn = self.__getconn()
  16. # self.cursor = self.conn.cursor()
  17. # 创建数据库连接池
  18. def __getconn(self, content):
  19. if self.__pool is None:
  20. self.__pool = PooledDB(
  21. creator=pymysql,
  22. mincached=int(content["mysql"]["db_min_cached"]),
  23. maxcached=int(content["mysql"]["db_max_cached"]),
  24. maxshared=int(content["mysql"]["db_max_shared"]),
  25. maxconnections=int(content["mysql"]["db_max_connecyions"]),
  26. blocking=content["mysql"]["db_blocking"],
  27. maxusage=content["mysql"]["db_max_usage"],
  28. setsession=content["mysql"]["db_set_session"],
  29. host=content["mysql"][content["dsp"]["active"]]["host"],
  30. port=content["mysql"][content["dsp"]["active"]]["port"],
  31. user=content["mysql"][content["dsp"]["active"]]["username"],
  32. passwd=content["mysql"][content["dsp"]["active"]]["password"],
  33. db=content["mysql"][content["dsp"]["active"]]["dbname"],
  34. use_unicode=False,
  35. charset=content["mysql"]["db_charset"]
  36. )
  37. return self.__pool.connection()
  38. # 释放连接池资源
  39. # def __exit__(self, exc_type, exc_val, exc_tb):
  40. # self.cursor.close()
  41. # self.conn.close()
  42. # 关闭连接归还给链接池
  43. def close(self):
  44. self.cursor.close()
  45. self.conn.close()
  46. # 从连接池中取出一个连接
  47. def getconn(self, content):
  48. conn = self.__getconn(content)
  49. cursor = conn.cursor()
  50. return cursor, conn
  51. # 获取连接池,实例化
  52. def get_my_connection(content):
  53. return MyConnectionPool(content)
  54. '''
  55. 执行语句查询有结果返回结果没有返回0;增/删/改返回变更数据条数,没有返回0
  56. '''
  57. class MySqLHelper(object):
  58. def __init__(self, content):
  59. logger.info("开始加载数据库连接池!")
  60. self.db = get_my_connection(content)
  61. logger.info("加载数据库连接池完成!")
  62. def __new__(cls, *args, **kwargs):
  63. if not hasattr(cls, 'inst'): # 单例
  64. cls.inst = super(MySqLHelper, cls).__new__(cls, *args, **kwargs)
  65. return cls.inst
  66. # 封装执行命令
  67. def execute(self, sql, param=None, autoclose=False):
  68. """
  69. 【主要判断是否有参数和是否执行完就释放连接】
  70. :param sql: 字符串类型,sql语句
  71. :param param: sql语句中要替换的参数"select %s from tab where id=%s" 其中的%s就是参数
  72. :param autoclose: 是否关闭连接
  73. :return: 返回连接conn和游标cursor
  74. """
  75. cursor, conn = self.db.getconn() # 从连接池获取连接
  76. count = 0
  77. try:
  78. # count : 为改变的数据条数
  79. if param:
  80. count = cursor.execute(sql, param)
  81. else:
  82. count = cursor.execute(sql)
  83. conn.commit()
  84. if autoclose:
  85. self.close(cursor, conn)
  86. except Exception as e:
  87. pass
  88. return cursor, conn, count
  89. # 执行多条命令
  90. # def executemany(self, lis):
  91. # """
  92. # :param lis: 是一个列表,里面放的是每个sql的字典'[{"sql":"xxx","param":"xx"}....]'
  93. # :return:
  94. # """
  95. # cursor, conn = self.db.getconn()
  96. # try:
  97. # for order in lis:
  98. # sql = order['sql']
  99. # param = order['param']
  100. # if param:
  101. # cursor.execute(sql, param)
  102. # else:
  103. # cursor.execute(sql)
  104. # conn.commit()
  105. # self.close(cursor, conn)
  106. # return True
  107. # except Exception as e:
  108. # print(e)
  109. # conn.rollback()
  110. # self.close(cursor, conn)
  111. # return False
  112. # 释放连接
  113. def close(self, cursor, conn):
  114. logger.info("开始释放数据库连接!")
  115. cursor.close()
  116. conn.close()
  117. logger.info("释放数据库连接完成!")
  118. # 查询所有
  119. def selectall(self, sql, param=None):
  120. try:
  121. cursor, conn, count = self.execute(sql, param)
  122. res = cursor.fetchall()
  123. return res
  124. except Exception as e:
  125. logger.error("查询所有数据异常:")
  126. logger.exception(e)
  127. self.close(cursor, conn)
  128. return count
  129. # 查询单条
  130. def selectone(self, sql, param=None):
  131. try:
  132. cursor, conn, count = self.execute(sql, param)
  133. res = cursor.fetchone()
  134. self.close(cursor, conn)
  135. return res
  136. except Exception as e:
  137. logger.error("查询单条数据异常:")
  138. logger.exception(e)
  139. self.close(cursor, conn)
  140. return count
  141. # 增加
  142. def insertone(self, sql, param):
  143. try:
  144. cursor, conn, count = self.execute(sql, param)
  145. # _id = cursor.lastrowid() # 获取当前插入数据的主键id,该id应该为自动生成为好
  146. conn.commit()
  147. self.close(cursor, conn)
  148. return count
  149. # 防止表中没有id返回0
  150. # if _id == 0:
  151. # return True
  152. # return _id
  153. except Exception as e:
  154. logger.error("新增数据异常:")
  155. logger.exception(e)
  156. conn.rollback()
  157. self.close(cursor, conn)
  158. return count
  159. # 增加多行
  160. def insertmany(self, sql, param):
  161. """
  162. :param sql:
  163. :param param: 必须是元组或列表[(),()]或((),())
  164. :return:
  165. """
  166. cursor, conn, count = self.db.getconn()
  167. try:
  168. cursor.executemany(sql, param)
  169. conn.commit()
  170. return count
  171. except Exception as e:
  172. logger.error("增加多条数据异常:")
  173. logger.exception(e)
  174. conn.rollback()
  175. self.close(cursor, conn)
  176. return count
  177. # 删除
  178. def delete(self, sql, param=None):
  179. try:
  180. cursor, conn, count = self.execute(sql, param)
  181. self.close(cursor, conn)
  182. return count
  183. except Exception as e:
  184. logger.error("删除数据异常:")
  185. logger.exception(e)
  186. conn.rollback()
  187. self.close(cursor, conn)
  188. return count
  189. # 更新
  190. def update(self, sql, param=None):
  191. try:
  192. cursor, conn, count = self.execute(sql, param)
  193. conn.commit()
  194. self.close(cursor, conn)
  195. return count
  196. except Exception as e:
  197. logger.error("更新数据异常:")
  198. logger.exception(e)
  199. conn.rollback()
  200. self.close(cursor, conn)
  201. return count