# -*- coding: UTF-8 -*-import pymysql# 数据库信息DB_TEST_HOST = "127.0.0.1"DB_TEST_PORT = 3306DB_TEST_DBNAME = "ball"DB_TEST_USER = "root"DB_TEST_PASSWORD = "123456"# 数据库连接编码DB_CHARSET = "utf8"# mincached : 启动时开启的闲置连接数量(缺省值 0 开始时不创建连接)DB_MIN_CACHED = 10# maxcached : 连接池中允许的闲置的最多连接数量(缺省值 0 代表不闲置连接池大小)DB_MAX_CACHED = 10# maxshared : 共享连接数允许的最大数量(缺省值 0 代表所有连接都是专用的)如果达到了最大数量,被请求为共享的连接将会被共享使用DB_MAX_SHARED = 20# maxconnecyions : 创建连接池的最大数量(缺省值 0 代表不限制)DB_MAX_CONNECYIONS = 100# blocking : 设置在连接池达到最大数量时的行为(缺省值 0 或 False 代表返回一个错误<toMany......> 其他代表阻塞直到连接数减少,连接被分配)DB_BLOCKING = True# maxusage : 单个连接的最大允许复用次数(缺省值 0 或 False 代表不限制的复用).当达到最大数时,连接会自动重新连接(关闭和重新打开)DB_MAX_USAGE = 0# setsession : 一个可选的SQL命令列表用于准备每个会话，如["set datestyle to german", ...]DB_SET_SESSION = None# creator : 使用连接数据库的模块DB_CREATOR = pymysql

from DBUtils.PooledDB import PooledDBimport db_config as config"""@功能：创建数据库连接池"""class MyConnectionPool(object):    __pool = None    # def __init__(self):    #     self.conn = self.__getConn()    #     self.cursor = self.conn.cursor()    # 创建数据库连接conn和游标cursor    def __enter__(self):        self.conn = self.__getconn()        self.cursor = self.conn.cursor()    # 创建数据库连接池    def __getconn(self):        if self.__pool is None:            self.__pool = PooledDB(                creator=config.DB_CREATOR,                mincached=config.DB_MIN_CACHED,                maxcached=config.DB_MAX_CACHED,                maxshared=config.DB_MAX_SHARED,                maxconnections=config.DB_MAX_CONNECYIONS,                blocking=config.DB_BLOCKING,                maxusage=config.DB_MAX_USAGE,                setsession=config.DB_SET_SESSION,                host=config.DB_TEST_HOST,                port=config.DB_TEST_PORT,                user=config.DB_TEST_USER,                passwd=config.DB_TEST_PASSWORD,                db=config.DB_TEST_DBNAME,                use_unicode=False,                charset=config.DB_CHARSET            )        return self.__pool.connection()    # 释放连接池资源    def __exit__(self, exc_type, exc_val, exc_tb):        self.cursor.close()        self.conn.close()    # 关闭连接归还给链接池    # def close(self):    #     self.cursor.close()    #     self.conn.close()    # 从连接池中取出一个连接    def getconn(self):        conn = self.__getconn()        cursor = conn.cursor()        return cursor, conn# 获取连接池,实例化def get_my_connection():    return MyConnectionPool()

from db_dbutils_init import get_my_connection"""执行语句查询有结果返回结果没有返回0；增/删/改返回变更数据条数，没有返回0"""class MySqLHelper(object):    def __init__(self):        self.db = get_my_connection()  # 从数据池中获取连接    def __new__(cls, *args, **kwargs):        if not hasattr(cls, 'inst'):  # 单例            cls.inst = super(MySqLHelper, cls).__new__(cls, *args, **kwargs)        return cls.inst    # 封装执行命令    def execute(self, sql, param=None, autoclose=False):        """        【主要判断是否有参数和是否执行完就释放连接】        :param sql: 字符串类型，sql语句        :param param: sql语句中要替换的参数"select %s from tab where id=%s" 其中的%s就是参数        :param autoclose: 是否关闭连接        :return: 返回连接conn和游标cursor        """        cursor, conn = self.db.getconn()  # 从连接池获取连接        count = 0        try:            # count : 为改变的数据条数            if param:                count = cursor.execute(sql, param)            else:                count = cursor.execute(sql)            conn.commit()            if autoclose:                self.close(cursor, conn)        except Exception as e:            pass        return cursor, conn, count    # 执行多条命令    # def executemany(self, lis):    #     """    #     :param lis: 是一个列表，里面放的是每个sql的字典'[{"sql":"xxx","param":"xx"}....]'    #     :return:    #     """    #     cursor, conn = self.db.getconn()    #     try:    #         for order in lis:    #             sql = order['sql']    #             param = order['param']    #             if param:    #                 cursor.execute(sql, param)    #             else:    #                 cursor.execute(sql)    #         conn.commit()    #         self.close(cursor, conn)    #         return True    #     except Exception as e:    #         print(e)    #         conn.rollback()    #         self.close(cursor, conn)    #         return False    # 释放连接    def close(self, cursor, conn):        """释放连接归还给连接池"""        cursor.close()        conn.close()    # 查询所有    def selectall(self, sql, param=None):        try:            cursor, conn, count = self.execute(sql, param)            res = cursor.fetchall()            return res        except Exception as e:            print(e)            self.close(cursor, conn)            return count    # 查询单条    def selectone(self, sql, param=None):        try:            cursor, conn, count = self.execute(sql, param)            res = cursor.fetchone()            self.close(cursor, conn)            return res        except Exception as e:            print("error_msg:", e.args)            self.close(cursor, conn)            return count    # 增加    def insertone(self, sql, param):        try:            cursor, conn, count = self.execute(sql, param)            # _id = cursor.lastrowid()  # 获取当前插入数据的主键id，该id应该为自动生成为好            conn.commit()            self.close(cursor, conn)            return count            # 防止表中没有id返回0            # if _id == 0:            #     return True            # return _id        except Exception as e:            print(e)            conn.rollback()            self.close(cursor, conn)            return count    # 增加多行    def insertmany(self, sql, param):        """        :param sql:        :param param: 必须是元组或列表[(),()]或（（），（））        :return:        """        cursor, conn, count = self.db.getconn()        try:            cursor.executemany(sql, param)            conn.commit()            return count        except Exception as e:            print(e)            conn.rollback()            self.close(cursor, conn)            return count    # 删除    def delete(self, sql, param=None):        try:            cursor, conn, count = self.execute(sql, param)            self.close(cursor, conn)            return count        except Exception as e:            print(e)            conn.rollback()            self.close(cursor, conn)            return count    # 更新    def update(self, sql, param=None):        try:            cursor, conn, count = self.execute(sql, param)            conn.commit()            self.close(cursor, conn)            return count        except Exception as e:            print(e)            conn.rollback()            self.close(cursor, conn)            return countif __name__ == '__main__':    db = MySqLHelper()    # # 查询单条    # sql1 = 'select * from userinfo where name=%s'    # args = 'python'    # ret = db.selectone(sql=sql1, param=args)    # print(ret)  # (None, b'python', b'123456', b'0')    # 增加单条    # sql2 = 'insert into userinfo (name,password) VALUES (%s,%s)'    # ret = db.insertone(sql2, ('old2','22222'))    # print(ret)    # 增加多条    # sql3 = 'insert into userinfo (name,password) VALUES (%s,%s)'    # li = li = [    #     ('分省', '123'),    #     ('到达','456')    # ]    # ret = db.insertmany(sql3,li)    # print(ret)    # 删除    # sql4 = 'delete from  userinfo WHERE name=%s'    # args = 'xxxx'    # ret = db.delete(sql4, args)    # print(ret)    # 更新    # sql5 = r'update userinfo set password=%s WHERE name LIKE %s'    # args = ('993333993', '%old%')    # ret = db.update(sql5, args)    # print(ret)

pip3 install DBUtils

#!/usr/bin/env python# -*- coding:utf-8 -*-import pymysql, os, configparserfrom pymysql.cursors import DictCursorfrom DBUtils.PooledDB import PooledDBclass Config(object):    """    # Config().get_content("user_information")    配置文件里面的参数    [notdbMysql]    host = 192.168.1.101    port = 3306    user = root    password = python123    """    def __init__(self, config_filename="myProjectConfig.cnf"):        file_path = os.path.join(os.path.dirname(__file__), config_filename)        self.cf = configparser.ConfigParser()        self.cf.read(file_path)    def get_sections(self):        return self.cf.sections()    def get_options(self, section):        return self.cf.options(section)    def get_content(self, section):        result = {}        for option in self.get_options(section):            value = self.cf.get(section, option)            result[option] = int(value) if value.isdigit() else value        return resultclass BasePymysqlPool(object):    def __init__(self, host, port, user, password, db_name=None):        self.db_host = host        self.db_port = int(port)        self.user = user        self.password = str(password)        self.db = db_name        self.conn = None        self.cursor = Noneclass MyPymysqlPool(BasePymysqlPool):    """    MYSQL数据库对象，负责产生数据库连接 , 此类中的连接采用连接池实现获取连接对象：conn = Mysql.getConn()            释放连接对象;conn.close()或del conn    """    # 连接池对象    __pool = None    def __init__(self, conf_name=None):        self.conf = Config().get_content(conf_name)        super(MyPymysqlPool, self).__init__(**self.conf)        # 数据库构造函数，从连接池中取出连接，并生成操作游标        self._conn = self.__getConn()        self._cursor = self._conn.cursor()    def __getConn(self):        """        @summary: 静态方法，从连接池中取出连接        @return MySQLdb.connection        """        if MyPymysqlPool.__pool is None:            __pool = PooledDB(creator=pymysql,                              mincached=1,                              maxcached=20,                              host=self.db_host,                              port=self.db_port,                              user=self.user,                              passwd=self.password,                              db=self.db,                              use_unicode=False,                              charset="utf8",                              cursorclass=DictCursor)        return __pool.connection()    def getAll(self, sql, param=None):        """        @summary: 执行查询，并取出所有结果集        @param sql:查询ＳＱＬ，如果有查询条件，请只指定条件列表，并将条件值使用参数[param]传递进来        @param param: 可选参数，条件列表值（元组/列表）        @return: result list(字典对象)/boolean 查询到的结果集        """        if param is None:            count = self._cursor.execute(sql)        else:            count = self._cursor.execute(sql, param)        if count > 0:            result = self._cursor.fetchall()        else:            result = False        return result    def getOne(self, sql, param=None):        """        @summary: 执行查询，并取出第一条        @param sql:查询ＳＱＬ，如果有查询条件，请只指定条件列表，并将条件值使用参数[param]传递进来        @param param: 可选参数，条件列表值（元组/列表）        @return: result list/boolean 查询到的结果集        """        if param is None:            count = self._cursor.execute(sql)        else:            count = self._cursor.execute(sql, param)        if count > 0:            result = self._cursor.fetchone()        else:            result = False        return result    def getMany(self, sql, num, param=None):        """        @summary: 执行查询，并取出num条结果        @param sql:查询ＳＱＬ，如果有查询条件，请只指定条件列表，并将条件值使用参数[param]传递进来        @param num:取得的结果条数        @param param: 可选参数，条件列表值（元组/列表）        @return: result list/boolean 查询到的结果集        """        if param is None:            count = self._cursor.execute(sql)        else:            count = self._cursor.execute(sql, param)        if count > 0:            result = self._cursor.fetchmany(num)        else:            result = False        return result    def insertMany(self, sql, values):        """        @summary: 向数据表插入多条记录        @param sql:要插入的ＳＱＬ格式        @param values:要插入的记录数据tuple(tuple)/list[list]        @return: count 受影响的行数        """        count = self._cursor.executemany(sql, values)        return count    def __query(self, sql, param=None):        if param is None:            count = self._cursor.execute(sql)        else:            count = self._cursor.execute(sql, param)        return count    def update(self, sql, param=None):        """        @summary: 更新数据表记录        @param sql: ＳＱＬ格式及条件，使用(%s,%s)        @param param: 要更新的  值 tuple/list        @return: count 受影响的行数        """        return self.__query(sql, param)    def insert(self, sql, param=None):        """        @summary: 更新数据表记录        @param sql: ＳＱＬ格式及条件，使用(%s,%s)        @param param: 要更新的  值 tuple/list        @return: count 受影响的行数        """        return self.__query(sql, param)    def delete(self, sql, param=None):        """        @summary: 删除数据表记录        @param sql: ＳＱＬ格式及条件，使用(%s,%s)        @param param: 要删除的条件 值 tuple/list        @return: count 受影响的行数        """        return self.__query(sql, param)    def begin(self):        """        @summary: 开启事务        """        self._conn.autocommit(0)    def end(self, option='commit'):        """        @summary: 结束事务        """        if option == 'commit':            self._conn.commit()        else:            self._conn.rollback()    def dispose(self, isEnd=1):        """        @summary: 释放连接池资源        """        if isEnd == 1:            self.end('commit')        else:            self.end('rollback')        self._cursor.close()        self._conn.close()if __name__ == '__main__':    mysql = MyPymysqlPool("notdbMysql")    sqlAll = "select * from myTest.aa;"    result = mysql.getAll(sqlAll)    print(result)    sqlAll = "select * from myTest.aa;"    result = mysql.getMany(sqlAll, 2)    print(result)    result = mysql.getOne(sqlAll)    print(result)    # mysql.insert("insert into myTest.aa set a=%s", (1))    # 释放资源    mysql.dispose()