第167回では、SQLiteのスキーマ変更を本番データベースへ直接適用せず、コピー上で実行しました。列定義や整合性を確認しても、移行後に予定している登録や状態更新が正しく動くとは限りません。
今回は、移行済みの候補データベースを読み取り専用で開き、OSの一時領域へ複製します。そのコピー上で、正常な登録と更新、制約違反の拒否、ロールバック後の無残留を確認します。元の候補データベースには書き込みません。
サンプルは、第167回の考え方を引き継いだ独立した動作確認用スキーマです。実際に使うときは、列名、状態遷移、外部キー、試験値を業務スキーマに合わせて変更してください。
構造検査と書き込み試験を分ける
構造検査では、必要な列やインデックスの存在を確認します。書き込み試験では、アプリケーションが使用するSQLと制約の組み合わせを確認します。
| 確認対象 | 今回の判定 |
|---|---|
| 正常な登録 | priorityを省略した登録が成功し、既定値0になる |
| 正常な更新 | 優先度3への変更と、確認済み状態への遷移が成功する |
| 範囲制約 | 優先度4への変更がsqlite3.IntegrityErrorになる |
| 一意性 | 同じバッチIDとレコードIDの再登録が拒否される |
| 状態制約 | 確定カテゴリがないままacceptedへ変更できない |
| 後始末 | 試験用トランザクションを取り消すと、行数と試験用IDが元に戻る |
| データベース検査 | integrity_check(1)とforeign_key_checkを通過する |
今回の実行契約
| 項目 | 仕様 |
|---|---|
| 動作環境 | Python 3.10以上、SQLite 3.37.0以上 |
| 入力 | 書き込みを停止した、単独の移行済みSQLiteファイル |
| 入力の扱い | mode=roとimmutable=1を指定して開き、SQLiteのバックアップAPIで一時ファイルへ複製する |
| 入力のサイドカー | -wal、-shm、-journalが残っている入力は拒否する |
| サンプル出力 | DB本体またはいずれかのサイドカー名が既に存在する場合は作成を拒否する |
| 書き込み試験の書き込み先 | TemporaryDirectory内のコピーだけ |
| 永続的な出力 | make-sampleで作る動作確認用DBを除き、なし |
| 結果 | ヘルプ表示を除き、標準出力または標準エラー出力へ1行のJSONを返す |
| 対象外 | 本番切り替え、アプリケーション全体の互換性、同時書き込み、性能試験、旧版への逆移行 |
-walファイルには、メインファイルへ反映されていない確定済みデータが含まれる場合があります。検査を通すためにサイドカーファイルを手動で削除してはいけません。書き込み元を正常に停止し、必要なチェックポイントや復旧をSQLiteに実行させてから候補ファイルを用意します。
make-sampleも、指定したDB本体だけでなく同名の-wal、-shm、-journalを作成前に検査します。既存のものが一つでもあれば処理を始めません。失敗時に削除するのは、排他的作成で今回生成したDB本体と同じファイルであることをデバイス番号とinode番号で確認できた場合だけです。既存または後から現れたサイドカーは削除しません。
この確認は、出力ディレクトリで別の処理が同じ名前を同時に作成、置換する状況を調停するロックではありません。サンプル作成中に同じ出力名を操作しない、信頼できるディレクトリで実行してください。
一時コピーにも入力データが含まれます。機密データを扱う場合は、一時領域の保存先と権限を確認してください。TemporaryDirectoryによる削除は、強制終了や電源断を含む安全な消去を保証しません。
スモークテストを実装する
次のコードをsmoke_review_db.pyとして保存します。外部ライブラリは使いません。
from __future__ import annotations
import argparse
import hashlib
import json
import os
import secrets
import sqlite3
import stat
import sys
import tempfile
from pathlib import Path
from typing import TextIO
EXPECTED_VERSION = 2
REQUIRED_COLUMNS = frozenset(
{
'batch_id',
'record_id',
'proposed_category',
'status',
'priority',
'decided_category',
}
)
SAMPLE_SQL = '''
BEGIN IMMEDIATE;
CREATE TABLE review_items (
batch_id TEXT NOT NULL CHECK (length(batch_id) > 0),
record_id TEXT NOT NULL CHECK (length(record_id) > 0),
proposed_category TEXT NOT NULL
CHECK (length(proposed_category) > 0),
status TEXT NOT NULL DEFAULT 'pending'
CHECK (
status IN ('pending', 'accepted', 'corrected', 'rejected')
),
priority INTEGER NOT NULL DEFAULT 0
CHECK (priority BETWEEN 0 AND 3),
decided_category TEXT,
PRIMARY KEY (batch_id, record_id),
CHECK (
(status = 'pending' AND decided_category IS NULL)
OR
(status IN ('accepted', 'corrected')
AND decided_category IS NOT NULL)
OR
(status = 'rejected' AND decided_category IS NULL)
)
) STRICT;
CREATE INDEX review_items_pending_priority_idx
ON review_items(priority DESC, batch_id, record_id)
WHERE status = 'pending';
INSERT INTO review_items (
batch_id, record_id, proposed_category, status
) VALUES (
'sample-batch', 'sample-record', 'sample-category', 'pending'
);
PRAGMA user_version = 2;
COMMIT;
'''
class ContractError(Exception):
pass
def json_line(payload: dict[str, object], stream: TextIO) -> None:
print(
json.dumps(payload, ensure_ascii=False, separators=(',', ':')),
file=stream,
)
class JsonArgumentParser(argparse.ArgumentParser):
def error(self, message: str) -> None:
json_line(
{'status': 'error', 'kind': 'argument', 'message': message},
sys.stderr,
)
raise SystemExit(2)
def require_sqlite() -> None:
if sqlite3.sqlite_version_info < (3, 37, 0):
raise ContractError(
f'SQLite 3.37.0以上が必要です: {sqlite3.sqlite_version}'
)
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open('rb') as source:
while chunk := source.read(1024 * 1024):
digest.update(chunk)
return digest.hexdigest()
def sidecar_paths(path: Path) -> tuple[Path, ...]:
return tuple(
Path(f'{path}{suffix}')
for suffix in ('-wal', '-shm', '-journal')
)
def existing_paths(paths: tuple[Path, ...]) -> list[Path]:
return [path for path in paths if os.path.lexists(path)]
def reject_sample_output_paths(path: Path) -> None:
present = existing_paths((path, *sidecar_paths(path)))
if present:
raise ContractError(
'出力先に既存ファイルがあります: '
+ ','.join(candidate.name for candidate in present)
)
def file_identity_from_fd(descriptor: int) -> tuple[int, int]:
information = os.fstat(descriptor)
return information.st_dev, information.st_ino
def has_file_identity(path: Path, identity: tuple[int, int]) -> bool:
try:
information = path.lstat()
except FileNotFoundError:
return False
return (
stat.S_ISREG(information.st_mode)
and (information.st_dev, information.st_ino) == identity
)
def cleanup_created_database(
path: Path,
identity: tuple[int, int],
) -> None:
if not has_file_identity(path, identity):
return
try:
path.unlink()
except FileNotFoundError:
pass
def make_sample(path: Path) -> dict[str, object]:
require_sqlite()
path = Path(os.path.abspath(path))
reject_sample_output_paths(path)
try:
descriptor = os.open(
path,
os.O_WRONLY | os.O_CREAT | os.O_EXCL,
0o600,
)
except FileExistsError as exc:
raise ContractError(
f'出力先が既に存在します: {path}'
) from exc
created_identity = file_identity_from_fd(descriptor)
descriptor_open = True
try:
os.close(descriptor)
descriptor_open = False
present_sidecars = existing_paths(sidecar_paths(path))
if present_sidecars:
raise ContractError(
'出力先にサイドカーファイルがあります: '
+ ','.join(
candidate.name for candidate in present_sidecars
)
)
connection = sqlite3.connect(path)
try:
journal = connection.execute(
'PRAGMA journal_mode = MEMORY'
).fetchone()
if journal is None or str(journal[0]).lower() != 'memory':
raise ContractError(
'サンプルDBをMEMORYジャーナルへ変更できません'
)
connection.execute('PRAGMA foreign_keys = ON')
connection.executescript(SAMPLE_SQL)
rows = int(
connection.execute(
'SELECT count(*) FROM review_items'
).fetchone()[0]
)
finally:
connection.close()
if not has_file_identity(path, created_identity):
raise ContractError('作成したDB本体が置き換えられました')
present_sidecars = existing_paths(sidecar_paths(path))
if present_sidecars:
raise ContractError(
'作成後にサイドカーファイルが見つかりました: '
+ ','.join(
candidate.name for candidate in present_sidecars
)
)
except BaseException:
if descriptor_open:
try:
os.close(descriptor)
except OSError:
pass
cleanup_created_database(path, created_identity)
raise
return {
'status': 'created',
'command': 'make-sample',
'database': path.name,
'user_version': EXPECTED_VERSION,
'rows': rows,
}
def reject_sidecars(path: Path) -> None:
present = existing_paths(sidecar_paths(path))
if present:
raise ContractError(
'入力にサイドカーファイルが残っています: '
+ ','.join(candidate.name for candidate in present)
)
def clone_read_only(source: Path, destination: Path) -> None:
source_uri = source.as_uri() + '?mode=ro&immutable=1'
source_connection = sqlite3.connect(source_uri, uri=True)
try:
source_connection.execute('PRAGMA query_only = ON')
destination_connection = sqlite3.connect(destination)
try:
source_connection.backup(destination_connection)
finally:
destination_connection.close()
finally:
source_connection.close()
def expect_integrity_error(
connection: sqlite3.Connection,
label: str,
sql: str,
parameters: tuple[object, ...],
) -> None:
rejected = False
connection.execute('SAVEPOINT expected_failure')
try:
connection.execute(sql, parameters)
except sqlite3.IntegrityError:
rejected = True
finally:
connection.execute('ROLLBACK TO expected_failure')
connection.execute('RELEASE expected_failure')
if not rejected:
raise ContractError(f'{label}が拒否されませんでした')
def check_integrity(connection: sqlite3.Connection) -> None:
result = connection.execute(
'PRAGMA integrity_check(1)'
).fetchone()
if result != ('ok',):
raise ContractError('integrity_checkを通過しませんでした')
violation = connection.execute(
'PRAGMA foreign_key_check'
).fetchone()
if violation is not None:
raise ContractError('外部キー違反があります')
def run_behavior_test(scratch: Path) -> tuple[int, list[str]]:
connection = sqlite3.connect(scratch, isolation_level=None)
try:
connection.execute('PRAGMA foreign_keys = ON')
foreign_keys = connection.execute(
'PRAGMA foreign_keys'
).fetchone()
if foreign_keys != (1,):
raise ContractError('外部キー制約を有効にできません')
journal = connection.execute(
'PRAGMA journal_mode = DELETE'
).fetchone()
if journal is None or str(journal[0]).lower() != 'delete':
raise ContractError('一時DBをDELETEモードへ変更できません')
version = int(
connection.execute('PRAGMA user_version').fetchone()[0]
)
if version != EXPECTED_VERSION:
raise ContractError(
f'user_version が{EXPECTED_VERSION}ではありません: '
f'{version}'
)
columns = {
row[1]
for row in connection.execute(
"PRAGMA table_xinfo('review_items')"
)
}
missing = sorted(REQUIRED_COLUMNS - columns)
if missing:
raise ContractError(
'必須列がありません: ' + ','.join(missing)
)
check_integrity(connection)
before_count = int(
connection.execute(
'SELECT count(*) FROM review_items'
).fetchone()[0]
)
for _ in range(5):
token = secrets.token_hex(16)
batch_id = f'__smoke__{token}'
record_id = 'record'
exists = connection.execute(
'''
SELECT 1
FROM review_items
WHERE batch_id = ? AND record_id = ?
''',
(batch_id, record_id),
).fetchone()
if exists is None:
break
else:
raise ContractError('試験用IDを確保できませんでした')
connection.execute('BEGIN IMMEDIATE')
try:
connection.execute(
'''
INSERT INTO review_items (
batch_id,
record_id,
proposed_category,
status
) VALUES (?, ?, ?, 'pending')
''',
(batch_id, record_id, 'smoke-category'),
)
row = connection.execute(
'''
SELECT status, priority, decided_category
FROM review_items
WHERE batch_id = ? AND record_id = ?
''',
(batch_id, record_id),
).fetchone()
if row != ('pending', 0, None):
raise ContractError('登録時の既定値が契約と異なります')
changed = connection.execute(
'''
UPDATE review_items
SET priority = 3
WHERE batch_id = ? AND record_id = ?
''',
(batch_id, record_id),
).rowcount
if changed != 1:
raise ContractError('優先度を1件更新できませんでした')
expect_integrity_error(
connection,
'priority範囲外の更新',
'''
UPDATE review_items
SET priority = 4
WHERE batch_id = ? AND record_id = ?
''',
(batch_id, record_id),
)
expect_integrity_error(
connection,
'主キーが重複する登録',
'''
INSERT INTO review_items (
batch_id,
record_id,
proposed_category,
status
) VALUES (?, ?, ?, 'pending')
''',
(batch_id, record_id, 'smoke-category'),
)
expect_integrity_error(
connection,
'確定カテゴリのないacceptedへの遷移',
'''
UPDATE review_items
SET status = 'accepted'
WHERE batch_id = ? AND record_id = ?
''',
(batch_id, record_id),
)
changed = connection.execute(
'''
UPDATE review_items
SET status = 'accepted',
decided_category = proposed_category
WHERE batch_id = ? AND record_id = ?
''',
(batch_id, record_id),
).rowcount
if changed != 1:
raise ContractError('正常な状態遷移を実行できませんでした')
row = connection.execute(
'''
SELECT status, priority, decided_category
FROM review_items
WHERE batch_id = ? AND record_id = ?
''',
(batch_id, record_id),
).fetchone()
if row != ('accepted', 3, 'smoke-category'):
raise ContractError('状態遷移後の値が契約と異なります')
finally:
connection.rollback()
after_count = int(
connection.execute(
'SELECT count(*) FROM review_items'
).fetchone()[0]
)
residue = connection.execute(
'''
SELECT 1
FROM review_items
WHERE batch_id = ? AND record_id = ?
''',
(batch_id, record_id),
).fetchone()
if after_count != before_count or residue is not None:
raise ContractError('ロールバック後に試験データが残りました')
check_integrity(connection)
finally:
connection.close()
return version, [
'foreign_keys_enabled',
'version_and_required_columns',
'valid_insert_priority_and_accept',
'priority_constraint',
'primary_key_constraint',
'state_constraint',
'rollback_without_residue',
'integrity_and_foreign_keys',
]
def test_database(path: Path) -> dict[str, object]:
require_sqlite()
if not path.is_file():
raise ContractError(f'入力DBが見つかりません: {path}')
source = path.resolve(strict=True)
reject_sidecars(source)
digest_before = sha256_file(source)
with tempfile.TemporaryDirectory(prefix='sqlite-smoke-') as directory:
scratch = Path(directory) / 'scratch.sqlite'
clone_read_only(source, scratch)
if sha256_file(source) != digest_before:
raise ContractError('複製中に入力DBの内容が変わりました')
version, behavior_checks = run_behavior_test(scratch)
reject_sidecars(source)
if sha256_file(source) != digest_before:
raise ContractError('試験中に入力DBの内容が変わりました')
return {
'status': 'passed',
'command': 'test',
'database': source.name,
'user_version': version,
'checks': [
'snapshot_created',
*behavior_checks,
'source_hash_unchanged',
],
}
def build_parser() -> JsonArgumentParser:
parser = JsonArgumentParser(
description='移行済みSQLiteの破棄可能な書き込み試験'
)
commands = parser.add_subparsers(dest='command', required=True)
sample = commands.add_parser(
'make-sample',
help='上書きせずに動作確認用DBを作る',
)
sample.add_argument('database', type=Path)
test = commands.add_parser(
'test',
help='入力を一時領域へ複製して試験する',
)
test.add_argument('database', type=Path)
return parser
def main() -> int:
arguments = build_parser().parse_args()
try:
if arguments.command == 'make-sample':
result = make_sample(arguments.database)
else:
result = test_database(arguments.database)
except KeyboardInterrupt:
json_line(
{'status': 'error', 'kind': 'interrupted'},
sys.stderr,
)
return 130
except ContractError as exc:
json_line(
{
'status': 'error',
'kind': 'contract',
'message': str(exc),
},
sys.stderr,
)
return 2
except (OSError, sqlite3.Error) as exc:
json_line(
{
'status': 'error',
'kind': 'runtime',
'message': str(exc),
},
sys.stderr,
)
return 3
except Exception as exc:
json_line(
{
'status': 'error',
'kind': 'unexpected',
'message': f'{type(exc).__name__}: {exc}',
},
sys.stderr,
)
return 3
json_line(result, sys.stdout)
return 0
if __name__ == '__main__':
raise SystemExit(main())
サンプル作成では、最初にDB本体と三種類のサイドカー名を検査し、DB本体をO_EXCLで排他的に作成します。作成中はディスク上のロールバックジャーナルを必要としないMEMORYモードを使い、接続を閉じた後にもサイドカーがないことを確認します。
作成後のDB本体が、排他的作成時に記録したデバイス番号とinode番号と一致しなければ成功にしません。失敗時も同じ照合を通過したDB本体だけを削除するため、以前からあったサイドカーを後始末の対象に含めません。
制約違反の確認では、各SQLをセーブポイント内で実行します。予定どおり拒否された場合だけでなく、誤って成功した場合にも、そのSQLを取り消してからテストを失敗させます。最後に試験全体のトランザクションも取り消します。
動作確認用データベースで実行する
最初に、記事内の契約と一致するサンプルを新しいファイルとして作ります。candidate.sqlite本体またはcandidate.sqlite-wal、candidate.sqlite-shm、candidate.sqlite-journalのいずれかが既にある場合、make-sampleは削除も上書きもせず、終了コード2で停止します。
python smoke_review_db.py make-sample candidate.sqlite成功時の出力は次の1行です。
{"status":"created","command":"make-sample","database":"candidate.sqlite","user_version":2,"rows":1}続いて、候補データベースを一時領域へ複製して試験します。
python smoke_review_db.py test candidate.sqliteすべての検査を通過すると、次の1行が返ります。
{"status":"passed","command":"test","database":"candidate.sqlite","user_version":2,"checks":["snapshot_created","foreign_keys_enabled","version_and_required_columns","valid_insert_priority_and_accept","priority_constraint","primary_key_constraint","state_constraint","rollback_without_residue","integrity_and_foreign_keys","source_hash_unchanged"]}| 終了コード | 意味 |
|---|---|
| 0 | サンプル作成またはスモークテストが完了した |
| 2 | 引数、スキーマ、制約、既存の出力、サイドカーファイルなどが契約を満たさない |
| 3 | ファイル操作、SQLiteの実行、または予期しない処理で失敗した |
| 130 | Ctrl+Cで中断した |
既存サイドカーを拒否することを確認する
次の例では空のblocked.sqlite-walを先に作ります。make-sampleはDB本体を作成せず、既存のサイドカーも削除しません。
python -c "open('blocked.sqlite-wal', 'xb').close()"
python smoke_review_db.py make-sample blocked.sqlite標準エラー出力へ次のJSONを返し、終了コード2で停止します。
{"status":"error","kind":"contract","message":"出力先に既存ファイルがあります: blocked.sqlite-wal"}この空ファイルは拒否動作だけを確認するためのものです。実際の候補DBに付随するサイドカーを、この手順で作成または削除してはいけません。
バージョン不一致を確認する
別のサンプルを作り、検査対象のuser_versionだけを1へ変更します。
python smoke_review_db.py make-sample wrong.sqlite
python -c "import sqlite3; db=sqlite3.connect('wrong.sqlite'); db.execute('PRAGMA user_version=1'); db.commit(); db.close()"
python smoke_review_db.py test wrong.sqlitetestは書き込み試験へ進まず、標準エラー出力へ次のJSONを返して終了コード2で停止します。
{"status":"error","kind":"contract","message":"user_version が2ではありません: 1"}実際の移行済みDBへ合わせる箇所
EXPECTED_VERSIONを切り替え対象のスキーマバージョンへ合わせるREQUIRED_COLUMNSと登録・更新・参照SQLを実際の列名へ置き換える- 成功させる操作と拒否すべき操作を業務上の状態遷移に合わせる
- 外部キーの親データが必要なら、試験用の親行も同じトランザクション内で登録し、最後に取り消す
- アプリケーションと同じSQLite実装を確認する場合は、そのアプリケーションと同じPython環境で実行する
実在する顧客IDや担当者IDを推測して試験値に使ってはいけません。専用の試験値を同じトランザクション内で作れないスキーマでは、匿名化した検証用データベースを別途用意します。
合格結果が保証しないこと
このスクリプトが確認するのは、現在のPythonに組み込まれたSQLiteで、指定したSQLと制約が一時コピー上で動くことです。アプリケーションの画面、認証、外部API、添付データベース、同時実行時の待ち時間までは確認しません。
入力ファイルのSHA-256比較は、試験中の内容変更を検出するための確認です。書き込みを排除するロックでも、改ざんを証明する仕組みでもありません。
immutable=1を指定すると、SQLiteは入力が変化しないものとして扱い、ロックや変更検出を省略します。入力が実際に変化すると、正しくない結果を返す可能性があります。実行前に書き込み元を停止し、サイドカーファイルがないことを確認する運用が前提です。
また、移行後のデータベースへ新しい業務データを書き込んだ後で古いファイルへ戻すと、その間の更新が失われます。今回のスモークテストは切り替え前の受け入れ判定であり、切り替え後の逆移行手順ではありません。
仕様を確認する一次資料
- Python sqlite3のバックアップAPI
- SQLiteのURIファイル名とimmutable指定
- SQLiteのWALとサイドカーファイル
- SQLiteのPRAGMA
- SQLiteのSTRICTテーブル
次の一歩
実DBへ適用する前に、専用の検証コピーでSQLと試験値を調整します。内部の正常操作と拒否検査がすべて期待どおりなら、testコマンド全体が終了コード0になります。
本番切り替え手順には、書き込み停止、最終バックアップ、移行、スモークテスト、切り替え可否の判断を別々の工程として記録します。