SQLite schema migrationを安全に設計する｜user_version・transaction・段階移行

schema変更を手作業にすると再現できない

SQLiteは単一fileで配布しやすいため、デスクトップツール、CAD補助アプリ、計測端末、ローカルWebツールの保存先として使われます。しかし、運用開始後にcolumn、constraint、index、参照関係を変更すると、利用者ごとに異なる世代のdatabaseが残ります。

起動時に不足columnを見つけてその場で追加するだけでは、どの変更が適用済みか、途中で失敗したか、同じ処理を再実行してよいかが分かりません。database本体、アプリ、API、取込fileのschemaがずれると、SQLエラーだけでなく、NULLや単位の取り違えといった静かな不整合も起こります。

安全なmigrationには、次の契約が必要です。

• 現在のschema versionをdatabase内で識別する
• versionごとの変更を順番どおり一度だけ適用する
• schema変更、data変換、version更新を同じtransactionで確定する
• 変更後にforeign key、integrity、必須fieldを検証する
• 旧アプリと新アプリが混在する期間を設計する
• 失敗時は途中状態を残さず、必要ならbackupから戻せるようにする

user_versionは何を保存する値か

SQLiteのPRAGMA user_versionは、database header内にあるapplication用の整数です。SQLite自身は値の意味を解釈しません。そのため、アプリ側で「1は初期schema、2はunit追加、3はunit必須化」のように規則を決めます。

PRAGMA user_version;
PRAGMA user_version = 3;

user_versionとschema_versionは別物です。schema_versionはSQLiteがschema変更の検知に使う内部値であり、applicationのmigration番号として書き換えてはいけません。

整数一つだけでは、適用日時、migration名、SQL内容の差し替えを確認できません。実務ではuser_versionを現在位置の高速な指標とし、別tableへ履歴とchecksumを残す構成が扱いやすくなります。

CREATE TABLE IF NOT EXISTS schema_migrations (
    version INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    checksum TEXT NOT NULL,
    applied_at TEXT NOT NULL
) STRICT;

migrationの入力と出力を定義する

migrationも通常のデータ変換と同様に、入力と出力を固定します。

未知のversion、想定より新しいversion、履歴とuser_versionが矛盾するdatabaseは、自動修復せず停止させます。「たぶん古い」と推測してSQLを当てると、別系統のdatabaseを破壊しかねません。

小さな変更とtable再構築を使い分ける

SQLiteが直接扱えるALTER TABLEには、table名変更、column名変更、column追加、column削除などがあります。ただし、既存columnの型や複雑なconstraint変更では、新しいtableを作ってdataを移す方法が基本になります。

追加だけで済むmigration

version 1から2では、既存アプリを壊さないようにNULL許容のunitを追加し、既存recordをbackfillします。

ALTER TABLE parts ADD COLUMN unit TEXT;
UPDATE parts SET unit = 'pcs' WHERE unit IS NULL;
CREATE INDEX idx_parts_unit ON parts(unit);

column追加と同時に厳しいconstraintを課すより、「追加→data補完→検証→必須化」と段階を分けた方が、既存dataの例外を確認できます。

constraint変更は新tableへ移す

version 2から3では、unitを必須にして許可値を制限します。

CREATE TABLE parts_new (
    part_id TEXT PRIMARY KEY,
    name TEXT NOT NULL,
    quantity REAL NOT NULL CHECK (quantity >= 0),
    unit TEXT NOT NULL CHECK (unit IN ('pcs', 'kg', 'm')),
    updated_at TEXT NOT NULL
) STRICT;

INSERT INTO parts_new (
    part_id, name, quantity, unit, updated_at
)
SELECT
    part_id, name, quantity, unit, updated_at
FROM parts;

DROP INDEX idx_parts_unit;
DROP TABLE parts;
ALTER TABLE parts_new RENAME TO parts;
CREATE INDEX idx_parts_unit ON parts(unit);

この順序では、古いtableを先にrenameしません。SQLite公式が示す再構築手順と同様に、新しい名前でtableを作り、dataを移し、古いtableを削除してから最終名へrenameします。index、trigger、viewがある場合は定義を事前に確認し、再作成します。

transaction境界を決める

schema変更、data変換、履歴追加、user_version更新は同じtransactionへ入れます。途中で例外が起きた場合、すべてrollbackされるので、再実行時には同じ旧versionから開始できます。

flowchart TD
  A[現在versionを確認] --> B[BEGIN IMMEDIATE]
  B --> C[schemaとdataを変換]
  C --> D[制約と件数を検証]
  D --> E[履歴とuser_versionを更新]
  E --> F[COMMIT]

PRAGMA foreign_keysの切替はtransaction中には行いません。table再構築で一時的に無効化する場合は、transaction開始前にOFF、変更後にPRAGMA foreign_key_check、COMMIT後にONという境界を守ります。

Pythonで順次migrationする

次の例では、version 1から3までを一段ずつ適用します。migration SQLは外部入力ではなく、アプリに同梱した固定codeとして扱います。

from __future__ import annotations

import hashlib
import sqlite3
from datetime import datetime, timezone
from pathlib import Path


TARGET_VERSION = 3
Migration = tuple[str, ...]


MIGRATIONS: dict[int, tuple[str, Migration]] = {
    2: (
        "add-and-backfill-unit",
        (
            "ALTER TABLE parts ADD COLUMN unit TEXT",
            "UPDATE parts SET unit = 'pcs' WHERE unit IS NULL",
            "CREATE INDEX idx_parts_unit ON parts(unit)",
        ),
    ),
    3: (
        "require-valid-unit",
        (
            """
            CREATE TABLE parts_new (
            part_id TEXT PRIMARY KEY,
            name TEXT NOT NULL,
            quantity REAL NOT NULL CHECK (quantity >= 0),
            unit TEXT NOT NULL CHECK (unit IN ('pcs', 'kg', 'm')),
            updated_at TEXT NOT NULL
            ) STRICT
            """,
            """
            INSERT INTO parts_new (part_id, name, quantity, unit, updated_at)
            SELECT part_id, name, quantity, unit, updated_at FROM parts
            """,
            "DROP INDEX idx_parts_unit",
            "DROP TABLE parts",
            "ALTER TABLE parts_new RENAME TO parts",
            "CREATE INDEX idx_parts_unit ON parts(unit)",
        ),
    ),
}


def migration_checksum(statements: Migration) -> str:
    source = "\n;\n".join(statement.strip() for statement in statements)
    return hashlib.sha256(source.encode("utf-8")).hexdigest()


def migrate_database(path: Path) -> int:
    conn = sqlite3.connect(path, isolation_level=None, timeout=5.0)
    try:
        conn.execute("PRAGMA foreign_keys = ON")
        current = conn.execute("PRAGMA user_version").fetchone()[0]

        if current < 1 or current > TARGET_VERSION:
            raise RuntimeError(f"unsupported schema version: {current}")

        while current < TARGET_VERSION:
            next_version = current + 1
            name, statements = MIGRATIONS[next_version]

            conn.execute("PRAGMA foreign_keys = OFF")
            conn.execute("BEGIN IMMEDIATE")
            try:
                conn.execute(
                    """
                    CREATE TABLE IF NOT EXISTS schema_migrations (
                        version INTEGER PRIMARY KEY,
                        name TEXT NOT NULL,
                        checksum TEXT NOT NULL,
                        applied_at TEXT NOT NULL
                    ) STRICT
                    """
                )
                for statement in statements:
                    conn.execute(statement)

                violations = conn.execute(
                    "PRAGMA foreign_key_check"
                ).fetchall()
                if violations:
                    raise RuntimeError(f"foreign key violations: {violations}")

                applied_at = datetime.now(timezone.utc).isoformat()
                conn.execute(
                    """
                    INSERT INTO schema_migrations (
                        version, name, checksum, applied_at
                    ) VALUES (?, ?, ?, ?)
                    """,
                    (
                        next_version,
                        name,
                        migration_checksum(statements),
                        applied_at,
                    ),
                )
                conn.execute(f"PRAGMA user_version = {next_version:d}")
                conn.execute("COMMIT")
                current = next_version
            except Exception:
                if conn.in_transaction:
                    conn.execute("ROLLBACK")
                raise
            finally:
                conn.execute("PRAGMA foreign_keys = ON")

        result = conn.execute("PRAGMA integrity_check").fetchone()[0]
        if result != "ok":
            raise RuntimeError(f"integrity check failed: {result}")
        return current
    finally:
        conn.close()

PRAGMA user_version = ...へ外部入力を連結してはいけません。この例では、code内の整数であるnext_versionだけを使用しています。通常のvalueはplaceholderでbindし、table名やcolumn名もallowlistまたは固定migrationから選びます。

Pythonのexecutescript()は複数statementに便利ですが、transaction制御を自動で補ってくれるわけではありません。接続状態によっては実行前にpending transactionをCOMMITするため、migrationの原子性をexecutescript()任せにしないことが重要です。

冪等性は「同じSQLを何度でも実行」とは限らない

ALTER TABLE ... ADD COLUMNは同じdatabaseへ二度適用すると失敗します。それでもmigration全体は再実行可能です。理由は、成功した変更とuser_versionが同じtransactionで確定し、失敗時には両方がrollbackされるからです。

「columnがなければ追加」のような存在確認だけに依存すると、名前は同じでも型やconstraintが違う状態を見逃します。version、履歴、checksum、期待するtable定義を組み合わせて検証します。公開後のmigration fileを書き換えず、修正は新しいversionとして追加するのが原則です。

旧アプリとの互換期間を設計する

database fileを複数versionのアプリが開く場合、schemaだけ先に破壊的変更すると旧アプリが停止します。段階移行では次の順序を使えます。

• expand：新columnや新tableを追加し、旧構造を残す
• migrate：既存dataをbackfillし、新旧両方へ対応する
• switch：新アプリの読み書きを新構造へ切り替える
• contract：旧アプリがなくなった後で旧columnや互換viewを削除する

APIやCSV取込も同じです。databaseだけversion 3にしても、受信JSONやCSVが旧field名を送り続ければ連携は壊れます。入口schema、変換code、database migration、出力schemaを一つのrelease単位で管理します。

正常系と異常系をtestする

本番databaseのcopyに対し、少なくとも次を自動確認します。

• version 1から3へ順番にmigrationできる
• version 2から再開して3へ到達できる
• 途中で例外を発生させるとschema、data、versionがrollbackされる
• partの件数、PRIMARY KEY、数量、日時が維持される
• NULLや未知のunitがあれば必須化前に停止する
• index、trigger、view、foreign keyが期待どおり再作成される
• foreign_key_checkが空、integrity_checkがokになる
• migration済みdatabaseへ再実行しても変更が増えない
• 未知versionや新しすぎるversionを拒否する

backupは作成しただけで終わらず、migration前のcopyから復元できることもtestします。downgrade SQLを無理に用意するより、旧versionのアプリと整合するbackupへ戻す方が安全な場合があります。

セキュリティと運用上の注意

database path、migration file、backupの書込先を外部入力から自由に選ばせないようにします。SQL識別子の動的生成を避け、値はplaceholderでbindします。migration logへcredential、個人情報、record全文を残さず、version、checksum、開始・終了時刻、結果、error codeだけを記録します。

複数processが同時に起動する環境では、BEGIN IMMEDIATEで一つのwriterだけがmigrationへ入るようにします。migration中は通常の更新受付を停止するか、maintenance状態を明示します。大規模tableの再構築は時間と追加disk領域を使うため、本番data量で所要時間と空き容量を測ります。

CAD・BOM・API連携へ再利用する

CAD属性やBOMでは、単位column追加、部品ID形式変更、revision導入、親子関係の正規化などが発生します。各変更をmigrationとして残せば、古い設計資産を現在schemaへ段階的に変換できます。

変換時には元のID、source version、変換規則versionを保持します。表示名をIDへ置き換える場合も、対応表を保存し、同じ入力から同じ結果を再生成できるようにします。database migrationを単なる保守作業ではなく、設計data資産を将来のAPI、検索、分析、自動化へ接続する変換契約として扱うことが重要です。

まとめ

SQLiteのschema migrationは、SQLを一度実行する作業ではありません。現在位置、変換順序、transaction境界、検証、互換期間、復元方法をまとめたdata契約です。

PRAGMA user_versionで現在versionを示し、履歴tableへ名前とchecksumを残します。schema変更、data変換、履歴追加、version更新を同じtransactionへ入れ、変更後にforeign keyとintegrityを確認します。破壊的変更は新tableへのcopyで行い、旧アプリとの混在期間にはexpand・migrate・switch・contractの段階移行を使います。

この仕組みをcodeとtestとして管理すれば、SQLite fileがどの世代でも、同じ手順で現在schemaへ到達できます。CAD、BOM、API、ローカルツールのdataを長期的に再利用する基盤にもなります。

参考情報

• SQLite PRAGMA user_version
• SQLite ALTER TABLE
• SQLite Transaction
• Python sqlite3 — DB-API 2.0 interface