前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >python DbUtils 使用教程

python DbUtils 使用教程

作者头像
py3study
发布2020-01-08 16:01:57
6.8K0
发布2020-01-08 16:01:57
举报
文章被收录于专栏:python3

下载:

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:

代码语言:javascript
复制
#-*- 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

代码语言:javascript
复制
#-*- 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

代码语言:javascript
复制
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";
本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2019/09/22 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • coding=utf-8
相关产品与服务
数据库
云数据库为企业提供了完善的关系型数据库、非关系型数据库、分析型数据库和数据库生态工具。您可以通过产品选择和组合搭建,轻松实现高可靠、高可用性、高性能等数据库需求。云数据库服务也可大幅减少您的运维工作量,更专注于业务发展,让企业一站式享受数据上云及分布式架构的技术红利!
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档