Python3数据库及连接池操作

淡淡的烟草味﹌ 2022-03-12 19:18 338阅读 0赞

### 一、PyMySQL 操作 MySQL ###

PyMySQL是一个纯 Python 实现的 MySQL 客户端操作库，支持事务、存储过程、批量执行等。

PyMySQL 遵循 Python 数据库 API v2.0 规范，并包含了 pure-Python MySQL 客户端库。

**安装：**

pip3 install PyMySQL

**demo：**

PyMySQL.py

#!/usr/bin/python3
    
    import pymysql
    
    # 打开数据库连接
    db = pymysql.connect("localhost", "root", "haoxin963", "pig")
    
    # 使用 cursor() 方法创建一个游标对象 cursor
    cursor = db.cursor()
    
    # 使用 execute()  方法执行 SQL 查询
    cursor.execute("SELECT * FROM sys_role")
    
    # 使用 fetchall() 方法获取s所有数据.
    data = cursor.fetchall()
    
    print(data)
    
    # 关闭数据库连接
    db.close()

### 二、DBUtils实现数据库连接池 ###

DBUtils是一套Python数据库连接池包，并允许对非线程安全的数据库接口进行线程安全包装。DBUtils来自Webware for Python。

DBUtils提供两种外部接口：

*  PersistentDB ：提供线程专用的数据库连接，并自动管理连接。
 *  PooledDB ：提供线程间可共享的数据库连接，并自动管理连接。

**安装：**

pip3 install DBUtils

**demo：**

mysql_DBUtils.py

#!/usr/bin/python3
    # -*- coding:utf-8 -*-
    import pymysql
    import os
    import configparser
    from pymysql.cursors import DictCursor
    from DBUtils.PooledDB import PooledDB
    
    
    class Config(object):
        """
        # Config().get_content("user_information")
        配置文件里面的参数
        [dbMysql]
        host = 192.168.1.180
        port = 3306
        user = root
        password = 123456
        """
    
        def __init__(self, config_filename="dbMysqlConfig.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 result
    
    
    class BasePymysqlPool(object):
        def __init__(self, host, port, user, password, db_name):
            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 = None
    
    
    class 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=True,
                                  charset="utf8",
                                  cursorclass=DictCursor)
                print("12211212")
            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("dbMysql")
        sqlAll = "select * from sys_role;"
        result = mysql.getAll(sqlAll)
        print(result)
        # 释放资源
        mysql.dispose()

**配置文件：**

dbMysqlConfig.cnf

[dbMysql]
    host = localhost
    port = 3306
    user = root
    password = haoxin963
    db_name = pig