[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該 SQL
class的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了,那我們下次見!