Python+sqlite:一個簡單的數據庫連接池

import sqlite3
import os
from threading import Lock, BoundedSemaphore


class ConnPool:

    def __init__(self, db_name, max_conn):
        self.__MaxConn = max_conn  # 最大連接數
        self.__CurConn = 0  # 當前連接數
        self.__FreeConn = 0  # 空閑連接數
        self.__lock = Lock()  # 鎖
        self.__connList = []  # 連接池

        self.dbName = db_name  # 數據庫名稱
        self.init_conn()  # 初始化

    def init_conn(self):
        for i in range(self.__MaxConn):
            try:
                conn = sqlite3.connect(os.path.abspath('.') + '\\' + self.dbName + '.db', check_same_thread=False)
                if conn is not None:
                    self.__connList.append(conn)  # 將獲取到的數據庫連接加入連接池
                    self.__FreeConn += 1
            except Exception as e:
                print("Get Sqlite Connection Failed: {0}".format(i) + e)
        self.__sem = BoundedSemaphore(self.__FreeConn)  # 初始化信號量
        self.__MaxConn = self.__FreeConn

    # 每當有請求時，從數據庫連接池中返回一個可用連接，並更新使用和空閑連接數
    def get_connection(self):
        if 0 == len(self.__connList):
            return None
        self.__sem.acquire()  # 消耗資源使計數器遞減，為0時阻塞調用
        self.__lock.acquire()  # 加鎖
        conn = self.__connList.pop()  # 從連接列表中彈出一個數據庫連接

        self.__FreeConn -= 1  # 空閑連接計數-1
        self.__CurConn += 1  # 當前連接計數+1

        self.__lock.release()  # 釋放鎖
        return conn

    # 釋放當前使用的連接
    def release_connection(self, conn):
        if conn is None:
            return False

        self.__lock.acquire()       # 加鎖

        self.__connList.append(conn)    # 歸還數據庫連接
        self.__FreeConn += 1
        self.__CurConn -= 1

        self.__lock.release()   # 釋放鎖
        try:
            self.__sem.release()    # 歸還資源使計數器遞增
        except ValueError:
            print("超出計數器上限.")
        return True

    # 銷毀數據庫連接池
    def destroy_pool(self):
        self.__lock.acquire()       # 加鎖
        if len(self.__connList) > 0:
            for it in self.__connList:
                it.close()
            self.__CurConn = 0
            self.__FreeConn = 0
            self.__connList.clear()
        self.__lock.release()       # 釋放鎖

    # 獲取空閑的連接數
    def get_free_conn(self):
        return self.__FreeConn

    # 析構函數
    def __del__(self):
        self.destroy_pool()


# 創建一個數據庫資源管理類，傳入一個類型為ConnPool對象，生成一個數據庫連接，析構函數中釋放此連接
class ConnPoolRAII:
    def __init__(self, conn_pool: ConnPool):
        self.connRAII = conn_pool.get_connection()
        self.poolRAII = conn_pool
        self.__lock = Lock()    # 數據庫操作鎖
        self.init_db()          # 初始化數據庫

    # 數據庫初始化
    def init_db(self):
        # 若用於存儲下位機的表不存在，則進行初始化創建
        if not self.is_exist('Client'):
            self.operate('''create table Client(
                            ID integer primary key autoincrement,
                            Client_name text(10) not null,
                            IPAddress text(20) not null,
                            Status numeric default 0
                            )''')

    # 檢查指定表是否存在
    def is_exist(self, table_name):
        res = self.query('''select count(*) from sqlite_master where type='table' and name='{0}'
                        '''.format(table_name))
        if res.fetchone()[0] == 0:
            return False
        return True

    # 為每一個客戶機創建一個用於存儲數據的表格
    def create_table(self, client_name):
        return self.operate('''
                    create table {0}(
                        ID integer primary key autoincrement,
                        Weight real default 0,
                        Temperature real default 0,
                        Date integer not null
                        );
                        '''.format(client_name))

    # 新增客戶機函數
    def add_client(self, ip):
        client_name = 'Client{0}'.format(ip[ip.rfind('.') + 1:])    # 獲取客戶端名稱，以Clientxxx為格式
        res = self.query("select * from Client where IPAddress=='{0}'".format(client_name))     # 查詢Client表中是否存在當前客戶端信息
        sql_insert = "insert into Client (Client_name, IPAddress) values ('{0}','{1}')".format(client_name, ip)
        if res and (res.fetchone() is None):
            if self.operate(sql_insert) and self.create_table(client_name):
                print("創建表成功")      # 測試用，后續改為日志
                return True
            else:
                print("創建表失敗,檢查客戶端是否已存在！")
                return False

    # 用於查詢類操作，要求輸入sql語句
    def query(self, sql):
        self.__lock.acquire()       # 加鎖
        c = self.connRAII.cursor()  # 獲取cursor
        try:
            res = c.execute(sql)    # 執行sql
            self.__lock.release()   # 釋放鎖
            return res              # 返回查詢結果
        except Exception as e:
            print(e)
            self.__lock.release()   # 釋放鎖
            return False            # 返回False

    # 增刪改等操作，提供sql語句，返回bool值
    def operate(self, sql):
        self.__lock.acquire()       # 加鎖
        c = self.connRAII.cursor()  # 獲取cursor
        try:
            c.execute(sql)          # 執行sql
            self.connRAII.commit()  # 提交執行結果
            self.__lock.release()   # 釋放鎖
            return True             # 返回操作成功
        except Exception as e:
            print(e)                # 打印錯誤信息
            self.__lock.release()   # 釋放鎖
            return False            # 返回False

    # 析構函數
    def __del__(self):
        self.poolRAII.release_connection(self.connRAII)