第166回では、SQLiteバックアップを隔離環境へ復元し、ハッシュ、整合性、外部キー、スキーマバージョンを検査しました。復元できることを確認したら、スキーマ変更も本番データベースへ直接適用せず、コピー上で試します。
AIレビュー項目へ優先度を追加するときは、入力のデータベースを直接変更しないことが大切です。入力は読み取り専用で開き、新しいSQLiteファイルへコピーします。そのコピーに対して、列追加、インデックス作成、バージョン更新を一つのトランザクションで実行します。
サンプルのテーブルは動作確認用です。第163回以降のデータベースへそのまま適用するパッチではありません。実データへ使う場合は、列定義とDDLを実際のスキーマに合わせ、第166回で作ったような隔離コピーで検証してください。
移行の契約を固定する
| 項目 | 仕様 |
|---|---|
| 入力 | PRAGMA user_versionが1で、所定のreview_items列を持つSQLite |
| 入力の扱い | 更新を止めた隔離コピーをmode=roで開き、SQLによる書き込みを行わない |
| 出力 | 実行前には存在しない新しいSQLiteファイル。既存ファイルは上書きしない |
| 変更内容 | 0から3までのpriority列と、未処理項目を優先順に読むための部分インデックスを追加する |
| 既存行 | 業務上の優先度を推測せず、すべて0から開始する |
| 移行単位 | 列追加、インデックス作成、user_version=2を一つのトランザクションで確定する |
| 検査 | 列定義、インデックス定義、行数、初期優先度、整合性、外部キーを確認する |
| 失敗時 | トランザクションを取り消し、作成した出力ファイルを削除する |
| 対象外 | アプリケーションの切り替え、優先度の業務判定、同時マイグレーション、v2からv1への逆変換 |
動作環境はPython 3.10以上、SQLite 3.37.0以上とします。SQLite 3.37.0では、ADD COLUMNで追加するCHECK制約を既存行に対して検査する動作が導入されました。詳細はSQLiteのALTER TABLE仕様で確認できます。
Pythonが使用するSQLiteの版は、次のコマンドで確認します。
python3 -c "import sqlite3; print(sqlite3.sqlite_version)"コピーとトランザクションの役割
新しいファイルへコピーするのは、移行内容に誤りがあっても入力を残すためです。トランザクションは、出力側で列だけが追加され、インデックスやバージョン更新が欠けた状態を確定させないために使います。
user_versionはアプリケーションが管理する整数であり、SQLiteが自動更新する値ではありません。値が1であることだけを信用せず、PRAGMA table_xinfoで列名、型、NULL制約、既定値、主キー、隠し列の区分を照合します。user_versionの仕様はSQLiteのPRAGMA文書で確認できます。
1．動作確認用のv1データベースを作る
次のコードをcreate_review_v1.pyとして保存します。2件のデータは移行結果を確かめるためのサンプルです。
from __future__ import annotations
import json
import os
import sqlite3
from contextlib import closing
from pathlib import Path
path = Path("review_v1.sqlite")
try:
fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
except FileExistsError:
raise SystemExit("review_v1.sqlite already exists")
os.close(fd)
completed = False
try:
with closing(sqlite3.connect(path)) as conn:
conn.execute("PRAGMA foreign_keys = ON")
conn.execute(
"""
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')),
created_at TEXT NOT NULL
)
"""
)
conn.executemany(
"""
INSERT INTO review_items(record_id, status, created_at)
VALUES (?, ?, ?)
""",
[
("R-001", "pending", "2026-01-10T09:00:00Z"),
("R-002", "accepted", "2026-01-10T09:05:00Z"),
],
)
conn.execute("PRAGMA user_version = 1")
conn.commit()
completed = True
finally:
if not completed:
path.unlink(missing_ok=True)
print(
json.dumps(
{"status": "created", "database": str(path), "rows": 2},
ensure_ascii=False,
separators=(",", ":"),
)
)実行すると、新しいreview_v1.sqliteが作られます。同名ファイルがある場合は上書きしません。
python3 create_review_v1.py{"status":"created","database":"review_v1.sqlite","rows":2}2．コピー上でv1からv2へ移行する
次のコードをmigrate_review_db.pyとして保存します。標準ライブラリ以外は使用しません。
from __future__ import annotations
import argparse
import json
import os
import sqlite3
import sys
from contextlib import closing
from pathlib import Path
from typing import Any, TextIO
FROM_VERSION = 1
TO_VERSION = 2
MIN_SQLITE = (3, 37, 0)
INDEX_NAME = "ix_review_items_pending_priority"
INDEX_SQL = """
CREATE INDEX ix_review_items_pending_priority
ON review_items(priority DESC, created_at, item_id)
WHERE status = 'pending'
"""
V1_COLUMNS = [
("item_id", "INTEGER", 0, None, 1, 0),
("record_id", "TEXT", 1, None, 0, 0),
("status", "TEXT", 1, None, 0, 0),
("created_at", "TEXT", 1, None, 0, 0),
]
V2_COLUMNS = V1_COLUMNS + [
("priority", "INTEGER", 1, "0", 0, 0),
]
class ContractError(Exception):
pass
def emit(stream: TextIO, payload: dict[str, Any]) -> None:
print(
json.dumps(payload, ensure_ascii=False, separators=(",", ":")),
file=stream,
)
class JsonParser(argparse.ArgumentParser):
def error(self, message: str) -> None:
emit(sys.stderr, {"status": "rejected", "message": message})
raise SystemExit(2)
def readonly_uri(path: Path) -> str:
return path.as_uri() + "?mode=ro"
def column_signature(conn: sqlite3.Connection) -> list[tuple[Any, ...]]:
rows = conn.execute("PRAGMA table_xinfo('review_items')").fetchall()
return [
(
str(row[1]),
str(row[2]).upper(),
int(row[3]),
row[4],
int(row[5]),
int(row[6]),
)
for row in rows
]
def normalize_sql(sql: str) -> str:
return " ".join(sql.split())
def validate_contract(conn: sqlite3.Connection, version: int) -> None:
actual_version = int(conn.execute("PRAGMA user_version").fetchone()[0])
if actual_version != version:
raise ContractError(
f"user_versionが{version}ではありません: {actual_version}"
)
expected = V1_COLUMNS if version == FROM_VERSION else V2_COLUMNS
if column_signature(conn) != expected:
raise ContractError(f"review_itemsの列定義がv{version}契約と一致しません")
index_row = conn.execute(
"""
SELECT tbl_name, sql
FROM sqlite_schema
WHERE type = 'index' AND name = ?
""",
(INDEX_NAME,),
).fetchone()
if version == FROM_VERSION and index_row is not None:
raise ContractError(f"v1に予約済みインデックス名があります: {INDEX_NAME}")
if version == TO_VERSION:
if (
index_row is None
or index_row[0] != "review_items"
or index_row[1] is None
or normalize_sql(str(index_row[1])) != normalize_sql(INDEX_SQL)
):
raise ContractError("v2の優先度インデックス定義が一致しません")
def check_database(conn: sqlite3.Connection) -> None:
integrity = conn.execute("PRAGMA integrity_check(1)").fetchone()
if integrity is None or str(integrity[0]).lower() != "ok":
raise ContractError("SQLiteの整合性検査に合格しませんでした")
if conn.execute("PRAGMA foreign_key_check").fetchone() is not None:
raise ContractError("外部キー違反があります")
def row_count(conn: sqlite3.Connection) -> int:
return int(conn.execute("SELECT COUNT(*) FROM review_items").fetchone()[0])
def copy_database(source: Path, target: Path) -> None:
with closing(
sqlite3.connect(readonly_uri(source), uri=True, timeout=5.0)
) as src:
src.execute("PRAGMA query_only = ON")
with closing(sqlite3.connect(target, timeout=5.0)) as dst:
src.backup(dst)
def migrate_copy(target: Path) -> int:
with closing(
sqlite3.connect(target, timeout=5.0, isolation_level=None)
) as conn:
conn.execute("PRAGMA foreign_keys = ON")
journal = conn.execute("PRAGMA journal_mode = DELETE").fetchone()
if journal is None or str(journal[0]).lower() != "delete":
raise ContractError("出力をDELETEジャーナルモードへ変更できません")
validate_contract(conn, FROM_VERSION)
check_database(conn)
before = row_count(conn)
conn.execute("BEGIN IMMEDIATE")
try:
conn.execute(
"""
ALTER TABLE review_items
ADD COLUMN priority INTEGER NOT NULL DEFAULT 0
CHECK (priority BETWEEN 0 AND 3)
"""
)
conn.execute(INDEX_SQL)
conn.execute("PRAGMA user_version = 2")
if row_count(conn) != before:
raise ContractError("移行中にreview_itemsの行数が変わりました")
invalid_priority = int(
conn.execute(
"""
SELECT COUNT(*)
FROM review_items
WHERE priority IS NULL OR priority <> 0
"""
).fetchone()[0]
)
if invalid_priority != 0:
raise ContractError("既存行のpriorityが0で初期化されていません")
validate_contract(conn, TO_VERSION)
check_database(conn)
conn.commit()
except (Exception, KeyboardInterrupt):
conn.rollback()
raise
return before
def verify_output(target: Path, expected_rows: int) -> None:
with closing(sqlite3.connect(readonly_uri(target), uri=True)) as conn:
conn.execute("PRAGMA foreign_keys = ON")
conn.execute("PRAGMA query_only = ON")
validate_contract(conn, TO_VERSION)
check_database(conn)
if row_count(conn) != expected_rows:
raise ContractError("再オープン後の行数が一致しません")
invalid_priority = int(
conn.execute(
"""
SELECT COUNT(*)
FROM review_items
WHERE priority IS NULL OR priority <> 0
"""
).fetchone()[0]
)
if invalid_priority != 0:
raise ContractError("再オープン後のpriority検査に失敗しました")
def resolve_paths(source_text: str, target_text: str) -> tuple[Path, Path]:
try:
source = Path(source_text).expanduser().resolve(strict=True)
except (OSError, RuntimeError) as exc:
raise ContractError(f"入力を解決できません: {source_text}") from exc
if not source.is_file():
raise ContractError(f"入力は通常ファイルではありません: {source_text}")
target_input = Path(target_text).expanduser()
if os.path.lexists(target_input):
raise ContractError(f"出力先は既に存在します: {target_text}")
try:
target = target_input.resolve(strict=False)
except (OSError, RuntimeError) as exc:
raise ContractError(f"出力先を解決できません: {target_text}") from exc
if source == target:
raise ContractError("入力と出力に同じパスは指定できません")
if not target.parent.is_dir():
raise ContractError(f"出力先ディレクトリがありません: {target.parent}")
return source, target
def migrate(source_text: str, target_text: str) -> dict[str, Any]:
if sqlite3.sqlite_version_info < MIN_SQLITE:
raise ContractError(
f"SQLite 3.37.0以上が必要です: {sqlite3.sqlite_version}"
)
source, target = resolve_paths(source_text, target_text)
created = False
try:
try:
fd = os.open(target, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
except FileExistsError as exc:
raise ContractError(f"出力先は既に存在します: {target_text}") from exc
created = True
os.close(fd)
copy_database(source, target)
rows = migrate_copy(target)
verify_output(target, rows)
except (Exception, KeyboardInterrupt):
if created:
target.unlink(missing_ok=True)
raise
return {
"status": "migrated",
"from_version": FROM_VERSION,
"to_version": TO_VERSION,
"rows": rows,
"output": target_text,
}
def main() -> int:
parser = JsonParser(description="SQLiteのv1コピーをv2へ移行します")
parser.add_argument("source", help="読み取り元のv1 SQLite")
parser.add_argument("output", help="存在しない出力先")
args = parser.parse_args()
try:
report = migrate(args.source, args.output)
except ContractError as exc:
emit(sys.stderr, {"status": "rejected", "message": str(exc)})
return 2
except KeyboardInterrupt:
emit(sys.stderr, {"status": "interrupted"})
return 130
except Exception as exc:
emit(
sys.stderr,
{
"status": "error",
"error_type": type(exc).__name__,
"message": str(exc),
},
)
return 3
emit(sys.stdout, report)
return 0
if __name__ == "__main__":
raise SystemExit(main())3．マイグレーションを実行する
入力とは異なる、まだ存在しない出力先を指定します。
python3 migrate_review_db.py review_v1.sqlite review_v2.sqlite成功時は標準出力へ次の1行を返し、終了コードは0です。
{"status":"migrated","from_version":1,"to_version":2,"rows":2,"output":"review_v2.sqlite"}出力を読み取り専用で開き、バージョンと既存行の優先度を確認します。
python3 -c "import sqlite3; c=sqlite3.connect('file:review_v2.sqlite?mode=ro', uri=True); print(c.execute('PRAGMA user_version').fetchone()[0]); print(c.execute('SELECT record_id, status, priority FROM review_items ORDER BY item_id').fetchall()); c.close()"2
[('R-001', 'pending', 0), ('R-002', 'accepted', 0)]元のreview_v1.sqliteはバージョン1のままです。既存行へ高い優先度を自動設定しないため、業務担当者が確認した後の更新処理は別に用意します。
4．誤ったバージョンを拒否する
作成したv2を、もう一度v1入力として渡します。スクリプトはバージョン不一致を検出し、終了コード2を返します。作成した出力ファイルも削除します。
python3 migrate_review_db.py review_v2.sqlite rejected.sqlite
echo $?
test ! -e rejected.sqlite && echo absent{"status":"rejected","message":"user_versionが1ではありません: 2"}
2
absentすでに存在する出力先を指定した場合も終了コード2で停止します。事前確認に加えて、os.open()による排他的作成を使うため、確認直後に同名ファイルが作られても上書きしません。
終了コード
| 終了コード | 意味 | 後続処理 |
|---|---|---|
| 0 | コピー、移行、再オープン後の検査が完了 | アプリケーション側の読み書き試験へ進む |
| 2 | バージョン、列定義、出力先などが契約外 | 入力と移行手順を確認する |
| 3 | SQLite、ファイル操作、または予期しない実行エラー | エラー種別と実行環境を確認する |
| 130 | Ctrl+Cによる中断 | 出力が残っていないことを確認してから再実行する |
--helpは通常のヘルプ文を表示します。それ以外にスクリプトが処理した結果は、標準出力または標準エラー出力へ1行のJSONとして返します。
この実装で保証しないこと
- 移行の成功は、アプリケーションがv2へ対応している証明ではない。実際に使用するSELECT、INSERT、UPDATEを隔離環境で試す。
- 部分インデックスを作成しても、既存の取得処理が自動的に優先順へ変わるわけではない。取得SQLにも
WHERE status = 'pending'とORDER BY priority DESC, created_at, item_idが必要になる。 created_atはTEXTとして保持するだけで、日時書式や時系列上の正しさまでは検査しない。- 入力は更新されていない隔離コピーとし、出力先も別プロセスが操作しない、権限を制御したローカルディレクトリを使う。
- 強制終了、OS停止、電源断ではPythonの後始末が動かず、不完全な出力が残る場合がある。終了コード0と成功JSONを確認できないファイルは採用しない。
- v2へ新しい業務データを書き込んだ後にv1ファイルへ戻しても、追加データは引き継がれない。切り戻し条件とデータ変換は本番切り替え前に決める。
実データへ適用する前の確認
- 実際の
review_itemsに合わせて、v1とv2の列定義を固定する。 - 追加列の既定値を推測で決めず、既存行へ安全に適用できる値を選ぶ。
- 検証済みバックアップから作った、更新されない隔離データベースを入力にする。
- 終了コード0、成功JSON、行数、整合性検査の結果を確認する。
- v2対応のアプリケーションで、読み取りと書き込みのスモークテストを行う。
- 入力となったv1コピーは、受け入れ確認が終わるまで変更せずに保管する。
コピー作成、契約検査、トランザクション、再オープン後の検査を分けると、DDLを実行できたことと、業務で採用できることを別々に判断できます。