自动化测试框架搭建-数据库连接池-方法改为类

news2025/1/16 1:31:41

为什么改为类形式？

前序的方法创建连接池，依旧存在该方法被多次调用，创建多个连接池的风险
解决方法：使用类的形式创建，提前实例化一个对象，都使用一个对象即可

如何解决？

1、使用单例模式封装连接池（python自带的特性）

class A：
	def __init__(self):
		xxx
	def __func__(self):
		xxx

a = A()

2、其他包倒入这个实例a，使用实例a提供的方法即可

3、以后对数据库调用都是用已经创建好的实例，可以保证使用的是同一个连接池

代码实现

一、将数据库池的创建，放到类的初始化里

import pymysql
# 数据库连接配置信息
from dbutils.persistent_db import PersistentDB
from dbutils.pooled_db import PooledDB

from conf.dbConfig import getMysqlSetting

config = getMysqlSetting()


class DbConnectionPool:
    def __init__(self):
        self.pool = PooledDB(
            # 指定数据库连接驱动
            creator=pymysql,
            # 连接池允许的最大连接数,0和None表示没有限制
            maxconnections=20,
            # 初始化时,连接池至少创建的空闲连接,0表示不创建
            mincached=2,
            # 连接池中空闲的最多连接数,0和None表示没有限制
            maxcached=5,
            # # 连接池中最多共享的连接数量,0和None表示全部共享
            # maxshared=3,
            # 连接池中如果没有可用共享连接后,是否阻塞等待,True表示等等,
            # False表示不等待然后报错
            blocking=True,
            # 开始会话前执行的命令列表
            setsession=[],
            # ping Mysql服务器检查服务是否可用
            ping=0,
            **config
        )

    def create_conn(self):
        # 从数据库连接池中取出一条连接
        conn = self.pool.connection()
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        return conn, cursor

    # 关闭连接
    def close_conn(self, cursor):
        self.close()
        cursor.close()


dbconnectionpool = DbConnectionPool()

二、将调用实例方法，完成简单数据查询方法封装

from DButil.DbConnectionPool import dbconnectionpool
create_conn = dbconnectionpool.create_conn()
close_conn = dbconnectionpool.close_conn()


# 查询一条
def select_one(sql, args):
    conn, cur = create_conn()
    cur.execute(sql, args)
    result = cur.fetchone()
    close_conn(conn, cur)
    return result


# 根据条件查询所有
def select_all(sql, args):
    conn, cur = create_conn()
    cur.execute(sql, args)
    result = cur.fetchall()
    close_conn(conn, cur)
    return result


# 新增一条记录
def insert_one(sql, args):
    conn, cur = create_conn()
    result = cur.execute(sql, args)
    conn.commit()
    close_conn(conn, cur)
    return result


# 新增一条纪录 并返回主键id
def insert_one_pk(sql, args):
    conn, cur = create_conn()
    result = cur.execute(sql, args)
    conn.commit()
    close_conn(conn, cur)
    return cur.lastrowid


# 删除一条记录
def delete_one(sql, args):
    conn, cur = create_conn()
    result = cur.execute(sql, args)
    conn.commit()
    close_conn(conn, cur)
    return result


# 更新一条记录
def update_one(sql, args):
    conn, cur = create_conn()
    result = cur.execute(sql, args)
    conn.commit()
    close_conn(conn, cur)
    return result