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()

複製代碼