python DbUtils 使用教程

发布时间：2019-09-22 07:45:18编辑：auto阅读（3113）

下载：

https://pypi.python.org/pypi/DBUtils/1.2





先下载你要安装的包，并解压到磁盘下；


进入到该文件的setup.py 目录下 ，打开cmd，并切换到该目录下；


先执行 python setup.py build


然后执行 python setup.py install




后面部分参考了：https://blog.csdn.net/u010939285/article/details/71088145

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

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

MyDbUtils：

#-*- coding: UTF-8 -*-
import pymysql
from DBUtils.PooledDB import PooledDB
import DB_config as Config
import MySQLdb
'''
@功能：PT数据库连接池
'''
class PTConnectionPool(object):
    __pool = None;
    # def __init__(self):
    #     self.conn = self.__getConn();
    #     self.cursor = self.conn.cursor();
    def __enter__(self):
        self.conn = self.__getConn();
        self.cursor = self.conn.cursor();
        print u"PT数据库创建con和cursor";
        return self;

    def __getConn(self):
        if self.__pool is None:
            self.__pool = PooledDB(creator=MySQLdb, 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()
"""
@summary: 释放连接池资源
"""
def __exit__(self, type, value, trace):
    self.cursor.close()
    self.conn.close()
    print u"PT连接池释放con和cursor";

#重连接池中取出一个连接
def getconn(self):
    conn = self.__getConn();
    cursor = conn.cursor();
    return cursor,conn

#关闭连接归还给连接池
# def close(self):
#     self.cursor.close()
#     self.conn.close()
#     print u"PT连接池释放con和cursor";


def getPTConnection():
    return PTConnectionPool()

配置文件：DB_config.py

#-*- coding: UTF-8 -*-

#TEST数据库信息
DB_TEST_HOST="192.168.88.6";
DB_TEST_PORT=3306;
DB_TEST_DBNAME="asterisk";
DB_TEST_USER="root";
DB_TEST_PASSWORD="kalamodo";


#数据库连接编码
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;

封装的mysqlhelp.py

coding=utf-8

import MySQLdb
from MyDbUtils import getPTConnection

class MysqlHelp(object):

mysql=None
def __init__(self):
    # self.connect()
    self.db=getPTConnection()
def __new__(cls, *args, **kwargs):
    if not hasattr(cls, 'inst'):
        cls.inst = super(MysqlHelp, cls).__new__(cls, *args, **kwargs)
    return cls.inst
#查询所有
def selectall(self,sql='',param=()):
    #判断是否连接,并设置重连机制
    # self.connected()
    try:
        cursor,conn=self.execute(sql,param)
        res=cursor.fetchall()
        self.close(cursor, conn)
        return res
    except Exception,e:
        print 'selectall except   ', e.args
        self.close(cursor, conn)
        return None

#查询一条
def selectone(self,sql='',param=()):
    # self.connected()
    try:
        # cur = self.db.cursor()
        cursor, conn=self.execute(sql,param)
        res = cursor.fetchone()
        self.close(cursor, conn)
        return res
    except Exception, e:
        print 'selectone except   ', e.args
        self.close(cursor, conn)
        return None

#增加
def insert(self,sql='',param=()):
    # self.connected()
    try:
        # self.db.getconn().execute(sql, param)
        cursor, conn=self.execute(sql,param)
        print '============'
        # _id=self.db.conn.insert_id()
        _id=cursor.lastrowid
        print '_id   ',_id
        conn.commit()
        self.close(cursor, conn)
        #防止表中没有id返回0
        if _id==0:
            return True
        return _id
    except Exception, e:
        print 'insert except   ', e.args
        conn.rollback()
        self.close(cursor, conn)
        # self.conn.rollback()
        return 0

#增加多行
def insertmany(self,sql='',param=()):
    # self.connected()
    cursor,conn=self.db.getconn()
    try:
        cursor.executemany(sql, param)
        # self.execute(sql,param)
        conn.commit()
        self.close(cursor, conn)
        return True
    except Exception, e:
        print 'insert many except   ', e.args
        conn.rollback()
        self.close(cursor, conn)
        # self.conn.rollback()
        return False

#删除
def delete(self,sql='',param=()):
    # self.connected()
    try:
        # cur = self.conn.cursor()
        # self.db.getconn().execute(sql, param)
        cursor,conn=self.execute(sql,param)
        # self.db.conn.commit()
        self.close(cursor, conn)
        return True
    except Exception, e:
        print 'delete except   ', e.args
        conn.rollback()
        self.close(cursor, conn)
        # self.conn.rollback()
        return False

#更新
def update(self,sql='',param=()):
    # self.connected()
    try:
        #cur = self.conn.cursor()
        # self.db.getconn().execute(sql, param)
        cursor,conn=self.execute(sql,param)
        # self.db.conn.commit()
        self.close(cursor, conn)
        return True
    except Exception, e:
        print 'update except   ',e.args
        conn.rollback()
        self.close(cursor, conn)
        # self.conn.rollback()
        return False

@classmethod
def getInstance(self):
    if MysqlHelp.mysql==None:
        MysqlHelp.mysql=MysqlHelp()
    return MysqlHelp.mysql

#执行命令
def execute(self,sql='',param=(),autoclose=False):
    cursor, conn = self.db.getconn()
    try:
        if param:
            cursor.execute(sql, param)
        else:
            cursor.execute(sql)
        conn.commit()
        if autoclose:
            self.close(cursor, conn)
    except Exception as e:
        pass
    return cursor, conn

#执行多条命令
'[{"sql":"xxx","param":"xx"}....]'
def executemany(self,list=[]):
    cursor,conn=self.db.getconn()
    try:
        for order in list:
            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 'execute failed========',e.args
        conn.rollback()
        self.close(cursor, conn)
        return False
def connect(self):
    self.conn = MySQLdb.connect(user='root', db='asterisk', passwd='kalamodo', host='192.168.88.6')

def close(self,cursor,conn):
    cursor.close()
    conn.close()
    print u"PT连接池释放con和cursor";

传送门 http://download.csdn.net/download/u010939285/9831459

关键字：