import pymysql
from DBUtils.PooledDB import PooledDB
from user_try.config import configuration
class MysqlConn:
"""
mysql Thread pool
"""
__my_pool = None
# How to return a dataset
TUPLE_CURSOR_MODE = pymysql.cursors.Cursor
DICT_DICTCURSOR_MODE = pymysql.cursors.DictCursor
TUPLE_SSCURSOR_MODE = pymysql.cursors.SSCursor
DICT_SSDICTCURSOR_MODE = pymysql.cursors.SSDictCursor
def __init__(self, database_name='DB', cur_type=pymysql.cursors.DictCursor):
self.conn = MysqlConn.get_connection(database_name)
self.cur = self.conn.cursor(cursor=cur_type)
@staticmethod
def get_connection(database_name):
"""
Get database connection
:param database_name:
:return:
"""
database = configuration.get_database_configuration(database_name)
host = database.get('host')
port = database.get('port')
password = database.get('pwd')
user = database.get('user')
if MysqlConn.__my_pool is None:
MysqlConn.__my_pool = PooledDB(creator=pymysql, mincached=1, maxcached=10, maxconnections=100,
blocking=True, # blocking parameter , When there are no idle connections in the pool True= Blocking , False= Don't wait to report an error directly
host=host, port=port, user=user, password=password, charset='utf8')
return MysqlConn.__my_pool.connection()
def close(self):
self.conn.close()
self.cur.close()
