使用 Python 中的连接池来管理 SQLite 数据库连接

pip install sqlite3

import sqlite3
from concurrent.futures import ThreadPoolExecutor

class SQLiteConnectionPool:
    def __init__(self, db_name, pool_size=5):
        self.db_name = db_name
        self.pool_size = pool_size
        self.pool = ThreadPoolExecutor(max_workers=self.pool_size)

    def get_connection(self):
        return sqlite3.connect(self.db_name)

    def execute(self, query, params=(), callback=None):
        def _execute_and_return(conn, query, params):
            cursor = conn.cursor()
            cursor.execute(query, params)
            result = cursor.fetchall()
            cursor.close()
            conn.commit()
            conn.close()
            return result

        return self.pool.submit(_execute_and_return, self.get_connection(), query, params).result()

    def close(self):
        self.pool.shutdown()

    def __del__(self):
        self.close()

from connection_pool import SQLiteConnectionPool

# 创建并初始化数据库
def init_db(conn_pool):
    create_table_query = '''
    CREATE TABLE IF NOT EXISTS websites (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        url TEXT NOT NULL
    );
    '''
    conn_pool.execute(create_table_query)

# 插入数据
def insert_website(conn_pool, name, url):
    insert_query = '''
    INSERT INTO websites (name, url)
    VALUES (?, ?);
    '''
    conn_pool.execute(insert_query, (name, url))

# 查询数据
def get_websites(conn_pool):
    select_query = '''
    SELECT id, name, url
    FROM websites;
    '''
    return conn_pool.execute(select_query)

def main():
    conn_pool = SQLiteConnectionPool("pidancode.db", pool_size=5)
    init_db(conn_pool)

    insert_website(conn_pool, "皮蛋编程", "https://www.pidancode.com")
    websites = get_websites(conn_pool)

    for website in websites:
        print(website)

    conn_pool.close()

if __name__ == "__main__":
    main()