[Python] 建立SQLite3 Singleton Class
0. 前言
大家好我是Murky! 這篇文章接下來會介紹如何將Python 的 SQLite3 套件寫成Singleton 的class,，讓程式中的不同地方都可以使用同一個SQLite3。更詳盡的程式碼可以參考Github: CS50_Week9_Finance_Hw_Poetry。
SQLite不同於其他資料庫，它使用時會產生單一一個 .db 檔案而不是Client/Server界面的關聯是資料庫。SQLite3 不需要其他依賴的package, 整體小於1MB, 輕量易使用。可以一次多讀，但同時只能有一次寫入。適合直接嵌入程式中使用。
1. 參考資料
2. 事前準備
我們可以先建立一個資料夾，然後裡面生成兩個檔案 schema.sql 和 sqlite.py ， schema.sql 會用來建立sqlite database的table，而 sqlite.py 則是程式碼的本體。
接著在 schema.sql 中貼入以下程式碼，以下會用來建立一個 users table，裡面有自動增加的 id 、使用者的名稱 username 、使用者的密碼 password 、使用者帳戶有多少錢 cash 。
PRAGMA foreign_keys 的設定可以幫我們在刪除table的時候無視 foreign key constrains(在範例沒有foreign key constrains) 讓我們可以輕鬆刪除database的table方便我們多次重建。
PRAGMA foreign_keys = OFF;
DROP TABLE IF EXISTS users;
PRAGMA foreign_keys = ON;
CREATE TABLE users (
id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
username TEXT NOT NULL,
password TEXT NOT NULL,
cash NUMERIC NOT NULL DEFAULT 10000.00
);3. Singleton Class
Python 的 sqlite3 套件不需要安裝，在 sqlite.py 貼入以下程式碼。首先是 _db_path的 path/to/store/your/database 會是存放database的地方，而 finance.db 是database的名稱，如果該位置沒有 finance.db 就會生成一個。
接著是 _instance ，這會是 SQL 這個class被init之後的 instance 放置的位置，第一次 init 之後，以後再init都會回傳同一個 _instance 。
_connect 是放置 sqlite.connect ，它用來與硬碟中的 finance.db 產生連線。在 sqlite3 套建中，除了Read以外的Create、Update和delete操作都需要在輸入 sql 之後還要使用 _connect.commit() 才會正式對database做操作。
_cursor 是放置 _connect.cursor() ，它用來執行 SQL語法，可以用 _cursor.execute()執行寫在程式碼內的SQL，也可以用 _cursor.executescript 執行寫在 .sql 檔案裡的SQL。
import sqlite3
class SQL:
"""
Singleton to access database by sqlite
"""
_db_path = "path/to/store/your/database/finance.db"
_instance = None
_connect = None
_cursor = None
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super(SQL, cls).__new__(cls, *args, **kwargs)
cls._connect = sqlite3.connect(cls._db_path, check_same_thread=False)
cls._connect.row_factory = SQL.dict_factory
cls._cursor = cls._connect.cursor()
return cls._instance
@staticmethod
def dict_factory(cursor: sqlite3.Cursor, row):
"""
auto mapping sql data to dictionary
"""
fields = [column[0] for column in cursor.description]
return {key: value for key, value in zip(fields, row)}而下面的 __new__ 會在建立instance被呼叫， cls 指的就是 SQL 這個 class， super().__new__(cls) 是呼叫 SQL的是呼叫parent class的 __new__，而 sql 的parent class 是 object ，產生 SQL class並放在 _instance 裡面。這些動作只有在 cls._instance 是空的時候才會執行。
cls._connect 放入 sqlite3.connect ，其中放入database的位置，而 check_same_thread=False 可以讓非create該 SQLclass的thread 也可以使用該連線。
cls._cursor 放置從 _connect 產生的cursor。
而dict_factory (可以參考: How to create and use row factories)可以用來調整 Read data產出來得Row的表示方法。
在沒有設置 row factory 之前，資料庫中的一個Row被拿出來的時候，會是一個 tuple,，假設一個row 有 id和username 兩個欄位，如果拿出一個row就會像是 (1, "tinymurky) 要選擇其中的值就須使用 index去抓欄位, row factory 會在一個row被抓出來之後先做預處理，在這裡是依照column的名稱去mapping，mapping後會變成下面這樣的dictionary。
{
"id": 1,
"username": "tinymurky"
}4. Migrate Function
接下來往下貼入以下程式碼，這是一個 static 的 migrate function，可以使用 SQL.migrate() 直接執行，不需要init一個instance。寫成 static 的好處是如果有使用 Poetry 等套件的話 可以在 pyproject.toml 的 [tool.poetry.scripts] 使用 類似於 db_migrate="path.to.sqlite:SQL.migrate" 直接執行該function，不用啟動整個專案就可以創造database。
@staticmethod
def migrate():
"""
migrate database
"""
if SQL._connect is None or SQL._cursor is None:
SQL._connect = sqlite3.connect(SQL._db_path, check_same_thread=False)
SQL._cursor = SQL._connect.cursor()
sql_schema_path = "path/to/your/schema/sql/file/schema.sql"
with open(sql_schema_path, "r", encoding="utf-8") as f:
sql_query = f.read()
SQL._cursor.executescript(sql_query)
SQL._connect.commit()在migrate檔之中，由於他是 static 的function, 需要先假設他在使用的時候 sql.connect和sql.cursor都還沒有initialize所以才會有：
if SQL._connect is None or SQL._cursor is None:
SQL._connect = sqlite3.connect(SQL._db_path, check_same_thread=False)
SQL._cursor = SQL._connect.cursor()而下面這部份則是從loacl的資料夾中讀出剛剛寫好的schema.sql然後執行。
sql_schema_path = get_abs_path("sql/schema.sql")
with open(sql_schema_path, "r", encoding="utf-8") as f:
sql_query = f.read()
SQL._cursor.executescript(sql_query)
SQL._connect.commit() # create 操作需要呼叫commit 才會執行6. Create Row
接著我們來create一個user看看，繼續向下貼入程式碼：
def create_user(self, username: str, password: str):
"""
create user
"""
assert self._connect is not None
assert self._cursor is not None
if not isinstance(username, str) or not isinstance(password, str):
raise Exception('InvalidDevInputArgument')
query = """
INSERT INTO users (username, password)
VALUES
(?, ?)
"""
self._cursor.execute(query, (username, password))
self._connect.commit()
return self._cursor.lastrowid可以看到開頭有兩個 assert ，這是檢查 self._connect 和 self._cursor 是否存在，這兩個 assert 不一定要加，單是如果有使用 Pylance的話，不加會有警告。
assert self._connect is not None
assert self._cursor is not None接著 query 是要執行的 SQL，其中的 ? 可以在 self._cursor.execute 的時候依照順序放入。
query = """
INSERT INTO users (username, password)
VALUES
(?, ?)
"""self._cursor.execute 後面放的是 tuple, 會依照順序填到 ? 裡面，要小心就算只有一個值要填到 ? 也需要寫成 tuple 像是 (username,)(記得打逗號)。
self._connect.commit() 則會正式讓 query 操作資料庫。
self._cursor.execute(query, (username, password))
self._connect.commit()最後的 return self._cursor.lastrowid 可以回傳最後新增的row的id，不一定要加。
7. Read
有了user 之後就可以 搜尋 user，下面這個寫法可以輸入 id 或是 username 來尋找，依照 identifier 的 type做區分。
而Read不需要 self._connect.commit() ，但是在 self._cursor.execute 後需要使用 self._cursor.fetchone() 取出一個值，而該值因為 dict_factory 的設定會是一個 dictionary (也可以用 fetchmany 和 fetchall 來取得多個值)
def find_unique_user(self, identifier: Union[str, int]) -> dict[str, Any]:
"""
get unique users from users table
"""
assert self._cursor is not None
if isinstance(identifier, str):
query = """
SELECT * FROM users
WHERE username = ?
LIMIT 1;
"""
elif isinstance(identifier, int):
query = """
SELECT * FROM users
WHERE id = ?
LIMIT 1;
"""
else:
raise Exception('InvalidDevInputArgument')
self._cursor.execute(query, (identifier,))
user = self._cursor.fetchone()
return user最後，只要使用下面的 程式碼就可以 init SQL 了！
sql_client = SQL()8. 結語
這樣我們就完成了一個Singleton的 SQLite3 的class了，那我們下次見！
