Python+Pymysql+PooledDB实现数据库连接池

1. 引言
在测试中，频繁地创建和销毁数据库连接会消耗大量的资源，并且可能导致数据库连接的泄漏大概性能下降。因此，利用连接池可以有用地管理数据库连接，提高步伐的性能和可靠性。
2. 实现MySQL连接池的类
我们将利用Python的pymysql库和dbutils库中的PooledDB来实现MySQL连接池。

• 终端输入下令安装两个库
  

1. pip install pymysql
2. pip install dbutils

复制代码

• 以下是实现连接池的主要类ConnectMysqlPool的详细先容：
  

1. import pymysql
2. from dbutils.pooled_db import PooledDB
3. import os
4. from common_methods.file_operation import f_p
6. class ConnectMysqlPool:
7.     """
8.     连接MySQL数据库的连接池类。
10.     属性:
11.     db_account (dict): 数据库账号信息，包括用户名和密码等。
12.     db (str): 数据库名称。
13.     pool (PooledDB): MySQL连接池对象。
15.     方法:
16.     __init__: 初始化连接池类实例。
17.     _obtaining_data: 从配置文件中获取测试数据。
18.     create_mysql_pool: 创建MySQL连接池。
19.     get_conn: 从连接池中获取一个连接。
20.     close: 关闭数据库连接和游标。
21.     execute: 使用连接执行SQL语句。
22.     """
23.     def __init__(self, db, db_account):
24.         """
25.         初始化连接池类实例。
27.         参数:
28.         db (str): 测试库名称。
29.         db_account (dict): 包含数据库账号信息的字典。
30.         """
31.         self._common_data = self._obtaining_data() # 获取测试数据
32.         self.db_account = db_account # 数据库账号信息
33.         self.db = db # 测试库
35.         # 创建连接池
36.         self.pool = self.create_mysql_pool()
38.     # 从配置文件中获取测试数据
39.     def _obtaining_data(self):
40.         """
41.         从指定的YAML文件中读取连接配置数据。
43.         返回:
44.         dict: 包含连接配置信息的字典。
45.         """
46.         file_path = os.path.join(
47.             os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
48.             "data_driven",
49.             "mysql_data.yaml"
50.         )
51.         common_data = f_p.red_yaml_file(file_path)
52.         return common_data
54.     # 创建MySQL连接池
55.     def create_mysql_pool(self):
56.         """
57.         根据配置信息创建MySQL连接池。
59.         返回:
60.         PooledDB: MySQL连接池对象。
61.         """
62.         pool = PooledDB(
63.             **self._common_data['connect_pool_config'],
64.             **self.db_account,
65.             db=self.db,
66.             creator=pymysql
67.         )
68.         return pool
70.     # 从连接池中获取一个连接
71.     def get_conn(self):
72.         """
73.         从连接池中获取一个数据库连接。
75.         返回:
76.         connection: 数据库连接对象。
77.         """
78.         return self.pool.connection()
80.     # 关闭数据库连接和游标
81.     def close(self, conn, cursor):
82.         """
83.         关闭数据库连接和游标。
85.         参数:
86.         conn (connection): 数据库连接对象。
87.         cursor (cursor): 数据库游标对象。
88.         """
89.         try:
90.             if cursor:
91.                 cursor.close()
92.             if conn:
93.                 conn.close()
94.         except Exception as e:
95.             print(f"关闭连接或游标时发成错误：{e}")
97.     # 使用连接执行sql
98.     def execute(self, sql, params):
99.         """
100.         使用获取的连接执行SQL语句。
102.         参数:
103.         sql (str): SQL语句。
104.         params (tuple): SQL参数。
106.         返回:
107.         list: 执行SQL语句后的结果集，若执行出错则返回None。
108.         """
109.         conn = self.get_conn()
110.         cursor = conn.cursor()
112.         try:
113.             cursor.execute(sql, params)
114.             conn.commit()
115.             result = cursor.fetchall()
116.             return result
117.         except Exception as e:
118.             print(f"表：{self.db}，执行sql：{sql}，报错：{e}")
119.             conn.rollback()
120.             return None
121.         finally:
122.             self.close(conn, cursor)

复制代码

• red_yaml_file：读取yaml文件方法，代码在下方
  

1. def red_yaml_file(self,file_path,name=None):
3.     try:
4.         with open(file_path,'r') as file:
5.             data = yaml.safe_load(file)
6.             return_data = data if name is None else data.get(name,{}) # 返回数据
7.             return return_data
8.     except Exception as e:
9.         print(f"{file_path}读取数据失败：{e}")
10.         return None

复制代码

• mysql_data.yaml：yaml文件写入的连接池设置数据
  

1. connect_pool_config:
2.   charset: "utf8"
3.   mincached: 2 # 初始化时，连接池中至少创建空闲的链接，0表示不创建
4.   maxcached: 50 # 连接池允许最大的连接数， 0和None表示不限制连接数
5.   maxshared: 0 # 连接池最多可共享的连接数量，0和None表示全部共享。PS：pymysql不支持事务
6.   maxconnections: 10 # 连接池最大并发连接数量
7.   blocking: True # 连接池中没有可用连接后，是否阻塞等待
8.   setsession: [] # 开始会话前执行的命令列表

复制代码

3. 利用示例
以下是利用ConnectMysqlPool类的示例代码：

1. # 创建连接池对象
2. db_account = {
3.     "host": "localhost",
4.     "port": 3306,
5.     "user": "root",
6.     "password": "password",
7.     "charset": "utf8mb4"
8. }
9. db_name = "test_db"
11. cmp = ConnectMysqlPool(db_name,db_account)
13. sql = "select * from table_name where id < %s"
14. params = (100,)
15. result = cmp.execute(sql,params) # 返回表中id小于100的数据

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。