Python MySQL数据库连接池组件封装

数据库封装类MySQLConnection.py,代码如下：

# -*- coding: utf-8 -*-
# @Author : 李
# @Time : 2020/12/15 14:33
import osimport pymysql
import logging
import configparserfrom dbutils.pooled_db import PooledDB"""数据库连接池相关
"""class MySQLConnection(object):"""数据库连接池代理对象查询参数主要有两种类型第一种：传入元祖类型,例如(12,13),这种方式主要是替代SQL语句中的%s展位符号第二种: 传入字典类型,例如{"id":13},此时我们的SQL语句需要使用键来代替展位符,例如：%(name)s"""def __init__(self,dbName="master"):# 读取数据库配置信息config = configparser.ConfigParser()path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'db.conf')config.read(path, encoding='UTF-8')sections = config.sections()# 数据库工厂dbFactory = {}for dbName in sections:# 读取相关属性maxconnections = config.get(dbName, "maxconnections")mincached = config.get(dbName, "mincached")maxcached = config.get(dbName, "maxcached")host = config.get(dbName, "host")port = config.get(dbName, "port")user = config.get(dbName, "user")password = config.get(dbName, "password")database = config.get(dbName, "database")databasePooled = PooledDB(creator=pymysql,maxconnections=int(maxconnections),mincached=int(mincached),maxcached=int(maxcached),blocking=True,cursorclass=pymysql.cursors.DictCursor,host=host,port=int(port),user=user,password=password,database=database)dbFactory[dbName] = databasePooledself.connect = dbFactory[dbName].connection()self.cursor = self.connect.cursor()logging.debug("获取数据库连接对象成功,连接池对象:{}".format(str(self.connect)))def execute(self,sql,param=None):"""基础更新、插入、删除操作:param sql::param param::return: 受影响的行数"""ret=Nonetry:if param==None:ret=self.cursor.execute(sql)else:ret=self.cursor.execute(sql,param)except TypeError as te:logging.debug("类型错误")logging.exception(te)return retdef query(self,sql,param=None):"""查询数据库:param sql: 查询SQL语句:param param: 参数:return: 返回集合"""self.cursor.execute(sql,param)result=self.cursor.fetchall()return resultdef queryOne(self,sql,param=None):"""查询数据返回第一条:param sql: 查询SQL语句:param param: 参数:return: 返回第一条数据的字典"""result=self.query(sql,param)if result:return result[0]else:return Nonedef listByPage(self,sql,current_page,page_size,param=None):"""分页查询当前表格数据:param sql: 查询SQL语句:param current_page: 当前页码:param page_size: 页码大小:param param:参数:return:"""countSQL="select count(*) ct from ("+sql+") tmp "logging.debug("统计SQL:{}".format(sql))countNum=self.count(countSQL,param)offset=(current_page-1)*page_sizetotalPage=int(countNum/page_size)if countNum % page_size>0:totalPage = totalPage + 1pagination={"current_page":current_page,"page_size":page_size,"count":countNum,"total_page":totalPage}querySql="select * from ("+sql+") tmp limit %s,%s"logging.debug("查询SQL:{}".format(querySql))# 判断是否有参数if param==None:# 无参数pagination["data"]=self.query(querySql,(offset,page_size))else:# 有参数的情况,此时需要判断参数是元祖还是字典if isinstance(param,dict):# 字典的情况,因此需要添加字典querySql="select * from ("+sql+") tmp limit %(tmp_offset)s,%(tmp_pageSize)s"param["tmp_offset"]=offsetparam["tmp_pageSize"]=page_sizepagination["data"]=self.query(querySql,param)elif isinstance(param,tuple):# 元祖的方式listtp=list(param)listtp.append(offset)listtp.append(page_size)pagination["data"]=self.query(querySql,tuple(listtp))else:# 基础类型listtp=[]listtp.append(param)listtp.append(offset)listtp.append(page_size)pagination["data"]=self.query(querySql,tuple(listtp))return paginationdef count(self,sql,param=None):"""统计当前表记录行数:param sql: 统计SQL语句:param param: 参数:return: 当前记录行"""ret=self.queryOne(sql,param)count=Noneif ret:for k,v in ret.items():count=vreturn countdef insert(self,sql,param=None):"""数据库插入:param sql: SQL语句:param param: 参数:return: 受影响的行数"""return self.execute(sql,param)def update(self,sql,param=None):"""更新操作:param sql: SQL语句:param param: 参数:return: 受影响的行数"""return self.execute(sql,param)def delete(self,sql,param=None):"""删除操作:param sql: 删除SQL语句:param param: 参数:return: 受影响的行数"""return self.execute(sql,param)def batch(self,sql,param=None):"""批量插入:param sql: 插入SQL语句:param param: 参数:return: 受影响的行数"""return self.cursor.executemany(sql,param)def commit(self,param=None):"""提交数据库:param param::return:"""if param==None:self.connect.commit()else:self.connect.rollback()def close(self):"""关闭数据库连接:return:"""if self.cursor:self.cursor.close()if self.connect:self.connect.close()logging.debug("释放数据库连接")return None

首先我们从配置文件中读取数据库的配置信息,放入dbFactory字典中,一个数据源key对应一个数据库连接池对象

使用
封装好我们的数据库组件后,接下来我们对于数据库的操作就很方便了

注意事项
我们在写SQL语句的时候需要注意我们的参数占位符的使用,注意有两种方式

%s：该方式你可以理解为任意字符,但是我们在参数传递的时候顺序必须正确,否则就会出现结果不一致的情况
%(fied)s:这种方式相对友好一些,我们传入的参数可以是字典对象,而不用关心我们的占位参数是否一致
获取不同的数据源连接
我们上面提到过,我们需要的是一个多数据源的情况,那么我们代码中应该如何使用呢?

# 第一种方式,获取默认数据库的连接,在配置文件中配置名称为master的数据库
connect=MySQLConnect()# 第二种方式,指定key名称获取数据库连接,如下：获取local111的数据库连接
connect=MySQLConnect("local111")

新增数据

新增数据

def insertByIndex():"""使用下标索引的方式插入数据"""sql="insert into user(id,name) values(%s,%s)"# 获取数据库连接connect=MySQLConnection()try:# 执行插入动作connect.insert(sql,("1","张三"))# 提交connect.commit()except Exception as e:logging.exception(e)finally:# 关闭数据库连接connect.close()def insertBatchByIndex():"""使用下标索引的方式批量插入数据"""sql="insert into user(id,name) values(%s,%s)"# 获取数据库连接connect=MySQLConnection()try:# 执行批量插入动作data=[]# 放入的是元祖data.append(("1","张三"))data.append(("2","张三2"))connect.batch(sql,data)# 提交connect.commit()except Exception as e:logging.exception(e)finally:# 关闭数据库连接connect.close()

看了以上的方式或许你会觉得麻烦,为什么呢,因为我们在开发的时候一般都是使用字典居多,都是key-value的方式,因此使用索引的方式需要我们再申明一个有顺序的元祖对象,很麻烦

使用字典的方式如下：

def insertByDict():"""使用下标索引的方式插入数据"""sql="insert into user(id,name) values(%(id)s,%(name)s)"# 获取数据库连接connect=MySQLConnection()try:# 执行插入动作# 此时我们使用的是字典connect.insert(sql,{"name":"张三","id":"1"})# 提交,必须connect.commit()except Exception as e:logging.exception(e)finally:# 关闭数据库连接,必须connect.close()def insertBatchByDict():"""使用下标索引的方式批量插入数据"""sql="insert into user(id,name) values(%(id)s,%(name)s)"# 获取数据库连接connect=MySQLConnection()try:# 执行批量插入动作data=[]# 放入的是自字典data.append({"name":"张三","id":"1"})data.append({"name":"张三1","id":"2"})connect.batch(sql,data)# 提交connect.commit()except Exception as e:logging.exception(e)finally:# 关闭数据库连接connect.close()

以上就是新增的方式,删除和修改同理也差不多.

比如分页查询的示例：

sql="select id,name,address from restaurant where province=%s "
connect=MySQLConnection()
try:page=1size=500pagination=connect.listByPage(sql,page,size,'浙江省')
except Exception as e:logging.exception(e)
finally:# 关闭数据库连接connect.close()

分页查询返回分页对象,是一个字典对象，主要包含属性：

current_page:当前页码
page_size:当前页码大小
count:当前统计SQL语句的表记录总数
total_page:当前总页码数
data:当前的数据集合,是一个字典集合对象
不分页的查询返回的是data集合字典对象，和分页使用方法类似

总结
以上就是一个简单的Python-MySQL数据库连接池组件,该组件同数据库交互过百万次,非常稳定,值得信赖~~!!!

附录
附上相关组件的说明文档：

https://cito.github.io/DBUtils/
https://pymysql.readthedocs.io/en/latest/modules/cursors.html

转载地址：https://blog.csdn.net/u010192145/article/details/102487255