第164回では、AIレビュー項目を期限付きで担当者へ割り当てる仕組みを作りました。レビューキューが運用データになると、次に必要なのは障害や操作ミスに備えたバックアップです。
SQLiteのデータベースを、稼働中に通常のファイルコピーだけで退避するのは避けます。WALモードでは確定済みの変更が別のファイルに残っている場合があり、コピー中に書き込みが発生する可能性もあるためです。
今回はPython標準ライブラリのsqlite3.Connection.backup()を使い、SQLiteが管理する整合したスナップショットを新しいファイルへ保存します。保存後にはデータベース構造、外部キー、SHA-256を検査し、結果をマニフェストへ記録します。特定のレビューキューの列構成には依存しません。
Connection.backup()は、ロックや並行更新が続くと再試行を繰り返し、完了しない可能性があります。接続時のtimeout=5.0はバックアップ処理全体の期限ではありません。この実装では、必須引数--deadline-secondsとprogressコールバック、単調時計を使って経過時間を確認し、期限を超えた実行を例外で終了させます。
今回固定するバックアップ契約
| 項目 | 仕様 |
|---|---|
| 入力 | ローカルディスク上に存在するSQLiteのメインデータベース |
| 入力の扱い | mode=roで開き、元データベースへ書き込まない |
| 出力 | 新しいSQLiteファイル。既存ファイルは上書きしない |
| 完了期限 | --deadline-secondsで正の有限秒数を必ず指定し、超過をprogressコールバックで検出する |
| マニフェスト | 出力ファイル名、SHA-256、サイズ、ページ情報、検査結果を隣接するJSONへ保存する |
| 検査 | integrity_check(1)とforeign_key_checkを実行する |
| 検証 | verifyコマンドでハッシュ、サイズ、ページ情報、SQLiteの検査をやり直す |
| 対象外 | 添付データベース、外部ファイル、未確定トランザクション、業務上の値の正しさ |
integrity_check(1)の引数1は、構造上の問題を最初の1件まで報告する指定です。外部キー違反はこの検査だけでは確認できないため、foreign_key_checkも別に実行します。
SHA-256は転送ミスや意図しない変更の検出には使えますが、マニフェスト自体も同時に書き換えられる環境では改ざんの証明になりません。攻撃者による変更まで検出する必要がある場合は、ハッシュをバックアップとは別のアクセス管理下に保管する設計が必要です。
バックアップCLIを実装する
Python 3.10以上を前提とします。次のコードをsqlite_snapshot.pyとして保存してください。バックアップ先とマニフェストは、作成できる権限を持つ既存ディレクトリに指定します。
from __future__ import annotations
import argparse
import hashlib
import json
import math
import os
import re
import sqlite3
import sys
import time
from contextlib import closing
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
FORMAT = 'sqlite-snapshot-v1'
MANIFEST_LIMIT = 16 * 1024
class ToolError(Exception):
def __init__(
self, message: str, exit_code: int, error: str
) -> None:
super().__init__(message)
self.exit_code = exit_code
self.error = error
def emit(payload: dict[str, Any], stream: Any) -> None:
print(
json.dumps(
payload, ensure_ascii=False, separators=(',', ':')
),
file=stream,
)
def positive_finite_seconds(value: str) -> float:
try:
seconds = float(value)
except ValueError as exc:
raise argparse.ArgumentTypeError(
'正の有限秒数を指定してください'
) from exc
if not math.isfinite(seconds) or seconds <= 0:
raise argparse.ArgumentTypeError(
'正の有限秒数を指定してください'
)
return seconds
def resolve_existing_file(path: Path) -> Path:
try:
resolved = path.expanduser().resolve(strict=True)
except FileNotFoundError as exc:
raise ToolError(
f'ファイルが存在しません: {path}', 2, 'input_error'
) from exc
except OSError as exc:
raise ToolError(
f'パスを確認できません: {path}', 3, 'io_error'
) from exc
if not resolved.is_file():
raise ToolError(
f'通常ファイルではありません: {resolved}',
2,
'input_error',
)
return resolved
def resolve_new_file(path: Path) -> Path:
candidate = path.expanduser()
try:
if candidate.exists() or candidate.is_symlink():
raise ToolError(
f'既存ファイルです: {candidate}', 2, 'input_error'
)
resolved = candidate.resolve(strict=False)
except ToolError:
raise
except OSError as exc:
raise ToolError(
f'出力パスを確認できません: {candidate}',
3,
'io_error',
) from exc
if not resolved.parent.is_dir():
raise ToolError(
f'出力ディレクトリが存在しません: {resolved.parent}',
2,
'input_error',
)
if resolved.exists():
raise ToolError(
f'既存ファイルです: {resolved}', 2, 'input_error'
)
return resolved
def read_only_connection(path: Path) -> sqlite3.Connection:
uri = path.as_uri() + '?mode=ro'
return sqlite3.connect(uri, uri=True, timeout=5.0)
def inspect_database(path: Path) -> dict[str, int]:
with closing(read_only_connection(path)) as database:
database.execute('PRAGMA query_only = ON')
integrity = database.execute(
'PRAGMA integrity_check(1)'
).fetchone()
if integrity != ('ok',):
raise ToolError(
'SQLiteの整合性検査に失敗しました',
4,
'verification_failed',
)
foreign_key_problem = database.execute(
'PRAGMA foreign_key_check'
).fetchone()
if foreign_key_problem is not None:
raise ToolError(
'外部キー違反を検出しました',
4,
'verification_failed',
)
return {
'page_count': int(
database.execute('PRAGMA page_count').fetchone()[0]
),
'page_size': int(
database.execute('PRAGMA page_size').fetchone()[0]
),
'user_version': int(
database.execute('PRAGMA user_version').fetchone()[0]
),
}
def file_signature(
value: os.stat_result,
) -> tuple[int, int, int, int]:
return (
value.st_dev,
value.st_ino,
value.st_size,
value.st_mtime_ns,
)
def hash_file(
path: Path,
) -> tuple[str, int, tuple[int, int, int, int]]:
digest = hashlib.sha256()
with path.open('rb') as handle:
before = file_signature(os.fstat(handle.fileno()))
while True:
chunk = handle.read(1024 * 1024)
if not chunk:
break
digest.update(chunk)
after = file_signature(os.fstat(handle.fileno()))
if before != after:
raise ToolError(
'ハッシュ計算中にファイルが変更されました',
4,
'verification_failed',
)
return digest.hexdigest(), after[2], after
def create_exclusive_file(path: Path) -> int:
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
try:
return os.open(path, flags, 0o600)
except FileExistsError as exc:
raise ToolError(
f'既存ファイルです: {path}', 2, 'input_error'
) from exc
except OSError as exc:
raise ToolError(
f'ファイルを作成できません: {path}', 3, 'io_error'
) from exc
def cleanup_created(paths: list[Path]) -> None:
for path in reversed(paths):
try:
path.unlink()
except FileNotFoundError:
pass
except OSError:
pass
def copy_snapshot(
source: Path,
destination: Path,
deadline_seconds: float,
) -> None:
started = time.monotonic()
def check_deadline(
status: int, remaining: int, total: int
) -> None:
del status, remaining, total
if time.monotonic() - started >= deadline_seconds:
raise ToolError(
'バックアップが完了期限を超えました',
5,
'backup_deadline_exceeded',
)
with closing(read_only_connection(source)) as source_database:
source_database.execute('PRAGMA query_only = ON')
with closing(
sqlite3.connect(destination, timeout=5.0)
) as destination_database:
destination_database.execute('PRAGMA synchronous = FULL')
source_database.backup(
destination_database,
pages=256,
progress=check_deadline,
sleep=0.1,
)
destination_database.commit()
journal_mode = destination_database.execute(
'PRAGMA journal_mode = DELETE'
).fetchone()
if (
journal_mode is None
or str(journal_mode[0]).lower() != 'delete'
):
raise ToolError(
'出力をDELETEジャーナルへ変更できませんでした',
3,
'sqlite_error',
)
destination_database.commit()
def create_backup(
raw_source: Path,
raw_destination: Path,
deadline_seconds: float,
) -> dict[str, Any]:
source = resolve_existing_file(raw_source)
destination = resolve_new_file(raw_destination)
manifest = resolve_new_file(
Path(str(destination) + '.manifest.json')
)
if source == destination or source == manifest:
raise ToolError(
'入力と出力には別のパスを指定してください',
2,
'input_error',
)
created_paths: list[Path] = []
try:
destination_fd = create_exclusive_file(destination)
created_paths.append(destination)
os.close(destination_fd)
copy_snapshot(source, destination, deadline_seconds)
database_info = inspect_database(destination)
digest, size_bytes, _ = hash_file(destination)
document = {
'format': FORMAT,
'created_at_utc': datetime.now(
timezone.utc
).strftime('%Y-%m-%dT%H:%M:%SZ'),
'backup_name': destination.name,
'sha256': digest,
'size_bytes': size_bytes,
**database_info,
'checks': {
'integrity_check': 'ok',
'foreign_key_check': 'ok',
},
}
manifest_fd = create_exclusive_file(manifest)
created_paths.append(manifest)
with os.fdopen(
manifest_fd, 'w', encoding='utf-8'
) as handle:
json.dump(
document,
handle,
ensure_ascii=False,
sort_keys=True,
separators=(',', ':'),
)
handle.write(chr(10))
handle.flush()
os.fsync(handle.fileno())
return {
'ok': True,
'action': 'backup',
'backup': str(destination),
'manifest': str(manifest),
'sha256': digest,
}
except BaseException:
cleanup_created(created_paths)
raise
def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
result: dict[str, Any] = {}
for key, value in pairs:
if key in result:
raise ValueError(f'重複キーです: {key}')
result[key] = value
return result
def reject_constant(value: str) -> Any:
raise ValueError(f'許可しない数値定数です: {value}')
def read_manifest(path: Path) -> dict[str, Any]:
if path.stat().st_size > MANIFEST_LIMIT:
raise ToolError(
'マニフェストが上限サイズを超えています',
4,
'verification_failed',
)
try:
with path.open('r', encoding='utf-8') as handle:
document = json.load(
handle,
object_pairs_hook=unique_object,
parse_constant=reject_constant,
)
except (UnicodeError, json.JSONDecodeError, ValueError) as exc:
raise ToolError(
'マニフェストを解釈できません',
4,
'verification_failed',
) from exc
if type(document) is not dict:
raise ToolError(
'マニフェストの最上位はオブジェクトではありません',
4,
'verification_failed',
)
expected_fields = {
'format',
'created_at_utc',
'backup_name',
'sha256',
'size_bytes',
'page_count',
'page_size',
'user_version',
'checks',
}
if set(document) != expected_fields:
raise ToolError(
'マニフェストの項目が契約と一致しません',
4,
'verification_failed',
)
if document['format'] != FORMAT:
raise ToolError(
'未対応のマニフェスト形式です',
4,
'verification_failed',
)
if type(document['backup_name']) is not str:
raise ToolError(
'backup_nameの型が不正です', 4, 'verification_failed'
)
if (
type(document['sha256']) is not str
or re.fullmatch('[0-9a-f]{64}', document['sha256']) is None
):
raise ToolError(
'sha256の形式が不正です', 4, 'verification_failed'
)
for key in ('size_bytes', 'page_count', 'page_size', 'user_version'):
if type(document[key]) is not int:
raise ToolError(
f'{key}の型が不正です', 4, 'verification_failed'
)
if (
document['size_bytes'] < 0
or document['page_count'] < 0
or document['page_size'] <= 0
):
raise ToolError(
'マニフェストの数値範囲が不正です',
4,
'verification_failed',
)
if document['checks'] != {
'integrity_check': 'ok',
'foreign_key_check': 'ok',
}:
raise ToolError(
'マニフェストの検査結果が不正です',
4,
'verification_failed',
)
if type(document['created_at_utc']) is not str:
raise ToolError(
'created_at_utcの型が不正です',
4,
'verification_failed',
)
try:
datetime.strptime(
document['created_at_utc'], '%Y-%m-%dT%H:%M:%SZ'
)
except ValueError as exc:
raise ToolError(
'created_at_utcの形式が不正です',
4,
'verification_failed',
) from exc
return document
def verify_backup(raw_backup: Path) -> dict[str, Any]:
backup = resolve_existing_file(raw_backup)
manifest = resolve_existing_file(
Path(str(backup) + '.manifest.json')
)
document = read_manifest(manifest)
if document['backup_name'] != backup.name:
raise ToolError(
'バックアップ名がマニフェストと一致しません',
4,
'verification_failed',
)
digest, size_bytes, signature = hash_file(backup)
if digest != document['sha256'] or size_bytes != document['size_bytes']:
raise ToolError(
'ファイルのハッシュまたはサイズが一致しません',
4,
'verification_failed',
)
database_info = inspect_database(backup)
for key in ('page_count', 'page_size', 'user_version'):
if database_info[key] != document[key]:
raise ToolError(
f'{key}がマニフェストと一致しません',
4,
'verification_failed',
)
if file_signature(backup.stat()) != signature:
raise ToolError(
'検証中にバックアップが変更されました',
4,
'verification_failed',
)
return {
'ok': True,
'action': 'verify',
'backup': str(backup),
'manifest': str(manifest),
'sha256': digest,
}
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description='SQLiteを新規ファイルへバックアップして検証する'
)
subparsers = parser.add_subparsers(dest='command', required=True)
backup_parser = subparsers.add_parser(
'backup', help='バックアップとマニフェストを作成する'
)
backup_parser.add_argument(
'--deadline-seconds',
type=positive_finite_seconds,
required=True,
help='バックアップ処理の完了期限を秒数で指定する',
)
backup_parser.add_argument('source', type=Path)
backup_parser.add_argument('destination', type=Path)
backup_parser.set_defaults(
handler=lambda args: create_backup(
args.source,
args.destination,
args.deadline_seconds,
)
)
verify_parser = subparsers.add_parser(
'verify', help='バックアップを再検証する'
)
verify_parser.add_argument('backup', type=Path)
verify_parser.set_defaults(
handler=lambda args: verify_backup(args.backup)
)
return parser
def main() -> int:
args = build_parser().parse_args()
try:
result = args.handler(args)
except ToolError as exc:
emit(
{
'ok': False,
'error': exc.error,
'message': str(exc),
},
sys.stderr,
)
return exc.exit_code
except KeyboardInterrupt:
emit(
{
'ok': False,
'error': 'interrupted',
'message': '処理を中断しました',
},
sys.stderr,
)
return 130
except sqlite3.Error as exc:
emit(
{
'ok': False,
'error': 'sqlite_error',
'message': str(exc),
},
sys.stderr,
)
return 3
except OSError as exc:
emit(
{
'ok': False,
'error': 'io_error',
'message': str(exc),
},
sys.stderr,
)
return 3
except Exception as exc:
emit(
{
'ok': False,
'error': 'internal_error',
'message': type(exc).__name__,
},
sys.stderr,
)
return 3
emit(result, sys.stdout)
return 0
if __name__ == '__main__':
raise SystemExit(main())
出力先は先に排他的に作成するため、同名ファイルが存在すれば停止します。作成途中で通常の例外、期限超過、Ctrl+Cが発生した場合は、この実行で作った出力だけを削除します。ただし、電源断やプロセスの強制終了では途中ファイルが残る可能性があります。成功時のJSONとマニフェストがそろうまでは、完成したバックアップとして扱わないでください。
完了期限の扱い
timeout=5.0は、SQLite接続のロック待ちに関する設定であり、backup()全体を5秒で打ち切る設定ではありません。バックアップAPIはSQLITE_BUSYまたはSQLITE_LOCKEDを受けると、指定したsleepの後で再試行します。並行更新による再開始が続く場合も、処理が完了しない可能性があります。
このコードはtime.monotonic()で経過時間を測り、バックアップAPIからprogressコールバックが呼ばれた時点で期限超過を検出します。検出するとToolErrorを送出し、終了コード5を返して、この実行で作成した出力を削除します。
期限の確認はprogressコールバックが呼ばれるタイミングで行うため、指定秒数ちょうどにプロセスを強制停止する仕組みではありません。プロセスを厳密な時刻で終了させる必要がある運用では、この内部期限に加えて、実行環境側でも監視時間を定めます。
再現用の小さなデータベースを作る
動作確認用として、個人情報を含まない小さなレビューキューを作ります。固定した時刻はテストデータであり、現在時刻を表すものではありません。次をmake_demo_db.pyとして保存します。
import json
import os
import sqlite3
from contextlib import closing
from pathlib import Path
path = Path('review-demo.sqlite3')
try:
file_descriptor = os.open(
path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600
)
except FileExistsError:
raise SystemExit('review-demo.sqlite3 はすでに存在します') from None
os.close(file_descriptor)
try:
with closing(sqlite3.connect(path)) as database:
database.execute('PRAGMA foreign_keys = ON')
database.executescript(
'''
CREATE TABLE review_items (
item_id INTEGER PRIMARY KEY,
record_id TEXT NOT NULL UNIQUE,
status TEXT NOT NULL CHECK (
status IN (
'pending', 'accepted', 'corrected', 'rejected'
)
)
);
CREATE TABLE review_events (
event_id INTEGER PRIMARY KEY,
item_id INTEGER NOT NULL
REFERENCES review_items(item_id),
event_type TEXT NOT NULL CHECK (
event_type IN ('enqueued', 'decided')
),
occurred_at INTEGER NOT NULL
);
PRAGMA user_version = 1;
'''
)
database.executemany(
'INSERT INTO review_items VALUES (?, ?, ?)',
[
(1, 'R-001', 'pending'),
(2, 'R-002', 'accepted'),
],
)
database.executemany(
'INSERT INTO review_events VALUES (?, ?, ?, ?)',
[
(1, 1, 'enqueued', 1700000000),
(2, 2, 'enqueued', 1700000000),
(3, 2, 'decided', 1700000060),
],
)
database.commit()
except BaseException:
path.unlink(missing_ok=True)
raise
print(
json.dumps(
{'ok': True, 'database': str(path)},
ensure_ascii=False,
separators=(',', ':'),
)
)
実行順を確認する
- Pythonのバージョンを確認します。
python --version- デモ用データベースを作ります。
python make_demo_db.py- 空の出力ディレクトリを作り、バックアップします。次の30秒は動作確認用の指定です。実運用ではデータベースの大きさ、書き込み状況、失敗を検知するまでの許容時間に合わせて明示的に決めます。
mkdir backup-out
python sqlite_snapshot.py backup --deadline-seconds 30 review-demo.sqlite3 backup-out/review-demo.sqlite3成功時は標準出力へ1行のJSONを返し、終了コードは0になります。backup、manifest、sha256には、その実行で生成した値が入ります。SQLiteのバージョンやファイル内容によってハッシュは変わるため、記事内では固定値を示しません。
- 作成した成果物を別の読み取りとして再検証します。
python sqlite_snapshot.py verify backup-out/review-demo.sqlite3verifyは、隣接するreview-demo.sqlite3.manifest.jsonを自動的に読みます。バックアップとマニフェストの名前を個別に変更すると対応関係を確認できなくなるため、2ファイルを同じ名前の組として扱います。
- 同じ出力先でもう一度バックアップし、上書きされないことを確認します。
python sqlite_snapshot.py backup --deadline-seconds 30 review-demo.sqlite3 backup-out/review-demo.sqlite3この実行は終了コード2になり、既存のバックアップとマニフェストを変更しません。新しいバックアップには、日時や連番を含む別名を運用側で指定します。
終了コードと後続処理
| 終了コード | 意味 | 対応 |
|---|---|---|
| 0 | バックアップまたは検証が完了 | 成果物を保管工程へ渡す |
| 2 | 入力パス、出力先、期限、既存ファイルなどの指定不備 | 引数と保存先を確認する |
| 3 | ファイル操作またはSQLite処理の失敗 | 権限、空き容量、ロック状況、元DBを確認する |
| 4 | ハッシュ、構造、外部キー、マニフェストの検証失敗 | 成果物を使用せず、原因を調査する |
| 5 | バックアップの完了期限を超過 | 成果物を使用せず、書き込み状況と期限設定を確認する |
| 130 | Ctrl+Cによる中断 | 途中ファイルの有無を確認して新しい名前で再実行する |
argparseが検出する引数不足、正でない期限、有限でない期限、未知のオプションは、標準の使用方法を標準エラー出力へ表示して終了コード2を返します。それ以外の実行結果は、標準出力または標準エラー出力へ1行のJSONとして返します。
この実装で保証しないこと
- バックアップ中に確定した書き込みはSQLiteが整合する状態で処理しますが、開始時刻と完全に同一の瞬間を指定する仕組みではありません。
- ロックや並行更新が続く場合、バックアップAPIは再試行を続ける可能性があります。この実装は
progressコールバックで期限超過を検出しますが、指定秒数ちょうどにプロセスを強制終了するものではありません。 - 新しい接続の
mainデータベースだけを対象とします。別ファイルとして管理する添付DB、CSV、AIの元レスポンス、添付資料は個別に退避します。 - 構造検査と外部キー検査は、レビュー判断や業務ルールの正しさを証明しません。必要な件数や状態別件数は、復旧訓練でアプリケーション固有の条件として確認します。
- 暗号化は行いません。出力時に可能な環境では所有者だけが読み書きできるモードで作成しますが、OSや保存先のアクセス制御も別に設定します。
- バックアップとマニフェストの作成は、2ファイルをまたぐ単一の原子的操作ではありません。処理中の出力先を別プロセスから読み書きしない運用が前提です。
実運用へ入れる前の確認
- 元データベースと異なる障害範囲の保存先へ、バックアップとマニフェストを組で保管する
- バックアップの内部期限と、必要であれば実行環境側の監視時間を決める
- 保存期間、実行頻度、担当者、失敗時の連絡先を決める
verifyの成功だけで終わらせず、新しいテスト用パスで定期的に復旧確認を行う- 復旧確認ではレビュー総数、未確定件数、確定済み件数など、業務側の期待値も照合する
- 稼働中の正本をこのCLIから自動的に置き換えない
期限付きロックやレビュー結果を安全に運用するには、処理コードだけでなく、終了しない処理を検知する仕組みと、壊れたときに戻せる手順が必要です。まずはデモDBでバックアップと検証を通し、その後に完了期限、権限、保存先、復旧時の業務確認を含めた手順へ落とし込んでください。