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

2025/3/17 20:58:04 来源：https://blog.csdn.net/qq_41522951/article/details/145110487 浏览: 次关键词：自动化测试框架搭建-数据库连接池-方法改为类

为什么改为类形式？

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

如何解决？

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

class A：def __init__(self):xxxdef __func__(self):xxxa = A()

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

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

代码实现

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

import pymysql
# 数据库连接配置信息
from dbutils.persistent_db import PersistentDB
from dbutils.pooled_db import PooledDBfrom conf.dbConfig import getMysqlSettingconfig = 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