一. 前言：有一個 DB 檔，不等於有一份能還原的備份 #
嗨，我是拍拍君。
你的小工具持續寫入 SQLite，想每天存一份備份。
於是複製 live.db，看到檔案存在，就安心關掉終端機。
先等一下：WAL 模式下，已提交的資料可能還在 -wal，不在主檔。
只搬主檔可能遺漏資料；一邊寫一邊分別複製各檔，也不是一致快照。
這是 SQLite WAL 文件 特別提醒的邊界。
sqlite3 入門文 說明過 CRUD、交易與 WAL 開關。 今天不重做資料管理 App，而是完成一件不同的事： 備份運作中的資料庫，驗收後才掛上正式檔名，再真的還原一次。
二. 先訂契約：這份備份代表哪個時點？ #
我們只用自己產生的 events 測試資料，不讀任何現成 App 資料庫。
流程是：
- 另一條連線以唯讀方式開啟來源，建立並固定讀取快照。
- 記錄這個快照的 schema、筆數與排序後內容摘要。
- 備份到同一父目錄下的私人暫存子目錄。
- 關閉目標連線，重新開啟暫存檔，獨立驗收。
- 全部成功，才替換離線的備份檔；還原則使用全新目錄。
「一致」不是「包含備份結束前的每一筆提交」。 本例刻意固定來源讀取交易，所以稍後的成功提交不進這份備份。 若不固定讀取快照，增量備份中其他連線修改來源，可能使複製重新開始； 不能把呼叫函式的時間直接當成精準的業務切點。 參見 SQLite Backup API 的鎖定與重啟規則。
三. 安裝：標準函式庫就夠，版本仍要記錄 #
使用 Python 3.12 以上，不必安裝第三方套件。
建立一個全新的工作目錄，後面四段 Python 依序存進同一個 backup_lab.py：
mkdir pypy-sqlite-backup-lab
cd pypy-sqlite-backup-lab
python3 -c "import sys, sqlite3; print(sys.version); print(sqlite3.sqlite_version)"
本文實測 Python 3.14.3／SQLite 3.53.4。
autocommit=True 明確把交易交給 SQL 的 BEGIN／COMMIT 管理。
這不是修改整台電腦的資料庫設定；所有檔案都在臨時目錄，用完清除。
四. 先做驗收器：檔案開得起來，還不夠 #
第一段加入匯入、唯讀連線與這個範例專用的驗收器：
import hashlib
import json
import os
import sqlite3
import time
from contextlib import closing
from pathlib import Path
from tempfile import TemporaryDirectory
def connect_ro(path):
return sqlite3.connect(
Path(path).resolve().as_uri() + "?mode=ro",
uri=True, autocommit=True, timeout=0.2,
)
def inspect_db(conn):
if conn.execute("PRAGMA integrity_check").fetchall() != [("ok",)]:
raise ValueError("integrity check failed")
if conn.execute("PRAGMA foreign_key_check").fetchall():
raise ValueError("foreign key check failed")
version = conn.execute("PRAGMA user_version").fetchone()[0]
columns = conn.execute("PRAGMA table_info(events)").fetchall()
expected = [(0, "id", "INTEGER", 0, None, 1),
(1, "body", "TEXT", 1, None, 0)]
if version != 1 or columns != expected:
raise ValueError("unsupported application schema")
schema = conn.execute(
"SELECT type,name,tbl_name,sql FROM sqlite_schema "
"WHERE name NOT LIKE 'sqlite_%' ORDER BY type,name"
).fetchall()
digest = hashlib.sha256()
count = 0
for row in conn.execute("SELECT id,body FROM events ORDER BY id"):
digest.update((json.dumps(row, ensure_ascii=False) + "\n").encode())
count += 1
if not count:
raise ValueError("empty lab database")
return version, schema, count, digest.hexdigest()
這裡有三層驗收，責任不同：
| 層次 | 驗收什麼 | 不代表什麼 |
|---|---|---|
integrity_check／foreign_key_check |
SQLite 結構及外鍵錯誤 | 業務資料一定正確 |
user_version／欄位形狀 |
這個 App 支援的 schema | 所有其他 App 都能讀 |
| 筆數／內容摘要 | 同一排序、序列化規則下資料相符 | 來源可信、已加密或已異地備援 |
integrity_check 不涵蓋外鍵錯誤，因此另外檢查。
官方 PRAGMA 文件 有列出兩者的分工。
真實專案要依 schema 加上必要資料表、索引、觸發器及業務查詢檢查。
本例只有一張業務表，摘要逐列更新，不把全部列一次載入記憶體。
mode=ro 避免路徑拼錯時偷偷建立空來源，也禁止透過這條連線寫入。
Path.as_uri() 則處理路徑裡的空白、?、#。
別為了唯讀就加 immutable=1：來源仍在變，這個承諾不成立。
唯讀 WAL 仍可能需要既有的 sidecar 或可建立 sidecar 的目錄權限，
不是「無權讀完整來源，也能繞過去」；細節見 SQLite URI 參數。
五. 核心：暫存、固定快照、驗收，再換正式檔名 #
第二段接上備份函式：
def backup_checked(source, final, *, seconds=5.0, on_step=None):
source, final = Path(source).resolve(), Path(final).resolve()
if source == final or (final.exists() and os.path.samefile(source, final)):
raise ValueError("source and backup must differ")
deadline = time.monotonic() + seconds
def expired():
return time.monotonic() >= deadline
def progress(code, remaining, total):
if expired():
raise TimeoutError("backup lab deadline exceeded")
if on_step is not None:
on_step(code, remaining, total)
with TemporaryDirectory(prefix=".stage-", dir=final.parent) as folder:
stage = Path(folder) / "snapshot.db"
with closing(connect_ro(source)) as src:
src.set_progress_handler(expired, 1000)
src.execute("BEGIN")
# BEGIN alone does not establish a read snapshot.
src.execute("SELECT count(*) FROM sqlite_schema").fetchone()
expected = inspect_db(src)
with closing(sqlite3.connect(stage, autocommit=True)) as dst:
dst.set_progress_handler(expired, 1000)
src.backup(dst, pages=8, progress=progress, sleep=0.01)
mode = dst.execute("PRAGMA journal_mode=DELETE").fetchone()[0]
if mode != "delete":
raise ValueError("staged backup is not standalone")
src.execute("ROLLBACK")
with closing(connect_ro(stage)) as check:
check.set_progress_handler(expired, 1000)
if inspect_db(check) != expected:
raise ValueError("backup manifest mismatch")
if expired():
raise TimeoutError("backup lab deadline exceeded")
os.replace(stage, final)
return expected
src.backup(dst) 的方向是來源呼叫、目標當參數。
pages=8 是每一步複製的頁數，不是只備份八頁就回傳。
Python 的呼叫會持續到完成或發生例外；sleep 是忙碌重試間隔，
不是每成功八頁就固定休息。參見 Python backup 參數。
BEGIN 本身還沒固定快照，要真的讀取來源後才有讀取切點。
本例在同一條來源連線、同一個讀取交易裡取得 manifest 和執行備份。
這樣比較的不是「剛剛的來源」和「又過了一秒的來源」。
目標完成後轉成 DELETE journal mode，確認回傳值，
再關閉並重新開啟驗收，讓發布的是可獨立搬運的備份主檔。
這裡改的是暫存目標，不是對運作中的來源強制切換模式。
另外，with conn: 管交易，不能當成自動關連線。
所以程式使用 closing()，離開範圍確實 close()。
這個差別對檔案搬移、sidecar 收尾和資源釋放都很重要。
失敗時，舊備份會怎樣？ #
複製、驗收或 callback 丟例外，都還沒走到 os.replace()：
舊備份保留，當次暫存目錄由 context manager 清理，例外往外傳。
若作業程序被強制終止，暫存目錄可能留下；下次應辨識自己的殘留目錄再處理。
這個實驗假設單一備份工作、可信任的本機父目錄，正式備份沒有被任何程序開啟。 不要拿它直接覆蓋活躍資料庫，也不要讓兩個排程競爭同一個目的地。 同目錄替換提供檔名切換邊界，不等於斷電耐久、目錄 fsync、版本保留或異地保存全都完成。
六. 併發實驗：來源 257 筆，備份為何是 256 筆？ #
第三段建立足以跨多個 backup step 的假資料：
def create_source(conn):
if conn.execute("PRAGMA journal_mode=WAL").fetchone()[0] != "wal":
raise RuntimeError("WAL unavailable")
conn.execute("PRAGMA wal_autocheckpoint=0")
conn.execute("PRAGMA user_version=1")
conn.execute("CREATE TABLE events(id INTEGER PRIMARY KEY, body TEXT NOT NULL)")
conn.execute("BEGIN")
conn.executemany("INSERT INTO events VALUES (?, ?)",
((n, "拍拍醬-" + "x" * 1024) for n in range(1, 257)))
conn.execute("COMMIT")
第四段加入示範入口：
def demo():
with TemporaryDirectory(prefix="pypy-backup-") as folder:
root = Path(folder)
source, backup = root / "live.db", root / "backup.db"
with closing(sqlite3.connect(source, autocommit=True)) as writer:
create_source(writer)
committed = False
def commit_during_backup(code, remaining, total):
nonlocal committed
if not committed and code == sqlite3.SQLITE_OK and remaining:
writer.execute("INSERT INTO events VALUES (?, ?)", (257, "拍拍君"))
committed = True
report = backup_checked(source, backup, on_step=commit_during_backup)
assert committed, "fixture must span multiple backup steps"
assert writer.execute("SELECT count(*) FROM events").fetchone()[0] == 257
assert report[2] == 256
print("live=257, pinned backup=256, concurrent commit=OK")
# Restore into an unused directory, never over an active database.
restore_dir = root / "restore"
restore_dir.mkdir()
restored = restore_dir / "restored.db"
assert backup_checked(backup, restored) == report
with closing(sqlite3.connect(restored, autocommit=True)) as app:
assert inspect_db(app) == report
app.execute("INSERT INTO events VALUES (?, ?)", (257, "chatPTT"))
assert app.execute("SELECT count(*) FROM events").fetchone()[0] == 257
with closing(connect_ro(backup)) as check:
assert inspect_db(check) == report
print("restore read/write=OK, original backup unchanged")
if __name__ == "__main__":
demo()
on_step 只是這個實驗的接縫，刻意在第一個尚未完成的 step 後，
用另一條 writer connection 提交一筆資料，確保提交發生在備份呼叫期間。
它不是建議在正式備份 callback 裡做業務寫入，更不是重複送出外部請求的地方。
執行：
python3 backup_lab.py
應看到：
live=257, pinned backup=256, concurrent commit=OK
restore read/write=OK, original backup unchanged
這不是少備份了一筆。 第 257 筆在快照固定之後才提交；舊快照仍有完整的 256 筆。 若需要「月底結帳後」這種業務切點，應先協調業務交易與切點， 不能只看備份檔名的日期猜測內容。
還原演練也沒有碰原始來源：從已發布備份再產生一個全新路徑， 驗收後以可寫連線做一筆測試寫入，最後確認原備份沒有變。 「SELECT 能跑」和「App 能讀寫」是兩個值得分開驗收的結果。
七. 故意失敗：新任務不能破壞上一份成功備份 #
把下面另存為 test_backup_lab.py，和主程式放同一目錄：
import sqlite3
import unittest
from contextlib import closing
from pathlib import Path
from tempfile import TemporaryDirectory
from backup_lab import backup_checked, create_source, inspect_db, connect_ro
class BackupTest(unittest.TestCase):
def test_callback_failure_keeps_previous_backup(self):
with TemporaryDirectory() as folder:
source, final = Path(folder) / "live.db", Path(folder) / "backup.db"
with closing(sqlite3.connect(source, autocommit=True)) as writer:
create_source(writer)
before = backup_checked(source, final)
previous_bytes = final.read_bytes()
writer.execute("INSERT INTO events VALUES (257, '拍拍君')")
def fail(*args):
raise RuntimeError("injected failure")
with self.assertRaisesRegex(RuntimeError, "injected"):
backup_checked(source, final, on_step=fail)
self.assertEqual(final.read_bytes(), previous_bytes)
self.assertEqual(list(Path(folder).glob(".stage-*")), [])
with closing(connect_ro(final)) as check:
self.assertEqual(inspect_db(check), before)
if __name__ == "__main__":
unittest.main()
用標準函式庫執行：
python3 -m unittest -v
本次另外驗證了來源不存在不會建空檔、唯讀寫入拒絕、 同一路徑與 hardlink／symlink 別名拒絕、schema 不合、空資料拒絕、 同筆數但內容改變的摘要差異、未提交資料排除，以及只複製主檔會漏掉 WAL 內容。 這些是本文合成 fixture 的實測結果，不是對任意外部資料庫的保證。
八. 時限與正式使用的邊界 #
程式用 monotonic deadline、backup callback 和 SQL progress handler 做合作式中止。
connect(timeout=...) 不是整份備份的總時間上限。
若底層 I/O 卡住，合作式檢查也不是作業系統層級的硬 deadline。
執行這份有限資料量實驗時，可再用外層程序限制總時間：
import subprocess
import sys
subprocess.run([sys.executable, "backup_lab.py"], check=True, timeout=15)
長讀取快照會延後 WAL checkpoint 的推進，來源持續寫入時 WAL 可能成長。
正式排程需要觀測來源寫入負載、WAL 大小與備份耗時，不要無限重試。
本文只備份 main，不涵蓋多個 ATTACH 資料庫的跨庫原子快照。
SQL dump 是邏輯匯出，對特定需求有用，但不是把這條備份／驗收流程換個檔名。
更完整的備份制度還需要權限、加密、版本保留、異地保存及定期還原演練。
那些是下一層工作，不是 backup() 回傳就自動獲得的功能。
結語：備份的終點，是能在新地方重新工作 #
拍拍君會把這次流程記成四個動詞：固定、複製、驗收、還原。 來源仍可提交，不代表備份必須追到最後一秒；檔案存在，也不代表它是可用版本。 先定義要保存的狀態，再用可執行的檢查證明你真的拿到了它。
下一次替小工具加每日備份，別只在 log 裡寫「copy success」。 留下一份能獨立讀寫、能說清楚切點、失敗時不會毀掉上一版的備份，才算真的收工。