首先安裝資料連線池模組

pip3 install DBUtils

然後安裝mysql驅動包

pip3 install PyMySQL

安裝完成之後，在專案中settings檔案裡面配置好資料連線資訊，如下圖：

新建一個myql_help檔案，名稱自己取，然後複製參考以下程式碼，我這裡簡單做了一下封裝：

import pymysql
from pymysql.cursors import DictCursor
from DBUtils.PooledDB import PooledDB
from settings import host,port,user,password,db_name

#父類連線池，用於初始化資料庫連線
class BasePymysqlPool(object):
    def __init__(self):
        self.db_host = host
        self.db_port = int(port)
        self.user = user
        self.password = str(password)
        self.db = db_name
        self.conn = None
        self.cursor = None

class PymysqlPool(BasePymysqlPool):
    """
    MYSQL資料庫物件，負責產生資料庫連線 , 此類中的連線採用連線池實現獲取連線物件：conn = Mysql.getConn()
            釋放連線物件;conn.close()或del conn
    """
    # 連線池物件
    __pool = None

    def __init__(self):
        super(PymysqlPool, self).__init__()
        # 資料庫建構函式，從連線池中取出連線，並生成操作遊標
        self._conn = self.__getConn()
        self._cursor = self._conn.cursor()

    def __getConn(self):
        """
        @summary: 靜態方法，從連線池中取出連線
        @return MySQLdb.connection
        """
        if PymysqlPool.__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=True,#此處應設定為True，否則查詢出來的資料會變成bytes型別
                              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 = False
        return result

    def 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 = False
        return result

    def 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 = False
        return result

    def insertMany(self, sql, values):
        """
        @summary: 向資料表插入多條記錄
        @param sql:要插入的ＳＱＬ格式
        @param values:要插入的記錄資料tuple(tuple)/list[list]
        @return: count 受影響的行數
        """
        count = self._cursor.executemany(sql, values)
        return count

    def __query(self, sql, param=None):
        if param is None:
            count = self._cursor.execute(sql)
        else:
            count = self._cursor.execute(sql, param)
        return count

    def 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 = PymysqlPool()
    sql_one = "select * from airportinfonewthree"
    result = mysql.getOne(sql_one)
    print(result)
    # 釋放資源
    mysql.dispose()
 

參考地址：https://www.cnblogs.com/renfanzi/p/7656142.html