一覧DB-09-2 · 保存・複製・復旧

バックアップと変更履歴による復元

SQLiteのbackup APIによるコピーと、SQLの論理ダンプを作ります。原表を更新してから両方を復元し、保存時点の違いを確認します。後半では、この例で用意した連番付き業務変更ログを順に適用し、指定番号で止める復元を行います。

説明

バックアップは作成した時点の状態を戻すためのものです。その後に成功した変更まで戻したい場合には、バックアップ以降の変更履歴と、どこまで適用するかの境界が必要です。保存しただけでなく、別のDBとして復元して確認します。

1保存元のDBと変更番号

在庫10と、適用済み変更番号0を保存します。

import sqlite3,tempfile
from pathlib import Path
tmp = tempfile.TemporaryDirectory()
path = Path(tmp.name)/'source.db'
source = sqlite3.connect(path,isolation_level=None)
source.executescript('

CREATE TABLE stock保存する在庫(id INTEGER PRIMARY KEY,qty INTEGER); INSERT INTO stock VALUES(1,10); CREATE TABLE applied適用済み番号の記録

(seq INTEGER PRIMARY KEY); INSERT INTO applied VALUES(0);')
print(source.execute('SELECT * FROM stock').fetchall())

stockが業務データ、appliedが適用済みログの番号です。ここでは一件の変更ごとに番号を付け、同じ番号を二度適用しないための記録として使います。

2backup APIでDBをコピーする

SQLiteのバックアップ機能で別ファイルへ保存します。

backup_path = Path(tmp.name)/'backup.db'
backup = sqlite3.connect(backup_path)

source.backup(backup)SQLiteのAPIでスナップショットを取得


print(backup.execute('SELECT * FROM stock').fetchall())

backup.close()保存先の接続を閉じる

バックアップにも在庫10があります。backup APIはDBのページを整合したスナップショットとしてコピーする機能です。本体ファイルだけを任意のタイミングでコピーする方法とは異なります。

3SQLとして論理ダンプする

表定義と行をSQL文として保存します。

dump = '\n'.joinSQL文を改行で連結(source.iterdump()定義と行をSQL文へ変換

)
print(dump)

BEGIN TRANSACTION、CREATE TABLE、INSERT、COMMITが並びます。論理ダンプは復元先でSQLとして実行できる形式です。元のページ配置や索引の物理的な並びをそのまま複製するものではありません。

4バックアップ後に在庫を変更する

元のDBだけを15へ更新します。

source.execute('UPDATE stock SET qty=15保存元だけを変更

 WHERE id=1')
print('source:',source.execute('SELECT qty FROM stock').fetchone())
backup = sqlite3.connect(backup_path)

print('backup:'保存時点の値を再確認

,backup.execute('SELECT qty FROM stock').fetchone())
backup.close()

元のDBは15、保存済みバックアップは10です。バックアップは更新に追随する複製ではありません。どの時点を保存したかが、復元できる状態を決めます。

5物理コピーと論理ダンプを別DBへ復元する

バックアップからのコピーと、ダンプのSQL実行を行います。

physical = sqlite3.connect(':memory:',isolation_level=None)
backup = sqlite3.connect(backup_path)

backup.backup(physical)保存コピーから復元


backup.close()
logical = sqlite3.connect(':memory:',isolation_level=None)

logical.executescript(dump)SQLダンプから復元


print('physical:',physical.execute('SELECT * FROM stock').fetchall())
print('logical:',logical.execute('SELECT * FROM stock').fetchall())

両方とも(1,10)です。保存したものを別のDBで使えることを確認できました。原表の最新15ではなく、バックアップ時点の10へ戻っています。

6連番付きの業務変更ログを用意する

在庫の加減算を番号1～3で記録します。

events = [(1,5),(2,-2),(3,7)]順序と増減値を持つ変更履歴


print('seq,delta:',events)

番号1は+5、2は-2、3は+7です。ここではこの業務ログを使って復元後の変更を再適用します。SQLiteのWALをSQLへ変換して適用する操作ではなく、アプリケーションが記録する変更履歴という別の設計です。

7番号2までを一度だけ適用する

各変更と番号の記録を同じトランザクションで確定します。

target = 2復元する最終番号


for seq,delta in events:
    if seq>target:
        break
    physical.execute('BEGIN')
    added = physical.execute('

INSERT OR IGNORE INTO applied既に適用した番号を検出

 VALUES(?)',(seq,)).rowcount

if added:今回初めての番号だけ変更


        physical.execute('UPDATE stock SET qty=qty+? WHERE id=1',(delta,))

physical.execute('COMMIT')値と番号を一緒に確定


print(physical.execute('SELECT qty FROM stock').fetchone())
print(physical.execute('SELECT seq FROM applied ORDER BY seq').fetchall())

在庫は10+5-2=13、適用済みは0・1・2です。目標番号を超える3の+7は適用していません。番号の保存とデータ変更を一緒に確定するので、途中で再開する際の境界を管理できます。

8同じログを再適用しても二重に足さない

同じ番号1・2をもう一度渡します。

for seq,delta in events[:2]既に適用した二件

:
    physical.execute('BEGIN')
    added = physical.execute('INSERT OR IGNORE INTO applied VALUES(?)',(seq,)).rowcount

if added:重複なら金額変更をしない


        physical.execute('UPDATE stock SET qty=qty+? WHERE id=1',(delta,))
    physical.execute('COMMIT')
print(physical.execute('SELECT qty FROM stock').fetchone())

在庫は13のままです。番号の一意性によって再適用が無視されています。ただし実際の設計では、同じ番号に異なる内容が届いた場合や欠番も検査し、ログ全体を無条件に信用しない仕組みが必要です。

9次の番号まで進めて整合性を確認する

未適用の3だけを適用します。

seq,delta = events[2]まだ適用していない番号3


physical.execute('BEGIN')
physical.execute('INSERT INTO applied VALUES(?)',(seq,))
physical.execute('UPDATE stock SET

qty=qty+?最後の増分7を適用

 WHERE id=1',(delta,))
physical.execute('COMMIT')
print(physical.execute('SELECT qty FROM stock').fetchone())
print(physical.execute('

PRAGMA integrity_check復元先の構造を確認').fetchone())

在庫は20、構造検査はokです。どのバックアップを基点にし、どの番号までの履歴をそろえたかで復元結果が決まります。構造が正常であることと、必要な業務時点まで戻せたことは別々に確認します。

まとめ

保存時点のコピーを別DBへ復元し、値と定義を確認します。

指定時点へ進めるには、基点と変更履歴の連続した境界が必要です。

変更と適用済み番号を一緒に確定し、再実行時の二重適用を防ぎます。

操作履歴と解説を振り返れます。

Python / SQLite

import sqlite3,tempfile
from pathlib import Path
tmp = tempfile.TemporaryDirectory()
path = Path(tmp.name)/'source.db'
source = sqlite3.connect(path,isolation_level=None)
source.executescript('CREATE TABLE stock(id INTEGER PRIMARY KEY,qty INTEGER); INSERT INTO stock VALUES(1,10); CREATE TABLE applied(seq INTEGER PRIMARY KEY); INSERT INTO applied VALUES(0);')
print(source.execute('SELECT * FROM stock').fetchall())

[(1, 10)]

backup_path = Path(tmp.name)/'backup.db'
backup = sqlite3.connect(backup_path)
source.backup(backup)
print(backup.execute('SELECT * FROM stock').fetchall())
backup.close()

[(1, 10)]

dump = '\n'.join(source.iterdump())
print(dump)

BEGIN TRANSACTION;
CREATE TABLE applied(seq INTEGER PRIMARY KEY);
INSERT INTO "applied" VALUES(0);
CREATE TABLE stock(id INTEGER PRIMARY KEY,qty INTEGER);
INSERT INTO "stock" VALUES(1,10);
COMMIT;

source.execute('UPDATE stock SET qty=15 WHERE id=1')
print('source:',source.execute('SELECT qty FROM stock').fetchone())
backup = sqlite3.connect(backup_path)
print('backup:',backup.execute('SELECT qty FROM stock').fetchone())
backup.close()

source: (15,)
backup: (10,)

physical = sqlite3.connect(':memory:',isolation_level=None)
backup = sqlite3.connect(backup_path)
backup.backup(physical)
backup.close()
logical = sqlite3.connect(':memory:',isolation_level=None)
logical.executescript(dump)
print('physical:',physical.execute('SELECT * FROM stock').fetchall())
print('logical:',logical.execute('SELECT * FROM stock').fetchall())

physical: [(1, 10)]
logical: [(1, 10)]

events = [(1,5),(2,-2),(3,7)]
print('seq,delta:',events)

seq,delta: [(1, 5), (2, -2), (3, 7)]

target = 2
for seq,delta in events:
    if seq>target:
        break
    physical.execute('BEGIN')
    added = physical.execute('INSERT OR IGNORE INTO applied VALUES(?)',(seq,)).rowcount
    if added:
        physical.execute('UPDATE stock SET qty=qty+? WHERE id=1',(delta,))
    physical.execute('COMMIT')
print(physical.execute('SELECT qty FROM stock').fetchone())
print(physical.execute('SELECT seq FROM applied ORDER BY seq').fetchall())

(13,)
[(0,), (1,), (2,)]

for seq,delta in events[:2]:
    physical.execute('BEGIN')
    added = physical.execute('INSERT OR IGNORE INTO applied VALUES(?)',(seq,)).rowcount
    if added:
        physical.execute('UPDATE stock SET qty=qty+? WHERE id=1',(delta,))
    physical.execute('COMMIT')
print(physical.execute('SELECT qty FROM stock').fetchone())

(13,)

seq,delta = events[2]
physical.execute('BEGIN')
physical.execute('INSERT INTO applied VALUES(?)',(seq,))
physical.execute('UPDATE stock SET qty=qty+? WHERE id=1',(delta,))
physical.execute('COMMIT')
print(physical.execute('SELECT qty FROM stock').fetchone())
print(physical.execute('PRAGMA integrity_check').fetchone())

(20,)
('ok',)

演習完了