python DbUtils 封装

article/2025/10/2 23:35:55

python dbutils 简介及准备工作

dbutils封装文件传送门

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

DBUtils提供两种外部接口：

  • PersistentDB ：提供线程专用的数据库连接，并自动管理连接。
  • PooledDB ：提供线程间可共享的数据库连接，并自动管理连接。
    需要库
    1、DBUtils pip install DBUtils
    2、pymysql pip install pymysql/MySQLdb

创建DButils组件

db_config.py 配置文件

# -*- coding: UTF-8 -*-
import pymysql# 数据库信息
DB_TEST_HOST = "127.0.0.1"
DB_TEST_PORT = 3306
DB_TEST_DBNAME = "ball"
DB_TEST_USER = "root"
DB_TEST_PASSWORD = "123456"# 数据库连接编码
DB_CHARSET = "utf8"# mincached : 启动时开启的闲置连接数量(缺省值 0 开始时不创建连接)
DB_MIN_CACHED = 10# maxcached : 连接池中允许的闲置的最多连接数量(缺省值 0 代表不闲置连接池大小)
DB_MAX_CACHED = 10# maxshared : 共享连接数允许的最大数量(缺省值 0 代表所有连接都是专用的)如果达到了最大数量,被请求为共享的连接将会被共享使用
DB_MAX_SHARED = 20# maxconnecyions : 创建连接池的最大数量(缺省值 0 代表不限制)
DB_MAX_CONNECYIONS = 100# blocking : 设置在连接池达到最大数量时的行为(缺省值 0 或 False 代表返回一个错误<toMany......> 其他代表阻塞直到连接数减少,连接被分配)
DB_BLOCKING = True# maxusage : 单个连接的最大允许复用次数(缺省值 0 或 False 代表不限制的复用).当达到最大数时,连接会自动重新连接(关闭和重新打开)
DB_MAX_USAGE = 0# setsession : 一个可选的SQL命令列表用于准备每个会话，如["set datestyle to german", ...]
DB_SET_SESSION = None# creator : 使用连接数据库的模块
DB_CREATOR = pymysql
db_dbutils_init.py 创建数据池初始化
from DBUtils.PooledDB import PooledDB
import db_config as config"""
@功能：创建数据库连接池
"""class MyConnectionPool(object):__pool = None# def __init__(self):#     self.conn = self.__getConn()#     self.cursor = self.conn.cursor()# 创建数据库连接conn和游标cursordef __enter__(self):self.conn = self.__getconn()self.cursor = self.conn.cursor()# 创建数据库连接池def __getconn(self):if self.__pool is None:self.__pool = PooledDB(creator=config.DB_CREATOR,mincached=config.DB_MIN_CACHED,maxcached=config.DB_MAX_CACHED,maxshared=config.DB_MAX_SHARED,maxconnections=config.DB_MAX_CONNECYIONS,blocking=config.DB_BLOCKING,maxusage=config.DB_MAX_USAGE,setsession=config.DB_SET_SESSION,host=config.DB_TEST_HOST,port=config.DB_TEST_PORT,user=config.DB_TEST_USER,passwd=config.DB_TEST_PASSWORD,db=config.DB_TEST_DBNAME,use_unicode=False,charset=config.DB_CHARSET)return self.__pool.connection()# 释放连接池资源def __exit__(self, exc_type, exc_val, exc_tb):self.cursor.close()self.conn.close()# 关闭连接归还给链接池# def close(self):#     self.cursor.close()#     self.conn.close()# 从连接池中取出一个连接def getconn(self):conn = self.__getconn()cursor = conn.cursor()return cursor, conn# 获取连接池,实例化
def get_my_connection():return MyConnectionPool()

制作mysqlhelper.py

from db_dbutils_init import get_my_connection"""执行语句查询有结果返回结果没有返回0；增/删/改返回变更数据条数，没有返回0"""class MySqLHelper(object):def __init__(self):self.db = get_my_connection()  # 从数据池中获取连接def __new__(cls, *args, **kwargs):if not hasattr(cls, 'inst'):  # 单例cls.inst = super(MySqLHelper, cls).__new__(cls, *args, **kwargs)return cls.inst# 封装执行命令def execute(self, sql, param=None, autoclose=False):"""【主要判断是否有参数和是否执行完就释放连接】:param sql: 字符串类型，sql语句:param param: sql语句中要替换的参数"select %s from tab where id=%s" 其中的%s就是参数:param autoclose: 是否关闭连接:return: 返回连接conn和游标cursor"""cursor, conn = self.db.getconn()  # 从连接池获取连接count = 0try:# count : 为改变的数据条数if param:count = cursor.execute(sql, param)else:count = cursor.execute(sql)conn.commit()if autoclose:self.close(cursor, conn)except Exception as e:passreturn cursor, conn, count# 执行多条命令# def executemany(self, lis):#     """#     :param lis: 是一个列表，里面放的是每个sql的字典'[{"sql":"xxx","param":"xx"}....]'#     :return:#     """#     cursor, conn = self.db.getconn()#     try:#         for order in lis:#             sql = order['sql']#             param = order['param']#             if param:#                 cursor.execute(sql, param)#             else:#                 cursor.execute(sql)#         conn.commit()#         self.close(cursor, conn)#         return True#     except Exception as e:#         print(e)#         conn.rollback()#         self.close(cursor, conn)#         return False# 释放连接def close(self, cursor, conn):"""释放连接归还给连接池"""cursor.close()conn.close()# 查询所有def selectall(self, sql, param=None):try:cursor, conn, count = self.execute(sql, param)res = cursor.fetchall()return resexcept Exception as e:print(e)self.close(cursor, conn)return count# 查询单条def selectone(self, sql, param=None):try:cursor, conn, count = self.execute(sql, param)res = cursor.fetchone()self.close(cursor, conn)return resexcept Exception as e:print("error_msg:", e.args)self.close(cursor, conn)return count# 增加def insertone(self, sql, param):try:cursor, conn, count = self.execute(sql, param)# _id = cursor.lastrowid()  # 获取当前插入数据的主键id，该id应该为自动生成为好conn.commit()self.close(cursor, conn)return count# 防止表中没有id返回0# if _id == 0:#     return True# return _idexcept Exception as e:print(e)conn.rollback()self.close(cursor, conn)return count# 增加多行def insertmany(self, sql, param):""":param sql::param param: 必须是元组或列表[(),()]或（（），（））:return:"""cursor, conn, count = self.db.getconn()try:cursor.executemany(sql, param)conn.commit()return countexcept Exception as e:print(e)conn.rollback()self.close(cursor, conn)return count# 删除def delete(self, sql, param=None):try:cursor, conn, count = self.execute(sql, param)self.close(cursor, conn)return countexcept Exception as e:print(e)conn.rollback()self.close(cursor, conn)return count# 更新def update(self, sql, param=None):try:cursor, conn, count = self.execute(sql, param)conn.commit()self.close(cursor, conn)return countexcept Exception as e:print(e)conn.rollback()self.close(cursor, conn)return countif __name__ == '__main__':db = MySqLHelper()# # 查询单条# sql1 = 'select * from userinfo where name=%s'# args = 'python'# ret = db.selectone(sql=sql1, param=args)# print(ret)  # (None, b'python', b'123456', b'0')# 增加单条# sql2 = 'insert into userinfo (name,password) VALUES (%s,%s)'# ret = db.insertone(sql2, ('old2','22222'))# print(ret)# 增加多条# sql3 = 'insert into userinfo (name,password) VALUES (%s,%s)'# li = li = [#     ('分省', '123'),#     ('到达','456')# ]# ret = db.insertmany(sql3,li)# print(ret)# 删除# sql4 = 'delete from  userinfo WHERE name=%s'# args = 'xxxx'# ret = db.delete(sql4, args)# print(ret)# 更新# sql5 = r'update userinfo set password=%s WHERE name LIKE %s'# args = ('993333993', '%old%')# ret = db.update(sql5, args)# print(ret)

python3 实现mysql数据库连接池

首先声明一下，这篇博客进行了通过自己的代码方式，加上这篇博客，最后总结出这段代码。参考博客连接：http://blog.csdn.net/zbc1090549839/article/details/51336458

原理

python编程中可以使用MySQLdb进行数据库的连接及诸如查询/插入/更新等操作，但是每次连接mysql数据库请求时，都是独立的去请求访问，相当浪费资源，

而且访问数量达到一定数量时，对mysql的性能会产生较大的影响。

因此，实际使用中，通常会使用数据库的连接池技术，来访问数据库达到资源复用的目的。

安装数据库连接池模块DBUtils

pip3 install DBUtils

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

DBUtils提供两种外部接口：
  • * PersistentDB ：提供线程专用的数据库连接，并自动管理连接。
  • * PooledDB ：提供线程间可共享的数据库连接，并自动管理连接。

下载地址：DBUtils   下载解压后，使用python setup.py install 命令进行安装

下面利用MySQLdb和DBUtils建立自己的mysql数据库连接池工具包

在工程目录下新建package命名为:dbConnecttion，并新建module命名为MySqlConn，下面是MySqlConn.py，该模块创建Mysql的连接池对象，并创建了如查询/插入等通用的操作方法。该部分代码实现如下：

还有很多其他参数可以配置：

dbapi ：数据库接口
mincached ：启动时开启的空连接数量
maxcached ：连接池最大可用连接数量
maxshared ：连接池最大可共享连接数量
maxconnections ：最大允许连接数量
blocking ：达到最大数量时是否阻塞
maxusage ：单个连接最大复用次数
根据自己的需要合理配置上述的资源参数，以满足自己的实际需要。

代码：
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pymysql, os, configparser
from pymysql.cursors import DictCursor
from DBUtils.PooledDB import PooledDBclass Config(object):"""# Config().get_content("user_information")配置文件里面的参数[notdbMysql]host = 192.168.1.101port = 3306user = rootpassword = python123"""def __init__(self, config_filename="myProjectConfig.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 valuereturn resultclass BasePymysqlPool(object):def __init__(self, host, port, user, password, db_name=None):self.db_host = hostself.db_port = int(port)self.user = userself.password = str(password)self.db = db_nameself.conn = Noneself.cursor = Noneclass MyPymysqlPool(BasePymysqlPool):"""MYSQL数据库对象，负责产生数据库连接 , 此类中的连接采用连接池实现获取连接对象：conn = Mysql.getConn()释放连接对象;conn.close()或del conn"""# 连接池对象__pool = Nonedef __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=False,charset="utf8",cursorclass=DictCursor)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 = Falsereturn resultdef 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 = Falsereturn resultdef 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 = Falsereturn resultdef insertMany(self, sql, values):"""@summary: 向数据表插入多条记录@param sql:要插入的ＳＱＬ格式@param values:要插入的记录数据tuple(tuple)/list[list]@return: count 受影响的行数"""count = self._cursor.executemany(sql, values)return countdef __query(self, sql, param=None):if param is None:count = self._cursor.execute(sql)else:count = self._cursor.execute(sql, param)return countdef 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("notdbMysql")sqlAll = "select * from myTest.aa;"result = mysql.getAll(sqlAll)print(result)sqlAll = "select * from myTest.aa;"result = mysql.getMany(sqlAll, 2)print(result)result = mysql.getOne(sqlAll)print(result)# mysql.insert("insert into myTest.aa set a=%s", (1))# 释放资源mysql.dispose()

 


http://chatgpt.dhexx.cn/article/4qI4UGkJ.shtml

相关文章

DBUtils工具

DBUtils工具

DBUtils工具 为了更加简单地使用JDBC&#xff0c;Apache组织提供了一个工具类库commons-dbutils&#xff0c;它是操作数据库地一个组件&#xff0c;实现一个对JDBC的简单封装&#xff0c;可以在不影响性能的情况下极大地简化JDBC地编码工作量。 API介绍 commons-dbutils的核…
阅读更多...
java dbutils_Java学习之DBUtils

java dbutils_Java学习之DBUtils

DBUtils&#xff1a;只是帮我们简化了CRUD 的代码。数据库连接的创建以及获取工作&#xff0c;不在其工作范围 一、DBUtils应用 ①、DBUtils的jar文件的下载 ②、Sql Server 、C3P0、DBUtils环境搭建(导入相应jar文件) Sql Server &#xff1a;sqljdbc42.jar(JDBC驱动) C3P0&am…
阅读更多...
DBUtils介绍与使用

DBUtils介绍与使用

目录 一、DBUtils介绍二、DBUtils使用&#xff08;一&#xff09;导入相关Jar包&#xff08;二&#xff09;使用DBUtils完成数据库操作 一、DBUtils介绍 DBUtils&#xff1a;是为了简化JDBC的快发而产生的开发工具包&#xff0c;是对JDBC的一种封装。 DBUtils是个小巧的JDBC轻…
阅读更多...
DBUtils详解

DBUtils详解

目录 DBUtils查询数据 匿名内部类方式 反射方式 反射方式查询数据的触发器分类 多表查询 DBUtils增加,删除,修改数据 前言 使用DBUtils操作数据库,会更加的方便,因为它封装了JDBC的代码,这里我们配合c3p0使用 并不是说一定要用c3p0,因为使用DBUtils 的时候需要手动传递…
阅读更多...
JavaWeb——（13）DBUtils

JavaWeb——（13）DBUtils

目录 一、DBUtils介绍 二、DBUtils核心对象 三、DBUtils使用简单案例 四、QueryRunner对象 4.1构造函数 4.2方法 五、ResultSetHandler接口 六、DBUtils控制事务 6.1ThreadLocal 6.2模拟转账业务 一、DBUtils介绍 DBUtils是java编程中的数据库操作实用工具&#xff…
阅读更多...
DBUtils工具类的使用方法详解

DBUtils工具类的使用方法详解

DBUtils使用方法详解 目录 DBUtils使用方法详解一、前言二、JDBC介绍1.基本概念2.JDBC访问数据库的流程 三、DBUtils介绍1.基本概念2.配置文件3.创建JDBCUtils类4.实现对数据表的增删改查 四、对以上代码的说明五、总结 一、前言 本文是关于DBUtils使用方法的介绍&#xff0c;…
阅读更多...
【优化笔记7】KKT条件

【优化笔记7】KKT条件

阅读更多...
如何用计算机解kkt条件,【直观详解】拉格朗日乘法和KKT条件

如何用计算机解kkt条件,【直观详解】拉格朗日乘法和KKT条件

【阅读时间】8min - 10mun 【内容简介】直观的解读了什么是拉格朗日乘子法，以及如何求解拉格朗日方程，并且给出几个直观的例子，针对不等式约束解读了KKT条件的必要条件和充分条件 What & Why 拉格朗日乘法(Lagrange multiplier)是一种在最优化的问题中寻找多元函数在其变…
阅读更多...
SVM中的KKT条件和拉格朗日对偶

SVM中的KKT条件和拉格朗日对偶

首先&#xff0c;我们要理解KKT条件是用来干嘛的&#xff1f; KKT条件用来判断一个解是否属于一个非线性优化问题。 求最优解&#xff1a; 约束条件分为 1、等式约束 2、不等式约束 对于等式约束的优化问题&#xff0c;可以直接应用拉格朗日乘子法去求取最优值&#xff1b; …
阅读更多...
拉格朗日乘法和KKT条件

拉格朗日乘法和KKT条件

附件&#xff1a;KKT
阅读更多...
KKT条件详解

KKT条件详解

KKT条件详解 主要参考这篇文章和这个知乎回答。 KKT最优化条件是Karush[1939]&#xff0c;以及Kuhn和Tucker[1951]先后独立发表出來的。这组最优化条件在Kuhn和Tucker发表之后才逐渐受到重视&#xff0c;因此许多情况下只记载成库恩塔克条件&#xff08;Kuhn-Tucker conditions…
阅读更多...
KKT条件总结

KKT条件总结

最近学习的时候用到了最优化理论&#xff0c;但是我没有多少这方面的理论基础。于是翻了很多大神的博客把容易理解的内容记载到这篇博客中。因此这是篇汇总博客&#xff0c;不算是全部原创&#xff0c;但是基础理论&#xff0c;应该也都差不多吧。因才疏学浅&#xff0c;有纰漏…
阅读更多...
KKT条件

KKT条件

无约束优化 首先给出一些基本概念定义&#xff1a; 凸集&#xff1a; 欧式空间中&#xff0c;集合中任意两点的连线都在集合中&#xff0c;我们就说这个集合是凸集&#xff1b;凸函数&#xff1a; 对于任意属于[0,1]的a和任意属于凸集的两点x, y&#xff0c;有 f ( a x ( 1 …
阅读更多...
KKT 直观理解

KKT 直观理解

KKT最优化条件是Karush[1939]&#xff0c;以及Kuhn和Tucker[1951]先后独立发表出來的。这组最优化条件在Kuhn和Tucker发表之后才逐渐受到重视&#xff0c;因此许多情况下只记载成库恩塔克条件&#xff08;Kuhn-Tucker conditions) 库恩塔克条件(Kuhn-Tucker conditions)是非线性…
阅读更多...
判断kkt条件的例题_kkt条件例题(kkt条件例题求解)

判断kkt条件的例题_kkt条件例题(kkt条件例题求解)

kkt条件例题(kkt条件例题求解) 2020-05-08 10:54:39 共10个回答 要看目标函数的斜率,不能单凭横坐标或纵坐标确定追问能举例说明吗回答一般线性规划的图像解法是通过平移一条直线,观察与可行域的焦点来求极值的这个还是线性规划里比较基础的问题.建议你找一本线性规划的书或者是…
阅读更多...
kkt条件的理解_直观理解KKT条件

kkt条件的理解_直观理解KKT条件

KKT最优化条件是Karush[1939]，以及Kuhn和Tucker[1951]先后独立发表出來的。这组最优化条件在Kuhn和Tucker发表之后才逐渐受到重视，因此许多情况下只记载成库恩塔克条件(Kuhn-Tucker conditions) 库恩塔克条件(Kuhn-Tucker conditions)是非线性规划领域里最重要的理论成果之一…
阅读更多...
对偶专题——KKT条件

对偶专题——KKT条件

[对偶专题——Duality and Dual problem (一) https://blog.csdn.net/jmh1996/article/details/85030323] 对于一般的带约束的优化问题&#xff1a; 介绍了如何通过构造原优化目标的一个下界函数 L ( x , λ , u ) L(x,\lambda,u) L(x,λ,u)&#xff0c;这一般通过添加一些线性…
阅读更多...
java telnet端口_Java 实现 telnet命令 验证主机端口的连通性

java telnet端口_Java 实现 telnet命令 验证主机端口的连通性

Java 实现 telnet命令 验证主机端口的连通性 1、Telnet 命令 Telnet协议是TCP/IP协议族中的一员，是Internet远程登录服务的标准协议和主要方式。它为用户提供了在本地计算机上完成远程主机工作的能力。在终端使用者的电脑上使用telnet程序，用它连接到服务器。终端使用者可以在…
阅读更多...
解决阿里云服务器telnet端口不成功的问题

解决阿里云服务器telnet端口不成功的问题

问题还原 阿里服务器部署了一个API站点&#xff0c;端口为8079&#xff0c;本地电脑怎么telnet到这个端口都不成功 排查 防火墙 要么把防火墙全部关闭&#xff0c;或者配置入站出站规则&#xff0c;这里我是直接关闭服务器的防火墙 查看端口是否正常监听 确保系统内部必须…
阅读更多...
能ping通，但是telnet端口连接失败

能ping通，但是telnet端口连接失败

背景&#xff1a;在windows上创建了客户端&#xff0c;使用腾讯云的轻量服务器创建的Linux服务器。 问题&#xff1a;服务器操作系统防火墙已关闭&#xff0c;使用ping命令没问题&#xff0c;但是使用telnet测试端口的时候&#xff0c;连接失败。 解决&#xff1a;一开始添加了…
阅读更多...
推荐文章

Copyright @ 2022~2023