Python 链接池 与MySQL

这里咱们使用adbapi 进行MySql的链接

import pymysql
# adbapi是python的数据库链接池
from twisted.enterprise import adbapi
from twisted.internet import reactor

class DBHelper(object):
    # 初始化链接池
    def __init__(self):
        print("初始化链接池")
        dbparams = dict(
            host ='47.101.181.53', 
            # 端口号不设置默认为3306，须要其余端口号请用Number类型。不然抛出类型异常
            port = 3307,
            db='boottest',
            user='root',
            passwd='520413',
            charset='utf8',  #编码要加上，不然可能出现中文乱码问题
            cursorclass=pymysql.cursors.DictCursor,
            use_unicode=False,
        )
        #**表示将字典扩展为关键字参数,至关于host=xxx,db=yyy....
        dbpool = adbapi.ConnectionPool('pymysql', **dbparams)
        self.dbpool = dbpool

    def connect(self):
        return self.dbpool

    def truncate(self):
        sql = "truncate table novel_tab";
        #调用插入的方法
        query = self.dbpool.runInteraction(self._truncate, sql)
        #调用异常处理方法
        query.addErrback(self._handle_error)


    def insert(self, item):
        # sql = "insert into tech_courses(title,image,brief,course_url,created_at) values(%s,%s,%s,%s,%s)"
        sql ="insert into dubbo_dog(appId,traceId,className,methodName) values(%s,%s,%s,%s)"
        #调用插入的方法
        query = self.dbpool.runInteraction(self._conditional_insert, sql, item)
        #调用异常处理方法
        query.addErrback(self._handle_error)
        return item
    
    #写入的数据
    def _conditional_insert(self,tx, sql, item):
        params = (item[0],item[1],item[2],item[3])
        tx.execute(sql, params)

    def update(self,item):
        sql = "update novel_tab set novelLastUpdateChapter = %s, novelLastUpdateChapterUrl = %s where novelUrl = %s"
        query = self.dbpool.runInteraction(self._conditional_insert, sql, item)
        #调用异常处理方法
        query.addErrback(self._handle_error)
        return item

    ## 须要修改的数据
    def _conditional_update(self, tx, sql, item):
        params = (item['lastUpdateChapter'], item['lastUpdateChapterUrl'], item['url'])
        tx.execute(sql, params)

    def _truncate(self, tx, sql):
        tx.execute(sql)

    #错误处理方法
    def _handle_error(self, failue):
        print('--------------database operation exception!!-----------------')
        print(failue)

if __name__ =='__main__':
    db = DBHelper()
    item = ['sdf',"f","fd","dfdsf"]
    db.insert(item)
    # 因为adbapi 为异步非阻塞，单独使用py文件时，
    # 必须设置异步返回的时间，不然返回结果时，因为没有回调，adbapi会进行回滚，数据库插入失败。
    # 而使用框架如Scrapy时，则没必要。
    reactor.callLater(10, reactor.stop)
    reactor.run()

复制代码