第168回 SQLite移行後の書き込みを本番前に試す:Pythonで作る破棄可能なスモークテスト

第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.sqlite

testは書き込み試験へ進まず、標準エラー出力へ次の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は入力が変化しないものとして扱い、ロックや変更検出を省略します。入力が実際に変化すると、正しくない結果を返す可能性があります。実行前に書き込み元を停止し、サイドカーファイルがないことを確認する運用が前提です。

また、移行後のデータベースへ新しい業務データを書き込んだ後で古いファイルへ戻すと、その間の更新が失われます。今回のスモークテストは切り替え前の受け入れ判定であり、切り替え後の逆移行手順ではありません。

仕様を確認する一次資料

次の一歩

実DBへ適用する前に、専用の検証コピーでSQLと試験値を調整します。内部の正常操作と拒否検査がすべて期待どおりなら、testコマンド全体が終了コード0になります。

本番切り替え手順には、書き込み停止、最終バックアップ、移行、スモークテスト、切り替え可否の判断を別々の工程として記録します。

第167回 SQLiteのスキーマ変更を本番前に試す:Pythonで作るコピー移行と失敗時の破棄

第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ファイルへ戻しても、追加データは引き継がれない。切り戻し条件とデータ変換は本番切り替え前に決める。

実データへ適用する前の確認

  1. 実際のreview_itemsに合わせて、v1とv2の列定義を固定する。
  2. 追加列の既定値を推測で決めず、既存行へ安全に適用できる値を選ぶ。
  3. 検証済みバックアップから作った、更新されない隔離データベースを入力にする。
  4. 終了コード0、成功JSON、行数、整合性検査の結果を確認する。
  5. v2対応のアプリケーションで、読み取りと書き込みのスモークテストを行う。
  6. 入力となったv1コピーは、受け入れ確認が終わるまで変更せずに保管する。

コピー作成、契約検査、トランザクション、再オープン後の検査を分けると、DDLを実行できたことと、業務で採用できることを別々に判断できます。

第166回 SQLiteバックアップを隔離環境へ戻す:Pythonで作る復元訓練と業務スキーマ確認

第165回では、稼働中のSQLiteから整合したバックアップを作り、SHA-256とSQLiteの検査結果をマニフェストへ残しました。ただし、バックアップを作成できたことと、必要な状態で復元できることは同じではありません。

今回は、本番データベースを上書きせず、新しい隔離用ディレクトリで復元訓練を行います。コピー後のファイルを再読して計算したSHA-256、SQLiteの整合性、外部キー、PRAGMA user_version、必須テーブル、列、行数範囲を確認します。

実装にはPython 3.10以上の標準ライブラリだけを使います。本番データベースとの切り替えや、アプリケーションからの書き込み試験は対象外です。

今回の復元訓練で確認すること

  1. 別に保管したSHA-256と、コピー後のファイルを再読して計算した値が一致する
  2. 復元先を新しく作成し、既存のファイルやディレクトリを上書きしない
  3. integrity_check(1)がokを返す
  4. foreign_key_checkで違反が見つからない
  5. PRAGMA user_versionが復元ポリシーと一致する
  6. 必要なテーブルと列が存在し、行数が許容範囲内にある
  7. すべての検査を通過した後でreport.jsonを確定する
項目 今回の契約
入力 SQLiteのバックアップファイル、期待するSHA-256、復元ポリシーJSON
復元先 実行前に存在しないディレクトリ
生成物 recovered.sqliteとreport.json
完了条件 report.jsonをJSONとして読み取れ、statusがpassedであること
上書き 既存の復元先は拒否する
対象外 本番への切り替え、業務データの意味上の正しさ、添付データベース、外部ファイル、アプリケーション固有の検索結果

入力には、SQLiteのバックアップAPIなどで作成した、単独で開けるバックアップファイルを使います。稼働中のメインデータベースだけをファイルコピーすると、WALファイルに残っている変更を含められないことがあります。

復元ポリシーを用意する

次の内容をrecovery-policy.jsonとして保存します。この例は、後ほど作成する動作確認用データベースに対応しています。

{
  "version": 1,
  "sqlite_user_version": 1,
  "tables": {
    "review_items": {
      "required_columns": [
        "batch_id",
        "record_id",
        "status"
      ],
      "min_rows": 2,
      "max_rows": 2
    }
  }
}

min_rowsとmax_rowsは、動作確認用データが2行であるため同じ値にしています。実際のバックアップでは、復旧時点として許容できる範囲を設定します。上限を設けない場合はmax_rowsをnullにします。

このポリシーは必須列を検査しますが、追加の列やテーブルは拒否しません。検査対象には通常列だけでなく、生成列と仮想テーブルの隠し列も含めます。そのため、列一覧の取得には、それらを省略するPRAGMA table_infoではなくPRAGMA table_xinfoを使います。列名は大文字と小文字を含めてポリシーと一致させます。PRAGMA user_versionはSQLiteが自動更新する値ではありません。スキーマ変更時に、データベースとポリシーを同じ値へ更新します。

復元訓練CLIを実装する

次のコードをrestore_drill.pyとして保存します。復元先ディレクトリを作成し、コピー後の候補ファイルを再読してSHA-256を計算してから、読み取り専用のSQLite接続で検査します。

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import sqlite3
import stat
import sys
from pathlib import Path
from typing import Any

CHUNK_SIZE = 1024 * 1024


class DrillError(Exception):
    def __init__(self, exit_code: int, kind: str, message: str) -> None:
        super().__init__(message)
        self.exit_code = exit_code
        self.kind = kind
        self.message = message


def write_json_line(payload: dict[str, Any], stream: Any) -> None:
    print(
        json.dumps(payload, ensure_ascii=False, separators=(",", ":")),
        file=stream,
    )


class JsonArgumentParser(argparse.ArgumentParser):
    def error(self, message: str) -> None:
        write_json_line(
            {"status": "error", "kind": "arguments", "message": message},
            sys.stderr,
        )
        raise SystemExit(2)


def reject_duplicate_keys(
    pairs: list[tuple[str, Any]],
) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ValueError(f"duplicate JSON key: {key}")
        result[key] = value
    return result


def reject_constant(value: str) -> None:
    raise ValueError(f"non-standard JSON number: {value}")


def validate_name(value: Any, label: str) -> str:
    if type(value) is not str:
        raise DrillError(2, "policy", f"{label} must be a string")
    if (
        not value
        or value != value.strip()
        or any(ord(char) < 32 or ord(char) == 127 for char in value)
    ):
        raise DrillError(
            2,
            "policy",
            f"{label} contains an invalid character",
        )
    return value


def load_policy(path: Path) -> dict[str, Any]:
    try:
        with path.open("r", encoding="utf-8") as policy_file:
            raw = json.load(
                policy_file,
                object_pairs_hook=reject_duplicate_keys,
                parse_constant=reject_constant,
            )
    except (OSError, UnicodeError, ValueError) as exc:
        raise DrillError(
            2,
            "policy",
            f"cannot load policy: {exc}",
        ) from exc

    required_top = {"version", "sqlite_user_version", "tables"}
    if type(raw) is not dict or set(raw) != required_top:
        raise DrillError(
            2,
            "policy",
            "policy keys must be version, sqlite_user_version, and tables",
        )
    if type(raw["version"]) is not int or raw["version"] != 1:
        raise DrillError(
            2,
            "policy",
            "policy version must be integer 1",
        )

    user_version = raw["sqlite_user_version"]
    if (
        type(user_version) is not int
        or not 0 <= user_version <= 2_147_483_647
    ):
        raise DrillError(
            2,
            "policy",
            "sqlite_user_version must be an integer from 0 to 2147483647",
        )

    tables = raw["tables"]
    if type(tables) is not dict or not tables:
        raise DrillError(
            2,
            "policy",
            "tables must be a non-empty object",
        )

    required_spec = {"required_columns", "min_rows", "max_rows"}
    for raw_table_name, spec in tables.items():
        table_name = validate_name(raw_table_name, "table name")
        if table_name.lower().startswith("sqlite_"):
            raise DrillError(
                2,
                "policy",
                "internal SQLite tables are not allowed",
            )
        if type(spec) is not dict or set(spec) != required_spec:
            raise DrillError(
                2,
                "policy",
                f"{table_name} must define required_columns, min_rows, and max_rows",
            )

        columns = spec["required_columns"]
        if type(columns) is not list or not columns:
            raise DrillError(
                2,
                "policy",
                f"{table_name}.required_columns must be a non-empty array",
            )
        checked_columns = [
            validate_name(column, f"{table_name} column")
            for column in columns
        ]
        if len(checked_columns) != len(set(checked_columns)):
            raise DrillError(
                2,
                "policy",
                f"{table_name}.required_columns contains a duplicate",
            )

        min_rows = spec["min_rows"]
        max_rows = spec["max_rows"]
        if type(min_rows) is not int or min_rows < 0:
            raise DrillError(
                2,
                "policy",
                f"{table_name}.min_rows must be a non-negative integer",
            )
        if max_rows is not None and (
            type(max_rows) is not int or max_rows < min_rows
        ):
            raise DrillError(
                2,
                "policy",
                f"{table_name}.max_rows must be null or at least min_rows",
            )

    return raw


def parse_expected_sha256(value: str) -> str:
    if re.fullmatch(r"[0-9a-fA-F]{64}", value) is None:
        raise DrillError(
            2,
            "arguments",
            "expected SHA-256 must contain exactly 64 hexadecimal characters",
        )
    return value.lower()


def copy_and_hash_destination(
    source: Path,
    destination: Path,
) -> tuple[str, int]:
    try:
        with source.open("rb") as source_file:
            source_mode = os.fstat(source_file.fileno()).st_mode
            if not stat.S_ISREG(source_mode):
                raise DrillError(
                    2,
                    "input",
                    "backup must be a regular file",
                )
            with destination.open("xb") as destination_file:
                while True:
                    chunk = source_file.read(CHUNK_SIZE)
                    if not chunk:
                        break
                    written = destination_file.write(chunk)
                    if written != len(chunk):
                        raise OSError("short write while copying backup")
                destination_file.flush()
                os.fsync(destination_file.fileno())
    except DrillError:
        raise
    except OSError as exc:
        raise DrillError(
            5,
            "io",
            f"backup copy failed: {exc}",
        ) from exc

    digest = hashlib.sha256()
    total = 0
    try:
        with destination.open("rb") as copied_file:
            while True:
                chunk = copied_file.read(CHUNK_SIZE)
                if not chunk:
                    break
                digest.update(chunk)
                total += len(chunk)
    except OSError as exc:
        raise DrillError(
            5,
            "io",
            f"cannot read copied backup: {exc}",
        ) from exc

    if total == 0:
        raise DrillError(4, "sqlite_check", "backup file is empty")
    return digest.hexdigest(), total


def quote_identifier(name: str) -> str:
    return '"' + name.replace('"', '""') + '"'


def validate_database(
    path: Path,
    policy: dict[str, Any],
) -> dict[str, Any]:
    uri = f"{path.resolve().as_uri()}?mode=ro&immutable=1"
    try:
        connection = sqlite3.connect(uri, uri=True)
        try:
            connection.execute("PRAGMA foreign_keys = ON")
            connection.execute("PRAGMA query_only = ON")

            integrity_rows = connection.execute(
                "PRAGMA integrity_check(1)"
            ).fetchall()
            if integrity_rows != [("ok",)]:
                raise DrillError(
                    4,
                    "sqlite_check",
                    "integrity_check did not return ok",
                )

            foreign_key_row = connection.execute(
                "PRAGMA foreign_key_check"
            ).fetchone()
            if foreign_key_row is not None:
                raise DrillError(
                    4,
                    "sqlite_check",
                    "foreign_key_check found a violation",
                )

            actual_user_version = connection.execute(
                "PRAGMA user_version"
            ).fetchone()[0]
            if actual_user_version != policy["sqlite_user_version"]:
                raise DrillError(
                    4,
                    "sqlite_check",
                    "SQLite user_version does not match the policy",
                )

            actual_tables = {
                row[0]
                for row in connection.execute(
                    "SELECT name FROM sqlite_schema WHERE type = 'table'"
                )
            }
            table_results: dict[str, Any] = {}

            for table_name, spec in policy["tables"].items():
                if table_name not in actual_tables:
                    raise DrillError(
                        4,
                        "sqlite_check",
                        f"required table is missing: {table_name}",
                    )

                table_sql = quote_identifier(table_name)
                columns = [
                    row[1]
                    for row in connection.execute(
                        f"PRAGMA table_xinfo({table_sql})"
                    )
                ]
                missing_columns = [
                    column
                    for column in spec["required_columns"]
                    if column not in columns
                ]
                if missing_columns:
                    raise DrillError(
                        4,
                        "sqlite_check",
                        f"required column is missing from {table_name}",
                    )

                row_count = connection.execute(
                    f"SELECT COUNT(*) FROM {table_sql}"
                ).fetchone()[0]
                if row_count < spec["min_rows"]:
                    raise DrillError(
                        4,
                        "sqlite_check",
                        f"row count is below the minimum for {table_name}",
                    )
                max_rows = spec["max_rows"]
                if max_rows is not None and row_count > max_rows:
                    raise DrillError(
                        4,
                        "sqlite_check",
                        f"row count is above the maximum for {table_name}",
                    )

                table_results[table_name] = {
                    "row_count": row_count,
                    "required_columns_checked": len(
                        spec["required_columns"]
                    ),
                }
        finally:
            connection.close()
    except DrillError:
        raise
    except sqlite3.Error as exc:
        raise DrillError(
            4,
            "sqlite_check",
            f"SQLite validation failed: {exc}",
        ) from exc

    return {
        "integrity_check": "ok",
        "foreign_key_violations": 0,
        "sqlite_user_version": actual_user_version,
        "tables": table_results,
    }


def write_report_candidate(path: Path, payload: dict[str, Any]) -> None:
    try:
        with path.open("x", encoding="utf-8") as report_file:
            json.dump(payload, report_file, ensure_ascii=False, indent=2)
            print(file=report_file)
            report_file.flush()
            os.fsync(report_file.fileno())
    except OSError as exc:
        raise DrillError(
            5,
            "io",
            f"cannot write report: {exc}",
        ) from exc


def rename_new(source: Path, target: Path) -> None:
    if os.path.lexists(target):
        raise DrillError(
            5,
            "output",
            f"internal destination already exists: {target}",
        )
    try:
        source.rename(target)
    except OSError as exc:
        raise DrillError(
            5,
            "io",
            f"cannot finalize {target}: {exc}",
        ) from exc


def cleanup_created_output(output_dir: Path, paths: list[Path]) -> None:
    for path in paths:
        try:
            path.unlink()
        except FileNotFoundError:
            pass
        except OSError:
            pass
    try:
        output_dir.rmdir()
    except OSError:
        pass


def build_parser() -> argparse.ArgumentParser:
    parser = JsonArgumentParser(
        description="Restore a SQLite backup into an isolated directory."
    )
    parser.add_argument("--backup", required=True, type=Path)
    parser.add_argument("--expected-sha256", required=True)
    parser.add_argument("--policy", required=True, type=Path)
    parser.add_argument("--output-dir", required=True, type=Path)
    return parser


def run(args: argparse.Namespace) -> dict[str, Any]:
    expected_sha256 = parse_expected_sha256(args.expected_sha256)
    policy = load_policy(args.policy)

    if not args.backup.is_file():
        raise DrillError(
            2,
            "input",
            "backup does not exist or is not a regular file",
        )
    if args.output_dir.exists():
        raise DrillError(2, "output", "output path already exists")
    if not args.output_dir.parent.is_dir():
        raise DrillError(
            2,
            "output",
            "output parent does not exist or is not a directory",
        )

    try:
        args.output_dir.mkdir(mode=0o700)
    except FileExistsError as exc:
        raise DrillError(
            2,
            "output",
            "output path already exists",
        ) from exc
    except OSError as exc:
        raise DrillError(
            5,
            "io",
            f"cannot create output directory: {exc}",
        ) from exc

    candidate = args.output_dir / ".recovered.sqlite.part"
    recovered = args.output_dir / "recovered.sqlite"
    report_candidate = args.output_dir / ".report.json.part"
    report_path = args.output_dir / "report.json"
    known_paths = [report_path, report_candidate, recovered, candidate]
    success = False

    try:
        actual_sha256, size_bytes = copy_and_hash_destination(
            args.backup,
            candidate,
        )
        if actual_sha256 != expected_sha256:
            raise DrillError(
                3,
                "hash_mismatch",
                "copied file does not match the expected SHA-256",
            )

        checks = validate_database(candidate, policy)
        report = {
            "status": "passed",
            "backup_name": args.backup.name,
            "output_dir": str(args.output_dir),
            "database": str(recovered),
            "report": str(report_path),
            "sha256": actual_sha256,
            "size_bytes": size_bytes,
            "checks": checks,
        }

        rename_new(candidate, recovered)
        write_report_candidate(report_candidate, report)
        rename_new(report_candidate, report_path)
        success = True
        return report
    finally:
        if not success:
            cleanup_created_output(args.output_dir, known_paths)


def main() -> int:
    args = build_parser().parse_args()
    try:
        report = run(args)
    except DrillError as exc:
        write_json_line(
            {
                "status": "error",
                "kind": exc.kind,
                "message": exc.message,
            },
            sys.stderr,
        )
        return exc.exit_code
    except KeyboardInterrupt:
        write_json_line(
            {
                "status": "error",
                "kind": "interrupted",
                "message": "interrupted by user",
            },
            sys.stderr,
        )
        return 130
    except Exception:
        write_json_line(
            {
                "status": "error",
                "kind": "unexpected",
                "message": "unexpected execution error",
            },
            sys.stderr,
        )
        return 5

    write_json_line(report, sys.stdout)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

immutable=1は、コピー後の候補ファイルを検査中に変更しないことを前提に使います。SQLiteはこの指定があるファイルのロックと変更検出を省略します。書き込みが続いている本番データベースには使えません。

report.jsonは最後に候補名から確定名へ変更します。途中で停止してrecovered.sqliteだけが残っても、完了レポートがなければ成功として扱いません。

動作確認用のSQLiteを作る

次のコードをmake_demo_backup.pyとして保存します。動作確認専用のSQLiteファイルを新規作成し、サンプルデータを2行登録します。

from __future__ import annotations

import hashlib
import json
import sqlite3
from pathlib import Path

DB_PATH = Path("queue-backup.sqlite")


def file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as source:
        while True:
            chunk = source.read(1024 * 1024)
            if not chunk:
                break
            digest.update(chunk)
    return digest.hexdigest()


def main() -> None:
    try:
        DB_PATH.touch(exist_ok=False)
    except FileExistsError as exc:
        raise SystemExit("queue-backup.sqlite already exists") from exc

    try:
        with sqlite3.connect(DB_PATH) as connection:
            connection.execute("PRAGMA foreign_keys = ON")
            connection.execute("PRAGMA user_version = 1")
            connection.execute(
                """
                CREATE TABLE review_items (
                    batch_id TEXT NOT NULL,
                    record_id TEXT NOT NULL,
                    status TEXT NOT NULL CHECK (
                        status IN (
                            'pending',
                            'accepted',
                            'corrected',
                            'rejected'
                        )
                    ),
                    PRIMARY KEY (batch_id, record_id)
                )
                """
            )
            connection.executemany(
                """
                INSERT INTO review_items (
                    batch_id,
                    record_id,
                    status
                ) VALUES (?, ?, ?)
                """,
                [
                    ("demo-batch", "R-001", "pending"),
                    ("demo-batch", "R-002", "accepted"),
                ],
            )
    except BaseException:
        DB_PATH.unlink(missing_ok=True)
        raise

    print(
        json.dumps(
            {
                "backup": str(DB_PATH),
                "sha256": file_sha256(DB_PATH),
            },
            ensure_ascii=False,
            separators=(",", ":"),
        )
    )


if __name__ == "__main__":
    main()

順番に実行する

以下はBashでの実行例です。最初に動作確認用SQLiteを作り、そのSHA-256をJSONへ保存します。

python make_demo_backup.py > demo-result.json
python -m json.tool demo-result.json

続いてJSONからSHA-256を取り出し、復元訓練を実行します。

EXPECTED_SHA256=$(python -c 'import json,pathlib; data=json.loads(pathlib.Path("demo-result.json").read_text(encoding="utf-8")); print(data["sha256"])')
python restore_drill.py --backup queue-backup.sqlite --expected-sha256 "$EXPECTED_SHA256" --policy recovery-policy.json --output-dir drill-20261007

成功時の終了コードは0です。標準出力には、実際のハッシュ、サイズ、検査結果を含む1行のJSONが出ます。復元先には次の2ファイルが残ります。

drill-20261007/
├── recovered.sqlite
└── report.json

レポートを読み直します。

python -m json.tool drill-20261007/report.json

statusがpassedであることに加え、sqlite_user_version、テーブルごとのrow_count、検査した必須列数を確認します。

失敗条件と終了コードを確認する

終了コード 条件 後続処理
0 すべての検査に合格 隔離環境での読み取り試験へ進める
2 引数、ポリシー、入力パス、復元先の問題 設定を修正する
3 コピー後のファイルのSHA-256が期待値と不一致 バックアップとハッシュの取得元を確認する
4 SQLite、外部キー、スキーマ、行数の検査に不合格 復元物を使用しない
5 ファイル操作または予期しない実行エラー 空き容量、権限、パス、実行環境を確認する
130 Ctrl+Cによる中断 残存ディレクトリを確認してから再実行する

動作確認では、次の条件を個別に試せます。試すたびに新しい復元先を指定します。

  • 表示されたSHA-256を1文字変更すると終了コード3になる
  • ポリシーのmin_rowsとmax_rowsをともに3へ変更すると終了コード4になる
  • 成功済みのdrill-20261007を再指定すると終了コード2になる
  • 存在しない列をrequired_columnsへ追加すると終了コード4になる

スクリプトが捕捉した失敗では、自身が作成した既知のファイルと、空になった復元先ディレクトリの削除を試みます。強制終了、停電、ファイルシステム障害、削除権限の問題がある場合は途中生成物が残り得ます。既存の復元先は再利用せず、内容を確認してから別の名前でやり直します。

実データへ適用するときの注意点

  • 期待するSHA-256は、バックアップ本体とは別に保護したマニフェストから取得する
  • バックアップと期待値を同時に書き換えられる環境では、SHA-256だけを改ざん対策にしない
  • 行数範囲は現在の本番件数ではなく、対象バックアップの復旧時点に対して決める
  • COUNT(*)を使う行数検査は、大きなテーブルでの実行時間を事前に測る
  • 復元先にはバックアップと同程度以上の空き容量を確保する
  • mode=0o700だけに依存せず、OSとファイルシステムの権限やACLも確認する
  • 同じOS利用者による競合操作や、突然の電源断に対するファイルシステム上の完全性までは保証しない

integrity_checkはSQLiteの構造を検査しますが、外部キー違反は検査しません。そのため、foreign_key_checkを別に実行しています。どちらも、レビュー内容の正しさや、必要な時点までデータが含まれていることまでは保証しません。

次の実務作業は、合格したrecovered.sqliteを隔離したアプリケーションへ読み取り専用で接続し、担当者が使う一覧表示や件数集計を確認することです。本番ファイルとの切り替えは、サービス停止、権限、切り戻し条件を含む別の手順として管理します。

一次資料

第165回 AIレビューキューを一貫した状態で退避する:PythonとSQLiteで作る期限・検証付きバックアップ

第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=(',', ':'),
    )
)

実行順を確認する

  1. Pythonのバージョンを確認します。
python --version
  1. デモ用データベースを作ります。
python make_demo_db.py
  1. 空の出力ディレクトリを作り、バックアップします。次の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のバージョンやファイル内容によってハッシュは変わるため、記事内では固定値を示しません。

  1. 作成した成果物を別の読み取りとして再検証します。
python sqlite_snapshot.py verify backup-out/review-demo.sqlite3

verifyは、隣接するreview-demo.sqlite3.manifest.jsonを自動的に読みます。バックアップとマニフェストの名前を個別に変更すると対応関係を確認できなくなるため、2ファイルを同じ名前の組として扱います。

  1. 同じ出力先でもう一度バックアップし、上書きされないことを確認します。
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でバックアップと検証を通し、その後に完了期限、権限、保存先、復旧時の業務確認を含めた手順へ落とし込んでください。

第164回 AIレビューを期限付きで割り当てる:PythonとSQLiteで作る担当ロック

第163回では、AIの要確認結果をSQLiteへ取り込み、担当者が採用・修正・却下を記録するレビューキューを作りました。担当者が複数になると、同じ項目を重複して処理しない仕組みが必要です。

今回は、レビュー項目を一定時間だけ担当者へ割り当てる期限付きロックを実装します。有効期限内は別の担当者による取得を拒否し、失効後は新しい担当者が再取得できます。古いトークンからの確定操作も拒否します。

Python標準ライブラリだけを使い、AIサービスへの通信や元データの更新は行いません。第163回のデータベースへ直接適用する移行スクリプトではないため、動作確認には新しいデータベースを使ってください。

今回実装する契約

項目 仕様
取得順 未確定かつ未取得、またはロックが失効した項目を、登録時刻、バッチID、レコードIDの順で1件取得する
ロック時間 30秒以上3600秒以下の整数。現在時刻を整数秒へ切り捨てるため、実際の有効時間は指定値より最大1秒未満短くなり得る
確定条件 有効期限内のロックトークンを持つこと
確定結果 accepted、corrected、rejectedのいずれか
時刻 time.time()から得たUnix秒を整数で保存する
出力 ヘルプを除き、標準出力または標準エラー出力へ1行のJSONを返す
対象環境 同じ端末上の少数プロセスが、ローカルディスク上のSQLiteを共有する構成

reviewerは担当者名の記録であり、本人確認ではありません。トークンも認証やアクセス制御の代わりにはなりません。OSのファイル権限や利用者認証は別に用意します。

期限付きロックの状態変化

現在の状態 操作 結果
未取得 claim 担当者、トークン、有効期限を記録する
有効な取得中 renew 現在時刻を基準に有効期限を設定し直す
有効な取得中 release 確定せずロックを解放する
有効な取得中 decide レビュー結果を確定し、ロック情報を消去する
失効済み 別の担当者がclaim 新しいトークンで再取得する
再取得後 古いトークンでdecide 確定を拒否する

レビュー用CLIを作る

次のコードをreview_leases.pyとして保存します。Python 3.10以上を前提とします。

from __future__ import annotations

import argparse
import json
import secrets
import sqlite3
import sys
import time
from contextlib import closing
from pathlib import Path
from typing import Any


SCHEMA_VERSION = 1
MIN_LEASE_SECONDS = 30
MAX_LEASE_SECONDS = 3600
MAX_CLAIM_FILE_BYTES = 16_384


class AppError(Exception):
    def __init__(self, code: str, message: str, exit_code: int = 2):
        super().__init__(message)
        self.code = code
        self.message = message
        self.exit_code = exit_code


class JsonArgumentParser(argparse.ArgumentParser):
    def error(self, message: str):
        raise AppError('argument_error', message)


def emit(payload: dict[str, Any], *, error: bool = False) -> None:
    stream = sys.stderr if error else sys.stdout
    print(
        json.dumps(payload, ensure_ascii=False, separators=(',', ':')),
        file=stream,
    )


def lease_seconds(value: str) -> int:
    try:
        seconds = int(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError('整数を指定してください') from exc

    if not MIN_LEASE_SECONDS <= seconds <= MAX_LEASE_SECONDS:
        raise argparse.ArgumentTypeError(
            f'{MIN_LEASE_SECONDS}以上{MAX_LEASE_SECONDS}以下で指定してください'
        )
    return seconds


def validate_text(value: str, field: str, max_length: int) -> str:
    if not value or value != value.strip():
        raise AppError('invalid_text', f'{field}は前後空白のない文字列にしてください')
    if len(value) > max_length:
        raise AppError('invalid_text', f'{field}は{max_length}文字以内にしてください')
    if any(ord(char) < 32 or ord(char) == 127 for char in value):
        raise AppError('invalid_text', f'{field}にASCII制御文字は使用できません')
    return value


def validate_token(value: str) -> str:
    valid_chars = all(
        char.isascii() and (char.isalnum() or char in '-_')
        for char in value
    )
    if not 20 <= len(value) <= 200 or not valid_chars:
        raise AppError('invalid_token', 'ロックトークンの形式が正しくありません')
    return value


def configure(connection: sqlite3.Connection) -> None:
    connection.row_factory = sqlite3.Row
    connection.execute('PRAGMA foreign_keys = ON')
    connection.execute('PRAGMA busy_timeout = 5000')


def connect_new(path: Path) -> sqlite3.Connection:
    if path.exists():
        raise AppError('database_exists', '初期化先がすでに存在します')
    connection = sqlite3.connect(
        str(path), timeout=5.0, isolation_level=None
    )
    configure(connection)
    return connection


def connect_existing(path: Path) -> sqlite3.Connection:
    if not path.is_file():
        raise AppError('database_not_found', 'データベースが見つかりません')
    uri = path.resolve().as_uri() + '?mode=rw'
    connection = sqlite3.connect(
        uri, uri=True, timeout=5.0, isolation_level=None
    )
    configure(connection)
    return connection


def verify_schema(connection: sqlite3.Connection) -> None:
    version = connection.execute('PRAGMA user_version').fetchone()[0]
    if version != SCHEMA_VERSION:
        raise AppError(
            'unsupported_schema',
            f'対応していないスキーマです: {version}',
        )


def load_claim_token(args: argparse.Namespace) -> str:
    if args.token is not None:
        return validate_token(args.token)

    try:
        with args.claim_json.open('rb') as source:
            data = source.read(MAX_CLAIM_FILE_BYTES + 1)
    except OSError as exc:
        raise AppError(
            'claim_file_error',
            f'取得結果を読めません: {exc}',
        ) from exc

    if len(data) > MAX_CLAIM_FILE_BYTES:
        raise AppError('claim_file_error', '取得結果のファイルが大きすぎます')

    try:
        payload = json.loads(data.decode('utf-8'))
    except (UnicodeError, json.JSONDecodeError) as exc:
        raise AppError(
            'claim_file_error',
            f'取得結果がUTF-8のJSONではありません: {exc}',
        ) from exc

    if not isinstance(payload, dict) or not isinstance(
        payload.get('claim_token'), str
    ):
        raise AppError(
            'claim_file_error',
            '取得結果にclaim_tokenがありません',
        )
    return validate_token(payload['claim_token'])


CREATE_CATEGORY_SQL = '''
CREATE TABLE allowed_category (
    category TEXT PRIMARY KEY,
    CHECK(length(trim(category)) > 0)
)
'''

CREATE_REVIEW_SQL = '''
CREATE TABLE review_item (
    batch_id TEXT NOT NULL,
    record_id TEXT NOT NULL,
    proposed_category TEXT NOT NULL
        REFERENCES allowed_category(category),
    reason TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'pending'
        CHECK(status IN ('pending', 'accepted', 'corrected', 'rejected')),
    final_category TEXT REFERENCES allowed_category(category),
    reviewed_by TEXT,
    decided_at INTEGER,
    created_at INTEGER NOT NULL,
    claim_token TEXT UNIQUE,
    claimed_by TEXT,
    claimed_at INTEGER,
    claim_expires_at INTEGER,
    PRIMARY KEY(batch_id, record_id),
    CHECK(length(batch_id) BETWEEN 1 AND 100),
    CHECK(length(record_id) BETWEEN 1 AND 100),
    CHECK(length(reason) BETWEEN 1 AND 1000),
    CHECK(
        (status = 'pending'
            AND final_category IS NULL
            AND reviewed_by IS NULL
            AND decided_at IS NULL)
        OR
        (status IN ('accepted', 'corrected')
            AND final_category IS NOT NULL
            AND reviewed_by IS NOT NULL
            AND decided_at IS NOT NULL)
        OR
        (status = 'rejected'
            AND final_category IS NULL
            AND reviewed_by IS NOT NULL
            AND decided_at IS NOT NULL)
    ),
    CHECK(status <> 'accepted' OR final_category = proposed_category),
    CHECK(status <> 'corrected' OR final_category <> proposed_category),
    CHECK(
        (claim_token IS NULL
            AND claimed_by IS NULL
            AND claimed_at IS NULL
            AND claim_expires_at IS NULL)
        OR
        (status = 'pending'
            AND claim_token IS NOT NULL
            AND claimed_by IS NOT NULL
            AND claimed_at IS NOT NULL
            AND claim_expires_at > claimed_at)
    )
)
'''

CREATE_INDEX_SQL = '''
CREATE INDEX review_item_claim_order
ON review_item(status, claim_expires_at, created_at, batch_id, record_id)
'''


def cmd_init(args: argparse.Namespace) -> dict[str, Any]:
    categories = [
        validate_text(item, 'category', 50) for item in args.category
    ]
    if len(categories) != len(set(categories)):
        raise AppError('duplicate_category', 'カテゴリが重複しています')

    with closing(connect_new(args.db)) as connection:
        with connection:
            connection.execute('BEGIN IMMEDIATE')
            connection.execute(CREATE_CATEGORY_SQL)
            connection.execute(CREATE_REVIEW_SQL)
            connection.execute(CREATE_INDEX_SQL)
            connection.executemany(
                'INSERT INTO allowed_category(category) VALUES (?)',
                [(category,) for category in categories],
            )
            connection.execute(f'PRAGMA user_version = {SCHEMA_VERSION}')

    return {
        'database': str(args.db),
        'category_count': len(categories),
    }


def cmd_add(args: argparse.Namespace) -> dict[str, Any]:
    batch_id = validate_text(args.batch_id, 'batch_id', 100)
    record_id = validate_text(args.record_id, 'record_id', 100)
    category = validate_text(args.proposed_category, 'proposed_category', 50)
    reason = validate_text(args.reason, 'reason', 1000)

    with closing(connect_existing(args.db)) as connection:
        verify_schema(connection)
        with connection:
            connection.execute('BEGIN IMMEDIATE')
            if connection.execute(
                'SELECT 1 FROM allowed_category WHERE category = ?',
                (category,),
            ).fetchone() is None:
                raise AppError('unknown_category', '許可されていないカテゴリです')

            if connection.execute(
                '''
                SELECT 1 FROM review_item
                WHERE batch_id = ? AND record_id = ?
                ''',
                (batch_id, record_id),
            ).fetchone() is not None:
                raise AppError('duplicate_item', 'レビュー項目がすでに存在します')

            connection.execute(
                '''
                INSERT INTO review_item(
                    batch_id, record_id, proposed_category, reason, created_at
                ) VALUES (?, ?, ?, ?, ?)
                ''',
                (batch_id, record_id, category, reason, int(time.time())),
            )

    return {'added': True, 'batch_id': batch_id, 'record_id': record_id}


def cmd_claim(args: argparse.Namespace) -> dict[str, Any]:
    reviewer = validate_text(args.reviewer, 'reviewer', 100)

    with closing(connect_existing(args.db)) as connection:
        verify_schema(connection)
        with connection:
            connection.execute('BEGIN IMMEDIATE')
            now = int(time.time())
            row = connection.execute(
                '''
                SELECT batch_id, record_id, proposed_category, reason
                FROM review_item
                WHERE status = 'pending'
                  AND (claim_token IS NULL OR claim_expires_at <= ?)
                ORDER BY created_at, batch_id, record_id
                LIMIT 1
                ''',
                (now,),
            ).fetchone()

            if row is None:
                result = {'claimed': False}
            else:
                token = secrets.token_urlsafe(24)
                expires_at = now + args.lease_seconds
                cursor = connection.execute(
                    '''
                    UPDATE review_item
                    SET claim_token = ?, claimed_by = ?,
                        claimed_at = ?, claim_expires_at = ?
                    WHERE batch_id = ? AND record_id = ?
                      AND status = 'pending'
                      AND (claim_token IS NULL OR claim_expires_at <= ?)
                    ''',
                    (
                        token,
                        reviewer,
                        now,
                        expires_at,
                        row['batch_id'],
                        row['record_id'],
                        now,
                    ),
                )
                if cursor.rowcount != 1:
                    raise AppError('claim_conflict', '取得状態が変わりました')

                result = {
                    'claimed': True,
                    'batch_id': row['batch_id'],
                    'record_id': row['record_id'],
                    'proposed_category': row['proposed_category'],
                    'reason': row['reason'],
                    'claimed_by': reviewer,
                    'claimed_at': now,
                    'claim_expires_at': expires_at,
                    'claim_token': token,
                }

    return result


def cmd_renew(args: argparse.Namespace) -> dict[str, Any]:
    token = load_claim_token(args)

    with closing(connect_existing(args.db)) as connection:
        verify_schema(connection)
        with connection:
            connection.execute('BEGIN IMMEDIATE')
            now = int(time.time())
            row = connection.execute(
                '''
                SELECT batch_id, record_id, claim_expires_at
                FROM review_item
                WHERE status = 'pending' AND claim_token = ?
                ''',
                (token,),
            ).fetchone()
            if row is None:
                raise AppError('claim_not_found', '有効な取得情報がありません')
            if row['claim_expires_at'] <= now:
                raise AppError('claim_expired', 'ロックはすでに失効しています')

            expires_at = now + args.lease_seconds
            connection.execute(
                'UPDATE review_item SET claim_expires_at = ? WHERE claim_token = ?',
                (expires_at, token),
            )

    return {
        'renewed': True,
        'batch_id': row['batch_id'],
        'record_id': row['record_id'],
        'claim_expires_at': expires_at,
    }


def cmd_release(args: argparse.Namespace) -> dict[str, Any]:
    token = load_claim_token(args)

    with closing(connect_existing(args.db)) as connection:
        verify_schema(connection)
        with connection:
            connection.execute('BEGIN IMMEDIATE')
            now = int(time.time())
            row = connection.execute(
                '''
                SELECT batch_id, record_id, claim_expires_at
                FROM review_item
                WHERE status = 'pending' AND claim_token = ?
                ''',
                (token,),
            ).fetchone()
            if row is None:
                raise AppError('claim_not_found', '有効な取得情報がありません')
            if row['claim_expires_at'] <= now:
                raise AppError('claim_expired', 'ロックはすでに失効しています')

            connection.execute(
                '''
                UPDATE review_item
                SET claim_token = NULL, claimed_by = NULL,
                    claimed_at = NULL, claim_expires_at = NULL
                WHERE claim_token = ?
                ''',
                (token,),
            )

    return {
        'released': True,
        'batch_id': row['batch_id'],
        'record_id': row['record_id'],
    }


def cmd_decide(args: argparse.Namespace) -> dict[str, Any]:
    token = load_claim_token(args)

    if args.decision == 'corrected':
        if args.category is None:
            raise AppError('category_required', 'correctedには--categoryが必要です')
        category = validate_text(args.category, 'category', 50)
    elif args.category is not None:
        raise AppError(
            'unexpected_category',
            'acceptedとrejectedでは--categoryを指定しません',
        )
    else:
        category = None

    with closing(connect_existing(args.db)) as connection:
        verify_schema(connection)
        with connection:
            connection.execute('BEGIN IMMEDIATE')
            now = int(time.time())
            row = connection.execute(
                '''
                SELECT batch_id, record_id, proposed_category,
                       claimed_by, claim_expires_at
                FROM review_item
                WHERE status = 'pending' AND claim_token = ?
                ''',
                (token,),
            ).fetchone()
            if row is None:
                raise AppError('claim_not_found', '有効な取得情報がありません')
            if row['claim_expires_at'] <= now:
                raise AppError('claim_expired', 'ロックはすでに失効しています')

            if args.decision == 'corrected':
                if connection.execute(
                    'SELECT 1 FROM allowed_category WHERE category = ?',
                    (category,),
                ).fetchone() is None:
                    raise AppError('unknown_category', '許可されていないカテゴリです')
                if category == row['proposed_category']:
                    raise AppError(
                        'category_unchanged',
                        'correctedでは提案と異なるカテゴリを指定してください',
                    )
                final_category = category
            elif args.decision == 'accepted':
                final_category = row['proposed_category']
            else:
                final_category = None

            cursor = connection.execute(
                '''
                UPDATE review_item
                SET status = ?, final_category = ?, reviewed_by = ?,
                    decided_at = ?, claim_token = NULL, claimed_by = NULL,
                    claimed_at = NULL, claim_expires_at = NULL
                WHERE batch_id = ? AND record_id = ?
                  AND status = 'pending' AND claim_token = ?
                ''',
                (
                    args.decision,
                    final_category,
                    row['claimed_by'],
                    now,
                    row['batch_id'],
                    row['record_id'],
                    token,
                ),
            )
            if cursor.rowcount != 1:
                raise AppError('decision_conflict', '確定状態が変わりました')

    return {
        'batch_id': row['batch_id'],
        'record_id': row['record_id'],
        'status': args.decision,
        'final_category': final_category,
        'reviewed_by': row['claimed_by'],
    }


def cmd_show(args: argparse.Namespace) -> dict[str, Any]:
    batch_id = validate_text(args.batch_id, 'batch_id', 100)
    record_id = validate_text(args.record_id, 'record_id', 100)

    with closing(connect_existing(args.db)) as connection:
        verify_schema(connection)
        row = connection.execute(
            '''
            SELECT batch_id, record_id, status, proposed_category,
                   final_category, reviewed_by, claim_token,
                   claimed_by, claim_expires_at
            FROM review_item
            WHERE batch_id = ? AND record_id = ?
            ''',
            (batch_id, record_id),
        ).fetchone()

    if row is None:
        raise AppError('item_not_found', 'レビュー項目が見つかりません')

    now = int(time.time())
    if row['status'] != 'pending':
        claim_state = 'not_applicable'
    elif row['claim_token'] is None:
        claim_state = 'unclaimed'
    elif row['claim_expires_at'] <= now:
        claim_state = 'expired'
    else:
        claim_state = 'active'

    return {
        'batch_id': row['batch_id'],
        'record_id': row['record_id'],
        'status': row['status'],
        'proposed_category': row['proposed_category'],
        'final_category': row['final_category'],
        'reviewed_by': row['reviewed_by'],
        'claimed_by': row['claimed_by'],
        'claim_state': claim_state,
    }


def add_claim_source(parser: argparse.ArgumentParser) -> None:
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument('--token')
    group.add_argument('--claim-json', type=Path)


def build_parser() -> argparse.ArgumentParser:
    parser = JsonArgumentParser(description='期限付きレビューキュー')
    subparsers = parser.add_subparsers(dest='command', required=True)

    init_parser = subparsers.add_parser('init')
    init_parser.add_argument('db', type=Path)
    init_parser.add_argument('--category', action='append', required=True)
    init_parser.set_defaults(handler=cmd_init)

    add_parser = subparsers.add_parser('add')
    add_parser.add_argument('db', type=Path)
    add_parser.add_argument('--batch-id', required=True)
    add_parser.add_argument('--record-id', required=True)
    add_parser.add_argument('--proposed-category', required=True)
    add_parser.add_argument('--reason', required=True)
    add_parser.set_defaults(handler=cmd_add)

    claim_parser = subparsers.add_parser('claim')
    claim_parser.add_argument('db', type=Path)
    claim_parser.add_argument('--reviewer', required=True)
    claim_parser.add_argument(
        '--lease-seconds', type=lease_seconds, default=300
    )
    claim_parser.set_defaults(handler=cmd_claim)

    renew_parser = subparsers.add_parser('renew')
    renew_parser.add_argument('db', type=Path)
    renew_parser.add_argument(
        '--lease-seconds', type=lease_seconds, default=300
    )
    add_claim_source(renew_parser)
    renew_parser.set_defaults(handler=cmd_renew)

    release_parser = subparsers.add_parser('release')
    release_parser.add_argument('db', type=Path)
    add_claim_source(release_parser)
    release_parser.set_defaults(handler=cmd_release)

    decide_parser = subparsers.add_parser('decide')
    decide_parser.add_argument('db', type=Path)
    decide_parser.add_argument(
        '--decision',
        choices=['accepted', 'corrected', 'rejected'],
        required=True,
    )
    decide_parser.add_argument('--category')
    add_claim_source(decide_parser)
    decide_parser.set_defaults(handler=cmd_decide)

    show_parser = subparsers.add_parser('show')
    show_parser.add_argument('db', type=Path)
    show_parser.add_argument('--batch-id', required=True)
    show_parser.add_argument('--record-id', required=True)
    show_parser.set_defaults(handler=cmd_show)

    return parser


def main() -> int:
    try:
        args = build_parser().parse_args()
        result = args.handler(args)
    except AppError as exc:
        emit(
            {'ok': False, 'error': exc.code, 'message': exc.message},
            error=True,
        )
        return exc.exit_code
    except sqlite3.Error as exc:
        emit(
            {
                'ok': False,
                'error': 'database_error',
                'message': f'SQLite処理に失敗しました: {exc}',
            },
            error=True,
        )
        return 3
    except OSError as exc:
        emit(
            {
                'ok': False,
                'error': 'io_error',
                'message': f'ファイル操作に失敗しました: {exc}',
            },
            error=True,
        )
        return 3
    except KeyboardInterrupt:
        emit(
            {'ok': False, 'error': 'interrupted', 'message': '中断しました'},
            error=True,
        )
        return 130

    emit({'ok': True, **result})
    return 0


if __name__ == '__main__':
    raise SystemExit(main())

1. データベースと確認用データを作る

既存の第163回用データベースではなく、新しいファイルを指定します。

python3 review_leases.py init review-leases.db --category 配送 --category 請求 --category その他
python3 review_leases.py add review-leases.db --batch-id batch-20261005 --record-id C-001 --proposed-category 配送 --reason '提案カテゴリを元データと照合する'

成功時の出力は次のとおりです。

{"ok":true,"database":"review-leases.db","category_count":3}
{"ok":true,"added":true,"batch_id":"batch-20261005","record_id":"C-001"}

initは既存ファイルを拒否します。カテゴリの追加や変更を行う管理コマンドは、この最小実装には含めていません。

2. レビュー項目を取得する

取得結果にはレビュー理由とロックトークンが含まれます。次の例では、未使用のファイルへ保存します。シェルのリダイレクトは既存ファイルを上書きすることがあるため、既存ファイルを出力先に指定しないでください。

python3 review_leases.py claim review-leases.db --reviewer tanaka --lease-seconds 300 > claim-tanaka.json
python3 -m json.tool claim-tanaka.json

取得に成功した場合、claim-tanaka.jsonには対象ID、提案カテゴリ、理由、担当者、取得時刻、有効期限、トークンが入ります。時刻とトークンは実行ごとに変わります。最初のコマンドの終了コードが0であり、JSONのclaimedがtrueであることを確認してから確定操作へ進みます。

この例では未確定項目が1件だけなので、有効期限内に別の担当者が取得すると、次の結果になります。

python3 review_leases.py claim review-leases.db --reviewer suzuki --lease-seconds 300
{"ok":true,"claimed":false}

claimed=falseは異常終了ではなく、現在取得できる項目がないことを表します。この結果にはトークンがないため、decideには使えません。

3. 必要なら期限を更新または解放する

レビューを続ける場合は、有効期限が切れる前にrenewを実行します。

python3 review_leases.py renew review-leases.db --claim-json claim-tanaka.json --lease-seconds 300

新しい期限は、以前の期限ではなくrenewの実行時刻を基準に計算されます。指定した秒数が現在の残り時間より短い場合は期限が前倒しされるため、通常は取得時と同じ値を指定し、失効間際になる前に更新します。更新結果のJSONに新しいclaim_expires_atが返りますが、保存済みのclaim-tanaka.json内の時刻は書き換わりません。トークンはそのまま利用できます。

確定せず別の担当者へ戻す場合はreleaseを使います。次のコマンドは、手順4とは別の分岐例です。手順4で同じ取得結果を使って確定する場合は実行しません。

python3 review_leases.py release review-leases.db --claim-json claim-tanaka.json

失効後のトークンでは更新も解放もできません。失効した項目は、次のclaimによって新しい担当者とトークンへ置き換えられます。

4. レビュー結果を確定する

提案された配送を請求へ修正する場合は、許可済みカテゴリを指定します。

python3 review_leases.py decide review-leases.db --claim-json claim-tanaka.json --decision corrected --category 請求
{"ok":true,"batch_id":"batch-20261005","record_id":"C-001","status":"corrected","final_category":"請求","reviewed_by":"tanaka"}

判定ごとの引数は次のとおりです。

  • accepted:提案カテゴリを採用する。--categoryは指定しない
  • corrected:提案と異なる許可カテゴリを--categoryで指定する
  • rejected:最終カテゴリを持たせない。--categoryは指定しない

確定後はトークンがデータベースから消えるため、同じ取得結果を再利用したdecideはclaim_not_foundで失敗します。失効後、まだ再取得されていないトークンを使った場合はclaim_expiredになります。

5. 保存状態を確認する

python3 review_leases.py show review-leases.db --batch-id batch-20261005 --record-id C-001
{"ok":true,"batch_id":"batch-20261005","record_id":"C-001","status":"corrected","proposed_category":"配送","final_category":"請求","reviewed_by":"tanaka","claimed_by":null,"claim_state":"not_applicable"}

showはトークンを表示しません。未確定項目のclaim_stateは、unclaimed、active、expiredのいずれかです。

同時取得を防ぐ仕組み

claimは、対象の検索前にBEGIN IMMEDIATEを実行します。SQLiteの書き込みトランザクションを確保してから、取得可能な行の検索と更新を同じトランザクション内で行います。このCLIだけがデータベースを更新する構成では、別プロセスが同じ行を同時に取得することを防げます。

ロック待機時間は5秒です。競合が長引いた場合はSQLiteエラーとなり、終了コード3を返すことがあります。高い並列処理性能を目的とした設計ではありません。

終了コード 意味
0 操作成功。取得対象がないclaimed=falseも含む
2 引数、カテゴリ、トークン、取得結果ファイル、有効期限、状態遷移など、入力または業務条件による拒否
3 SQLite処理、または想定外のファイル操作の失敗
130 Ctrl+Cによる中断

実運用へ入れる前の制約

  • 取得結果とデータベースにはトークンが含まれるため、アクセスできる利用者をOSの権限で限定する
  • 端末の時刻変更は有効期限の判定へ影響するため、各プロセスが同じ時刻基準を使える環境で運用する
  • プロセスが異常終了した場合、明示的な解放は行われず、有効期限までロックが残る
  • 期限切れ後に元担当者の作業画面を停止する仕組みはない。別の担当者が再取得した場合、元担当者からの確定は拒否されるが、レビュー作業自体は重複する可能性がある
  • ネットワーク共有上のSQLiteは、ファイルロックの挙動を検証できない限り使用しない
  • 多数の同時利用者、利用者認証、権限分離、通知が必要な場合は、SQLiteだけで要件を満たそうとしない
  • この実装が保存するのは現在状態だけであり、変更不能な監査ログは作らない。監査要件がある場合は、決定履歴を別の仕組みで記録する

まずはサンプル1件で、取得中の再取得拒否、期限内の確定、解放後の再取得を確認します。既存のレビューキューへ統合する場合は、データベースを退避したうえで、実際のテーブル名と制約に合わせた移行処理を別に用意してください。

第163回 AIの要確認結果を人へ回す:PythonとSQLiteで作るレビューキュー

第162回では、AIが返したJSONLの項目、型、許可カテゴリ、重複IDなどを検査しました。ただし、構造が正しいことと、判定内容が業務上正しいことは別です。needs_reviewがtrueの結果は、元データと照合してから確定します。

今回は、検証済みJSONLから要確認行だけをSQLiteへ取り込み、pending、accepted、corrected、rejectedの状態を管理します。Python標準ライブラリだけを使い、AIサービスへの通信や元データの更新は行いません。

今回作るレビューキューの契約

項目 仕様
入力 第162回と同じ4項目を持つ、BOMなしUTF-8の検証済みJSONL
登録対象 needs_reviewがtrueの行だけ
識別子 batch_idとrecord_idの組み合わせ。同じレコードを別バッチで再確認できる
カテゴリ データベース作成時に登録した許可カテゴリだけ
判定 提案を採用、許可カテゴリへ修正、または却下
再判定 このCLIでは確定済みの行を変更しない
書き込み バッチの取り込みと1件の判定を、それぞれSQLiteトランザクションで処理する
結果出力 ヘルプ表示を除き、出力に成功した場合は標準出力または標準エラー出力へ1行のJSONを返す

needs_review=falseの行は件数だけを数え、データベースには保存しません。検証済みJSONLは変更せず、別途保管します。

レビュー理由はAIの説明であり、正しさの証明ではありません。担当者はrecord_idを使い、権限のある元データと照合します。

SQLiteを使う範囲と制約

この実装は、1台の端末または少人数で扱う小規模なレビューキューを対象とします。複数人が同時に操作するWeb画面、認証、権限管理、改ざん検知は備えていません。--reviewerも利用者が入力する文字列であり、本人確認にはなりません。

  • Python 3.10以上を使用する
  • データベースはローカルディスクに置き、アクセス権とバックアップを管理する
  • 理由やレビュー注記に機密情報を入れる場合は、データベースとバックアップも同じ基準で保護する
  • カテゴリ方針を変えるときは、既存データとの整合性を確認して移行する。このスクリプトにはカテゴリ変更機能を設けない
  • 大人数での同時編集、ネットワーク共有上での利用、レビュー画面が必要な業務では、専用のデータベースとアプリケーションを検討する

1.動作確認用JSONLを用意する

次の架空データを、BOMなしUTF-8のvalid.jsonlとして保存します。1行に1つのJSONオブジェクトを書きます。

{"record_id":"C-001","category":"配送","needs_review":true,"reason":"配送先変更と到着日の質問が同時に含まれる"}
{"record_id":"C-002","category":"請求","needs_review":false,"reason":"請求書再発行の依頼が明確"}
{"record_id":"C-003","category":"配送","needs_review":true,"reason":"配送状況と請求額の両方に言及している"}

2.レビューキューを実装する

次のコードをreview_queue.pyとして保存します。

import argparse
import json
import os
import re
import sqlite3
import sys
from datetime import datetime, timezone
from pathlib import Path


SCHEMA_VERSION = 1
EXPECTED_KEYS = frozenset(
    {"record_id", "category", "needs_review", "reason"}
)
BATCH_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}")

CREATE_STATEMENTS = (
    """
    CREATE TABLE categories (
        category TEXT NOT NULL PRIMARY KEY,
        CHECK (length(category) > 0)
    )
    """,
    """
    CREATE TABLE review_batches (
        batch_id TEXT NOT NULL PRIMARY KEY,
        imported_at TEXT NOT NULL,
        source_name TEXT NOT NULL,
        total_rows INTEGER NOT NULL CHECK (total_rows >= 0),
        queued_rows INTEGER NOT NULL CHECK (queued_rows >= 0),
        skipped_rows INTEGER NOT NULL CHECK (skipped_rows >= 0),
        CHECK (length(batch_id) > 0),
        CHECK (length(source_name) > 0),
        CHECK (total_rows = queued_rows + skipped_rows)
    )
    """,
    """
    CREATE TABLE review_items (
        batch_id TEXT NOT NULL,
        record_id TEXT NOT NULL,
        proposed_category TEXT NOT NULL,
        reason TEXT NOT NULL,
        source_line INTEGER NOT NULL CHECK (source_line >= 1),
        status TEXT NOT NULL
            CHECK (status IN ('pending', 'accepted', 'corrected', 'rejected')),
        final_category TEXT,
        reviewer TEXT,
        review_note TEXT,
        decided_at TEXT,
        PRIMARY KEY (batch_id, record_id),
        UNIQUE (batch_id, source_line),
        CHECK (length(record_id) > 0),
        CHECK (length(reason) > 0),
        CHECK (
            (
                status = 'pending'
                AND final_category IS NULL
                AND reviewer IS NULL
                AND review_note IS NULL
                AND decided_at IS NULL
            )
            OR
            (
                status = 'accepted'
                AND final_category IS NOT NULL
                AND final_category = proposed_category
                AND reviewer IS NOT NULL
                AND length(reviewer) > 0
                AND review_note IS NOT NULL
                AND length(review_note) > 0
                AND decided_at IS NOT NULL
            )
            OR
            (
                status = 'corrected'
                AND final_category IS NOT NULL
                AND final_category <> proposed_category
                AND reviewer IS NOT NULL
                AND length(reviewer) > 0
                AND review_note IS NOT NULL
                AND length(review_note) > 0
                AND decided_at IS NOT NULL
            )
            OR
            (
                status = 'rejected'
                AND final_category IS NULL
                AND reviewer IS NOT NULL
                AND length(reviewer) > 0
                AND review_note IS NOT NULL
                AND length(review_note) > 0
                AND decided_at IS NOT NULL
            )
        ),
        FOREIGN KEY (batch_id)
            REFERENCES review_batches(batch_id) ON DELETE RESTRICT,
        FOREIGN KEY (proposed_category)
            REFERENCES categories(category) ON DELETE RESTRICT,
        FOREIGN KEY (final_category)
            REFERENCES categories(category) ON DELETE RESTRICT
    )
    """,
)


class AppError(Exception):
    def __init__(self, error, details=None):
        super().__init__(error)
        self.error = error
        self.details = details


class JsonArgumentParser(argparse.ArgumentParser):
    def error(self, message):
        raise AppError("invalid_arguments", {"message": message})


def emit(stream, payload):
    line = json.dumps(
        payload,
        ensure_ascii=False,
        separators=(",", ":"),
    ) + "\n"
    stream.write(line)
    stream.flush()


def has_control_character(value):
    return any(
        ord(character) < 32 or ord(character) == 127
        for character in value
    )


def validate_plain_text(value, field):
    if (
        value == ""
        or value != value.strip()
        or has_control_character(value)
    ):
        raise AppError("invalid_text_argument", {"field": field})
    return value


def validate_batch_id(value):
    if BATCH_PATTERN.fullmatch(value) is None:
        raise AppError(
            "invalid_batch_id",
            {
                "rule": (
                    "1-64 ASCII characters; first is a letter or digit; "
                    "the rest may also include dot, underscore, or hyphen"
                )
            },
        )
    return value


def validate_categories(values):
    categories = []
    seen = set()

    for value in values:
        category = validate_plain_text(value, "category")
        if category in seen:
            raise AppError("duplicate_category")
        seen.add(category)
        categories.append(category)

    return categories


def limit_value(raw_value):
    try:
        value = int(raw_value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("limit must be an integer") from exc

    if value < 1 or value > 500:
        raise argparse.ArgumentTypeError("limit must be between 1 and 500")
    return value


def utc_now():
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def connect_database(path):
    return sqlite3.connect(
        path,
        timeout=5.0,
        isolation_level=None,
    )


def configure_connection(connection):
    connection.row_factory = sqlite3.Row
    connection.execute("PRAGMA foreign_keys = ON")
    connection.execute("PRAGMA busy_timeout = 5000")


def safe_rollback(connection):
    try:
        connection.rollback()
    except sqlite3.Error:
        pass


def open_database(path):
    if not path.is_file():
        raise AppError("database_not_found", {"db": str(path)})

    connection = connect_database(path)
    try:
        configure_connection(connection)
        version = connection.execute("PRAGMA user_version").fetchone()[0]
        if version != SCHEMA_VERSION:
            raise AppError(
                "unsupported_schema_version",
                {"expected": SCHEMA_VERSION, "actual": version},
            )
    except BaseException:
        connection.close()
        raise

    return connection


def reject_constant(_value):
    raise ValueError("non-standard JSON number")


def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate JSON key")
        result[key] = value
    return result


def parse_result(raw_line, line_number):
    try:
        item = json.loads(
            raw_line,
            parse_constant=reject_constant,
            object_pairs_hook=unique_object,
        )
    except (json.JSONDecodeError, ValueError) as exc:
        raise AppError("invalid_json", {"line": line_number}) from exc

    if type(item) is not dict:
        raise AppError("json_object_required", {"line": line_number})

    keys = set(item)
    if keys != EXPECTED_KEYS:
        raise AppError(
            "invalid_fields",
            {
                "line": line_number,
                "missing": sorted(EXPECTED_KEYS - keys),
                "unknown": sorted(keys - EXPECTED_KEYS),
            },
        )

    for field in ("record_id", "category", "reason"):
        value = item[field]
        if type(value) is not str or value.strip() == "":
            raise AppError(
                "invalid_field_value",
                {"line": line_number, "field": field},
            )

    record_id = item["record_id"]
    if (
        record_id != record_id.strip()
        or has_control_character(record_id)
    ):
        raise AppError(
            "invalid_field_value",
            {"line": line_number, "field": "record_id"},
        )

    if type(item["needs_review"]) is not bool:
        raise AppError(
            "invalid_field_value",
            {"line": line_number, "field": "needs_review"},
        )

    return item


def cmd_init(args):
    categories = validate_categories(args.category)
    if args.db.exists():
        raise AppError("database_already_exists", {"db": str(args.db)})

    connection = None
    created = False

    try:
        descriptor = os.open(
            args.db,
            os.O_CREAT | os.O_EXCL | os.O_WRONLY,
            0o600,
        )
        created = True
        os.close(descriptor)

        connection = connect_database(args.db)
        configure_connection(connection)
        connection.execute("BEGIN IMMEDIATE")

        for statement in CREATE_STATEMENTS:
            connection.execute(statement)

        connection.executemany(
            "INSERT INTO categories(category) VALUES (?)",
            ((category,) for category in categories),
        )
        connection.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
        connection.commit()
    except BaseException:
        if connection is not None:
            safe_rollback(connection)
            connection.close()
            connection = None
        if created:
            try:
                args.db.unlink()
            except OSError:
                pass
        raise
    finally:
        if connection is not None:
            connection.close()

    return {
        "ok": True,
        "command": "init",
        "db": str(args.db),
        "schema_version": SCHEMA_VERSION,
        "category_count": len(categories),
    }


def cmd_import_results(args):
    batch_id = validate_batch_id(args.batch_id)
    if not args.input.is_file():
        raise AppError("input_not_found", {"input": str(args.input)})

    connection = open_database(args.db)
    total_rows = 0
    queued_rows = 0
    skipped_rows = 0
    seen_record_ids = {}
    imported_at = utc_now()

    try:
        connection.execute("BEGIN IMMEDIATE")

        existing_batch = connection.execute(
            "SELECT 1 FROM review_batches WHERE batch_id = ?",
            (batch_id,),
        ).fetchone()
        if existing_batch is not None:
            raise AppError("batch_already_exists", {"batch_id": batch_id})

        connection.execute(
            """
            INSERT INTO review_batches(
                batch_id, imported_at, source_name,
                total_rows, queued_rows, skipped_rows
            ) VALUES (?, ?, ?, 0, 0, 0)
            """,
            (batch_id, imported_at, args.input.name),
        )

        allowed_categories = {
            row["category"]
            for row in connection.execute("SELECT category FROM categories")
        }

        try:
            with args.input.open("r", encoding="utf-8", newline="") as source:
                for line_number, raw_line in enumerate(source, start=1):
                    total_rows += 1
                    if raw_line.strip() == "":
                        raise AppError("blank_line", {"line": line_number})

                    item = parse_result(raw_line, line_number)
                    record_id = item["record_id"]

                    if item["category"] not in allowed_categories:
                        raise AppError(
                            "category_not_allowed",
                            {"line": line_number},
                        )

                    if record_id in seen_record_ids:
                        raise AppError(
                            "duplicate_record_id",
                            {
                                "line": line_number,
                                "first_line": seen_record_ids[record_id],
                            },
                        )
                    seen_record_ids[record_id] = line_number

                    if item["needs_review"] is False:
                        skipped_rows += 1
                        continue

                    connection.execute(
                        """
                        INSERT INTO review_items(
                            batch_id, record_id, proposed_category,
                            reason, source_line, status
                        ) VALUES (?, ?, ?, ?, ?, 'pending')
                        """,
                        (
                            batch_id,
                            record_id,
                            item["category"],
                            item["reason"],
                            line_number,
                        ),
                    )
                    queued_rows += 1
        except UnicodeDecodeError as exc:
            raise AppError("input_not_utf8") from exc

        if total_rows == 0:
            raise AppError("empty_input")

        connection.execute(
            """
            UPDATE review_batches
            SET total_rows = ?, queued_rows = ?, skipped_rows = ?
            WHERE batch_id = ?
            """,
            (total_rows, queued_rows, skipped_rows, batch_id),
        )
        connection.commit()
    except BaseException:
        safe_rollback(connection)
        raise
    finally:
        connection.close()

    return {
        "ok": True,
        "command": "import-results",
        "batch_id": batch_id,
        "total_rows": total_rows,
        "queued_rows": queued_rows,
        "skipped_rows": skipped_rows,
    }


def cmd_pending(args):
    batch_id = validate_batch_id(args.batch_id)
    connection = open_database(args.db)

    try:
        connection.execute("BEGIN")
        batch = connection.execute(
            "SELECT 1 FROM review_batches WHERE batch_id = ?",
            (batch_id,),
        ).fetchone()
        if batch is None:
            raise AppError("batch_not_found", {"batch_id": batch_id})

        pending_count = connection.execute(
            """
            SELECT COUNT(*)
            FROM review_items
            WHERE batch_id = ? AND status = 'pending'
            """,
            (batch_id,),
        ).fetchone()[0]

        rows = connection.execute(
            """
            SELECT record_id, proposed_category, reason, source_line
            FROM review_items
            WHERE batch_id = ? AND status = 'pending'
            ORDER BY source_line
            LIMIT ?
            """,
            (batch_id, args.limit),
        ).fetchall()
        connection.commit()
    except BaseException:
        safe_rollback(connection)
        raise
    finally:
        connection.close()

    items = [
        {
            "record_id": row["record_id"],
            "proposed_category": row["proposed_category"],
            "reason": row["reason"],
            "source_line": row["source_line"],
        }
        for row in rows
    ]

    return {
        "ok": True,
        "command": "pending",
        "batch_id": batch_id,
        "pending_count": pending_count,
        "returned_count": len(items),
        "items": items,
    }


def cmd_decide(args):
    batch_id = validate_batch_id(args.batch_id)
    record_id = validate_plain_text(args.record_id, "record_id")
    reviewer = validate_plain_text(args.reviewer, "reviewer")
    note = validate_plain_text(args.note, "note")

    requested_category = None
    if args.decision == "correct":
        if args.final_category is None:
            raise AppError("final_category_required")
        requested_category = validate_plain_text(
            args.final_category,
            "final_category",
        )
    elif args.final_category is not None:
        raise AppError(
            "final_category_not_allowed_for_decision",
            {"decision": args.decision},
        )

    connection = open_database(args.db)

    try:
        connection.execute("BEGIN IMMEDIATE")
        row = connection.execute(
            """
            SELECT proposed_category, status
            FROM review_items
            WHERE batch_id = ? AND record_id = ?
            """,
            (batch_id, record_id),
        ).fetchone()

        if row is None:
            raise AppError("review_item_not_found")
        if row["status"] != "pending":
            raise AppError(
                "review_item_already_decided",
                {"status": row["status"]},
            )

        if args.decision == "accept":
            status = "accepted"
            final_category = row["proposed_category"]
        elif args.decision == "correct":
            status = "corrected"
            final_category = requested_category
            if final_category == row["proposed_category"]:
                raise AppError("correction_must_change_category")

            allowed = connection.execute(
                "SELECT 1 FROM categories WHERE category = ?",
                (final_category,),
            ).fetchone()
            if allowed is None:
                raise AppError("final_category_not_allowed")
        else:
            status = "rejected"
            final_category = None

        cursor = connection.execute(
            """
            UPDATE review_items
            SET status = ?, final_category = ?, reviewer = ?,
                review_note = ?, decided_at = ?
            WHERE batch_id = ? AND record_id = ? AND status = 'pending'
            """,
            (
                status,
                final_category,
                reviewer,
                note,
                utc_now(),
                batch_id,
                record_id,
            ),
        )
        if cursor.rowcount != 1:
            raise AppError("concurrent_update_detected")

        connection.commit()
    except BaseException:
        safe_rollback(connection)
        raise
    finally:
        connection.close()

    return {
        "ok": True,
        "command": "decide",
        "batch_id": batch_id,
        "record_id": record_id,
        "status": status,
        "final_category": final_category,
    }


def build_parser():
    parser = JsonArgumentParser(description="SQLite review queue")
    subparsers = parser.add_subparsers(dest="command", required=True)

    init_parser = subparsers.add_parser("init")
    init_parser.add_argument("--db", type=Path, required=True)
    init_parser.add_argument(
        "--category",
        action="append",
        required=True,
        help="Repeat for every allowed category",
    )
    init_parser.set_defaults(handler=cmd_init)

    import_parser = subparsers.add_parser("import-results")
    import_parser.add_argument("--db", type=Path, required=True)
    import_parser.add_argument("--input", type=Path, required=True)
    import_parser.add_argument("--batch-id", required=True)
    import_parser.set_defaults(handler=cmd_import_results)

    pending_parser = subparsers.add_parser("pending")
    pending_parser.add_argument("--db", type=Path, required=True)
    pending_parser.add_argument("--batch-id", required=True)
    pending_parser.add_argument("--limit", type=limit_value, default=50)
    pending_parser.set_defaults(handler=cmd_pending)

    decide_parser = subparsers.add_parser("decide")
    decide_parser.add_argument("--db", type=Path, required=True)
    decide_parser.add_argument("--batch-id", required=True)
    decide_parser.add_argument("--record-id", required=True)
    decide_parser.add_argument(
        "--decision",
        choices=("accept", "correct", "reject"),
        required=True,
    )
    decide_parser.add_argument("--final-category")
    decide_parser.add_argument("--reviewer", required=True)
    decide_parser.add_argument("--note", required=True)
    decide_parser.set_defaults(handler=cmd_decide)

    return parser


def main():
    exit_code = 0
    stream = sys.stdout

    try:
        args = build_parser().parse_args()
        payload = args.handler(args)
    except AppError as exc:
        payload = {"ok": False, "error": exc.error}
        if exc.details is not None:
            payload["details"] = exc.details
        stream = sys.stderr
        exit_code = 2
    except KeyboardInterrupt:
        payload = {"ok": False, "error": "interrupted"}
        stream = sys.stderr
        exit_code = 130
    except sqlite3.Error as exc:
        payload = {
            "ok": False,
            "error": "database_error",
            "details": {"type": type(exc).__name__},
        }
        stream = sys.stderr
        exit_code = 3
    except (OSError, UnicodeError) as exc:
        payload = {
            "ok": False,
            "error": "io_error",
            "details": {"type": type(exc).__name__},
        }
        stream = sys.stderr
        exit_code = 3
    except Exception as exc:
        payload = {
            "ok": False,
            "error": "unexpected_error",
            "details": {"type": type(exc).__name__},
        }
        stream = sys.stderr
        exit_code = 3

    try:
        emit(stream, payload)
    except KeyboardInterrupt:
        return 130
    except (OSError, UnicodeError):
        return 3
    except Exception:
        return 3

    return exit_code


if __name__ == "__main__":
    sys.exit(main())

接続ではisolation_level=Noneを指定し、自動的なトランザクション開始に依存しません。読み取りはBEGIN、取り込みと判定は書き込みを先に確保するBEGIN IMMEDIATEで開始し、成功時だけ明示的にコミットします。

emit()はJSONと改行を一度に書き、続けてflush()します。成功結果を含む最終出力も例外処理の対象にするため、出力時のOSErrorやUnicodeErrorは終了コード3、出力中の割り込みは終了コード130になります。ただし、出力先が閉じている場合などは、JSONが出ないか途中までしか書かれないことがあります。

取り込み時にも項目、型、重複キーなどを確認します。ただし、第162回で設定した行サイズや文字数の上限を、このスクリプトが代替するわけではありません。第162回の検証を通過したvalid.jsonlだけを入力にします。

3.許可カテゴリを固定してデータベースを作る

同名のreview.dbが存在しないディレクトリで実行します。--categoryは必要な数だけ繰り返せます。

python review_queue.py init \
  --db review.db \
  --category 配送 \
  --category 請求 \
  --category その他

成功すると、標準出力へ次の1行が出ます。

{"ok":true,"command":"init","db":"review.db","schema_version":1,"category_count":3}

既存ファイルは上書きしません。作成時には排他的なファイル作成を使います。POSIX環境ではモード0600を指定し、実際の権限にはOSのumaskも適用されます。保存先のアクセス制御も確認してください。

4.検証済み結果をバッチとして取り込む

python review_queue.py import-results \
  --db review.db \
  --input valid.jsonl \
  --batch-id 2026-10-02-a

サンプルでは3行を読み、2行をレビュー対象として登録します。

{"ok":true,"command":"import-results","batch_id":"2026-10-02-a","total_rows":3,"queued_rows":2,"skipped_rows":1}

バッチIDの重複、空行、不正なJSON、重複したrecord_id、未知のカテゴリが1件でもあれば、バッチ全体をロールバックします。一部の行だけが登録された状態にはしません。

5.未確認の項目を取得する

python review_queue.py pending \
  --db review.db \
  --batch-id 2026-10-02-a \
  --limit 50

出力は1つのJSONオブジェクトです。理由に改行などが含まれていても、JSON文字列としてエスケープされます。

{"ok":true,"command":"pending","batch_id":"2026-10-02-a","pending_count":2,"returned_count":2,"items":[{"record_id":"C-001","proposed_category":"配送","reason":"配送先変更と到着日の質問が同時に含まれる","source_line":1},{"record_id":"C-003","proposed_category":"配送","reason":"配送状況と請求額の両方に言及している","source_line":3}]}

pending_countは未確認の総数、returned_countは今回返した件数です。総数が上限を超える場合は、表示された項目を判定してから同じコマンドを再実行します。

6.採用、修正、却下を記録する

1つの項目に実行できる判定は1回だけです。ここでは採用例をC-001に実行した後、C-003について「修正」または「却下」のどちらか一方を選びます。C-003の2つの例は相互に排他的であり、同じデータベースへ続けて実行する例ではありません。

AIの提案カテゴリを採用する

python review_queue.py decide \
  --db review.db \
  --batch-id 2026-10-02-a \
  --record-id C-001 \
  --decision accept \
  --reviewer sato \
  --note 原文照合済み
{"ok":true,"command":"decide","batch_id":"2026-10-02-a","record_id":"C-001","status":"accepted","final_category":"配送"}

C-003を別の許可カテゴリへ修正する場合

python review_queue.py decide \
  --db review.db \
  --batch-id 2026-10-02-a \
  --record-id C-003 \
  --decision correct \
  --final-category 請求 \
  --reviewer sato \
  --note 請求内容を主題として確定
{"ok":true,"command":"decide","batch_id":"2026-10-02-a","record_id":"C-003","status":"corrected","final_category":"請求"}

correctでは、登録済みの許可カテゴリだけを指定できます。提案と同じカテゴリは、修正にならないため拒否します。

C-003を却下する場合

次の例を試す場合は、C-003の修正例を実行しないでください。すでに修正例を実行した場合は、別名の新しいデータベースを作成し、手順3と4をそのデータベース名に置き換えて実行してから、次のコマンドでも同じ名前を指定します。

却下では--decision rejectを指定し、--final-categoryは付けません。保存される最終カテゴリはNULLになり、JSON出力ではnullになります。

python review_queue.py decide \
  --db review.db \
  --batch-id 2026-10-02-a \
  --record-id C-003 \
  --decision reject \
  --reviewer sato \
  --note 元データだけでは分類不能
{"ok":true,"command":"decide","batch_id":"2026-10-02-a","record_id":"C-003","status":"rejected","final_category":null}

いずれの判定も、対象がpendingである場合に一度だけ実行できます。確定済みレコードへの再実行は終了コード2で停止し、保存済みの判定を変更しません。

終了コードと失敗時の扱い

終了コード 意味 確認事項
0 コマンド処理と結果出力が成功 成功JSONの内容を確認する
2 引数、入力、カテゴリ、状態遷移の契約違反 入力内容や現在の状態を直す
3 ファイル操作、データベース、結果出力、または予期しない実行エラー 成功JSONがなければ、再実行前にデータベースの状態を確認する
130 処理中または結果出力中の割り込み 再実行前にデータベースの状態を確認する

処理中の例外では未確定のトランザクションをロールバックします。ただし、コミット直後や結果出力中に入出力エラーや割り込みが起きた場合、データベースへの反映が完了していても、呼び出し側が完全な成功JSONを受け取れないことがあります。出力先そのものが利用できない場合は、エラーJSONも出力できません。終了コードだけで未反映と決めず、同じbatch_idや対象レコードを照会してから再実行します。

SQLiteの待機時間は5秒です。別プロセスが書き込みを保持し続けた場合は、データベースエラーとして終了します。実行中の別作業を確認してから再試行します。

運用へ入れる前の確認

  1. 第162回の検証ゲートを通過したJSONLだけを入力にする
  2. カテゴリ一覧を業務担当者と確認してからデータベースを作る
  3. 処理単位ごとに重複しないbatch_idを付ける
  4. AIの理由だけで判断せず、権限のある元データを照合する
  5. pending_countが0になったことを確認する
  6. データベースを元データと同じ機密区分で保護し、必要な期間だけ保存する

pending_count=0が示すのは、そのバッチの要確認行に判定が付いたことだけです。自動判定された行の正しさや、後続システムへの反映完了までは保証しません。

まずはサンプル用の新しいデータベースで、取り込み、一覧取得、判定、再判定の拒否までを確認します。その後、業務データへ適用してください。

仕様を確認する一次資料

第162回 AIのJSONL結果をそのまま使わない:Pythonで作る行単位の検証ゲート

第161回では、AIなどの後続処理へ渡すCSVを許可列だけに絞りました。入力を整えても、AIが返した結果を無検査でCSV更新や集計へ渡すことはできません。真偽値が文字列になっている、許可していない分類名が混ざる、同じIDが二度返る、といった問題があるためです。

今回は、AIの分類結果を1行1オブジェクトのJSONLとして受け取り、Python標準ライブラリだけで検査します。特定のAIサービスへ通信するコードではありません。実際のレスポンスをJSONLへ変換する部分は、利用するサービスの仕様に合わせて分離します。

今回固定する結果契約

各行のJSONオブジェクトは、次の4項目だけを持つものとします。

項目 条件
record_id 空でない文字列。前後空白、U+0000~U+001F、U+007Fを認めない。項目、型、値の検査を通過した行同士で重複させない
category ポリシーで許可した文字列のいずれか
needs_review JSONの真偽値であるtrueまたはfalse。文字列は不可
reason 空白だけではない文字列。ポリシーで決めた長さ以内

不足項目だけでなく、未知の項目が増えた場合も拒否します。PythonのJSONデコーダが既定で受け入れるNaNやInfinityも拒否します。同じオブジェクト内の重複キーも、後に書かれた値で上書きせずエラーにします。

長さの判定にはPythonのlen()を使います。結合文字や絵文字の並びを、画面上の見た目どおりの1文字として数える処理ではありません。上限は後続システムと同じ数え方で決める必要があります。

各行について記録するのは、最初に検出したエラー1件です。検査結果は新しいディレクトリへ出力し、元のJSONLは変更しません。

生成物 用途
valid.jsonl 契約を通過した行を、項目順をそろえて再出力
errors.jsonl 拒否した物理行番号とエラーコードを記録。元の値は複製しない
summary.json 件数、判定、入力とポリシーのSHA-256を記録

入力が0バイトの場合は、行ごとの判定を作れません。契約違反として終了コード2を返し、作成途中の出力ディレクトリを削除します。

1.検査ポリシーを用意する

次をUTF-8のresult_policy.jsonとして保存します。UTF-8 BOMが付いていても読み込めます。

{
  "schema_version": 1,
  "allowed_categories": ["配送", "請求", "その他"],
  "max_record_id_chars": 64,
  "max_reason_chars": 200,
  "max_line_bytes": 8192
}

64、200、8192は、この例の設定値です。AIサービス共通の上限ではありません。実データのID仕様、確認画面、後続システムの制約から決めます。max_line_bytesは1物理行のバイト数で、改行がある場合は改行コードも含みます。

2.拒否例を含むJSONLを用意する

次をUTF-8のai_results.jsonlとして保存します。JSONLなので、配列を囲む角括弧や行末のカンマは付けません。

{"record_id":"R-001","category":"配送","needs_review":false,"reason":"到着日の確認依頼"}
{"record_id":"R-002","category":"返金","needs_review":false,"reason":"許可されていない分類"}
{"record_id":"R-003","category":"請求","needs_review":"false","reason":"真偽値が文字列"}

1行目は契約どおりです。2行目の「返金」は許可カテゴリにありません。3行目のneeds_reviewは真偽値ではなく文字列です。

3.行単位の検証ゲートを実装する

次をvalidate_ai_jsonl.pyとして保存します。Python 3.10以上と標準ライブラリだけを使います。

from __future__ import annotations

import argparse
import hashlib
import json
import shutil
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any, BinaryIO, Iterator


FIELDS = {"record_id", "category", "needs_review", "reason"}


class ContractError(Exception):
    pass


class RowError(Exception):
    def __init__(self, code: str):
        super().__init__(code)
        self.code = code


@dataclass(frozen=True)
class Policy:
    allowed_categories: frozenset[str]
    max_record_id_chars: int
    max_reason_chars: int
    max_line_bytes: int


def has_surrogate(text: str) -> bool:
    return any(0xD800 <= ord(char) <= 0xDFFF for char in text)


def strict_object(pairs: list[tuple[str, object]]) -> dict[str, object]:
    result: dict[str, object] = {}
    for key, value in pairs:
        if key in result:
            raise RowError("DUPLICATE_KEY")
        result[key] = value
    return result


def reject_constant(_: str) -> object:
    raise RowError("NON_STANDARD_NUMBER")


def load_policy(path: Path) -> tuple[Policy, str]:
    raw = path.read_bytes()
    digest = hashlib.sha256(raw).hexdigest()

    try:
        text = raw.decode("utf-8-sig")
    except UnicodeDecodeError as exc:
        raise ContractError("policy must be UTF-8") from exc

    try:
        data = json.loads(
            text,
            object_pairs_hook=strict_object,
            parse_constant=reject_constant,
        )
    except (ValueError, RecursionError, RowError) as exc:
        raise ContractError("policy is not valid strict JSON") from exc

    expected = {
        "schema_version",
        "allowed_categories",
        "max_record_id_chars",
        "max_reason_chars",
        "max_line_bytes",
    }
    if type(data) is not dict or set(data) != expected:
        raise ContractError("policy fields do not match the required schema")
    if type(data["schema_version"]) is not int or data["schema_version"] != 1:
        raise ContractError("schema_version must be the integer 1")

    categories = data["allowed_categories"]
    if type(categories) is not list or not categories:
        raise ContractError("allowed_categories must be a non-empty array")
    if any(
        type(item) is not str
        or not item
        or item != item.strip()
        or has_surrogate(item)
        for item in categories
    ):
        raise ContractError("each category must be a trimmed Unicode string")
    if len(set(categories)) != len(categories):
        raise ContractError("allowed_categories contains duplicates")

    limits: dict[str, int] = {}
    for name in (
        "max_record_id_chars",
        "max_reason_chars",
        "max_line_bytes",
    ):
        value = data[name]
        if type(value) is not int or value <= 0:
            raise ContractError(f"{name} must be a positive integer")
        limits[name] = value

    return (
        Policy(
            allowed_categories=frozenset(categories),
            max_record_id_chars=limits["max_record_id_chars"],
            max_reason_chars=limits["max_reason_chars"],
            max_line_bytes=limits["max_line_bytes"],
        ),
        digest,
    )


def limited_lines(
    stream: BinaryIO,
    limit: int,
    digest: Any,
) -> Iterator[tuple[int, bytes | None]]:
    line_number = 0
    while True:
        chunk = stream.readline(limit + 1)
        if chunk == b"":
            return

        line_number += 1
        digest.update(chunk)
        too_long = len(chunk) > limit

        if too_long and not chunk.endswith(b"\n"):
            while True:
                rest = stream.readline(limit + 1)
                if rest == b"":
                    break
                digest.update(rest)
                if rest.endswith(b"\n"):
                    break

        yield line_number, None if too_long else chunk


def validate_record(value: object, policy: Policy) -> dict[str, object]:
    if type(value) is not dict:
        raise RowError("NOT_OBJECT")
    if set(value) != FIELDS:
        raise RowError("FIELDS_MISMATCH")

    record_id = value["record_id"]
    if type(record_id) is not str:
        raise RowError("RECORD_ID_TYPE")
    if not record_id or record_id != record_id.strip():
        raise RowError("RECORD_ID_FORMAT")
    if len(record_id) > policy.max_record_id_chars:
        raise RowError("RECORD_ID_TOO_LONG")
    if has_surrogate(record_id):
        raise RowError("RECORD_ID_UNICODE")
    if any(ord(char) < 32 or ord(char) == 127 for char in record_id):
        raise RowError("RECORD_ID_CONTROL_CHAR")

    category = value["category"]
    if type(category) is not str:
        raise RowError("CATEGORY_TYPE")
    if category not in policy.allowed_categories:
        raise RowError("CATEGORY_NOT_ALLOWED")

    needs_review = value["needs_review"]
    if type(needs_review) is not bool:
        raise RowError("NEEDS_REVIEW_TYPE")

    reason = value["reason"]
    if type(reason) is not str:
        raise RowError("REASON_TYPE")
    if not reason.strip():
        raise RowError("REASON_EMPTY")
    if len(reason) > policy.max_reason_chars:
        raise RowError("REASON_TOO_LONG")
    if "\x00" in reason or has_surrogate(reason):
        raise RowError("REASON_UNICODE")

    return {
        "record_id": record_id,
        "category": category,
        "needs_review": needs_review,
        "reason": reason,
    }


def json_line(value: object) -> bytes:
    text = json.dumps(value, ensure_ascii=False, separators=(",", ":"))
    return (text + "\n").encode("utf-8")


def write_rejection(
    stream: BinaryIO,
    line_number: int,
    code: str,
) -> None:
    stream.write(json_line({"line": line_number, "code": code}))


def process(
    input_path: Path,
    output_dir: Path,
    policy: Policy,
    policy_sha256: str,
) -> dict[str, object]:
    input_digest = hashlib.sha256()
    seen_ids: set[str] = set()
    total_lines = 0
    valid_records = 0
    rejected_records = 0

    with (
        input_path.open("rb") as source,
        (output_dir / "valid.jsonl").open("xb") as valid_stream,
        (output_dir / "errors.jsonl").open("xb") as error_stream,
    ):
        for line_number, raw in limited_lines(
            source,
            policy.max_line_bytes,
            input_digest,
        ):
            total_lines += 1
            error_code: str | None = None
            record: dict[str, object] | None = None

            if raw is None:
                error_code = "LINE_TOO_LONG"
            else:
                try:
                    encoding = "utf-8-sig" if line_number == 1 else "utf-8"
                    text = raw.decode(encoding)
                except UnicodeDecodeError:
                    error_code = "INVALID_UTF8"
                else:
                    if not text.strip():
                        error_code = "BLANK_LINE"
                    else:
                        try:
                            value = json.loads(
                                text,
                                object_pairs_hook=strict_object,
                                parse_constant=reject_constant,
                            )
                            record = validate_record(value, policy)
                        except (ValueError, RecursionError):
                            error_code = "JSON_SYNTAX"
                        except RowError as exc:
                            error_code = exc.code

            if error_code is None and record is not None:
                record_id = str(record["record_id"])
                if record_id in seen_ids:
                    error_code = "DUPLICATE_RECORD_ID"
                else:
                    seen_ids.add(record_id)
                    valid_stream.write(json_line(record))
                    valid_records += 1

            if error_code is not None:
                write_rejection(error_stream, line_number, error_code)
                rejected_records += 1

    if total_lines == 0:
        raise ContractError("input JSONL is empty")

    summary: dict[str, object] = {
        "schema_version": 1,
        "status": "ok" if rejected_records == 0 else "rejected",
        "input_file": input_path.name,
        "input_sha256": input_digest.hexdigest(),
        "policy_sha256": policy_sha256,
        "total_lines": total_lines,
        "valid_records": valid_records,
        "rejected_records": rejected_records,
    }
    with (output_dir / "summary.json").open("xb") as stream:
        stream.write(json_line(summary))
    return summary


def emit(stream: Any, value: object) -> None:
    text = json.dumps(value, ensure_ascii=False, separators=(",", ":"))
    stream.write(text + "\n")


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Validate provider-neutral AI result JSONL."
    )
    parser.add_argument("--input", required=True, type=Path)
    parser.add_argument("--policy", required=True, type=Path)
    parser.add_argument("--output-dir", required=True, type=Path)
    args = parser.parse_args()

    created_output = False
    try:
        if not args.input.is_file():
            raise ContractError("input must be an existing file")
        if not args.policy.is_file():
            raise ContractError("policy must be an existing file")
        if args.output_dir.exists():
            raise ContractError("output directory already exists")
        if not args.output_dir.parent.is_dir():
            raise ContractError("output parent directory does not exist")

        policy, policy_sha256 = load_policy(args.policy)

        try:
            args.output_dir.mkdir()
        except FileExistsError as exc:
            raise ContractError("output directory already exists") from exc
        created_output = True

        marker = args.output_dir / ".incomplete"
        marker.write_bytes(b"incomplete\n")
        summary = process(
            args.input,
            args.output_dir,
            policy,
            policy_sha256,
        )
        marker.unlink()

        result = {
            "status": summary["status"],
            "total_lines": summary["total_lines"],
            "valid_records": summary["valid_records"],
            "rejected_records": summary["rejected_records"],
            "output_dir": str(args.output_dir),
        }
        emit(sys.stdout, result)
        return 0 if summary["status"] == "ok" else 1

    except ContractError as exc:
        if created_output:
            shutil.rmtree(args.output_dir, ignore_errors=True)
        emit(
            sys.stderr,
            {
                "status": "error",
                "code": "CONTRACT_ERROR",
                "message": str(exc),
            },
        )
        return 2
    except KeyboardInterrupt:
        if created_output:
            shutil.rmtree(args.output_dir, ignore_errors=True)
        emit(sys.stderr, {"status": "error", "code": "INTERRUPTED"})
        return 130
    except OSError as exc:
        if created_output:
            shutil.rmtree(args.output_dir, ignore_errors=True)
        emit(
            sys.stderr,
            {
                "status": "error",
                "code": "IO_ERROR",
                "message": str(exc),
            },
        )
        return 3
    except Exception:
        if created_output:
            shutil.rmtree(args.output_dir, ignore_errors=True)
        emit(
            sys.stderr,
            {"status": "error", "code": "UNEXPECTED_ERROR"},
        )
        return 3


if __name__ == "__main__":
    raise SystemExit(main())

処理順と失敗時の扱い

  1. 入力ファイル、ポリシー、未作成の出力先を確認する
  2. ポリシーJSONの重複キー、項目、型、値を確認する
  3. 出力ディレクトリを排他的に作り、.incompleteを置く
  4. 入力をバイト列で1行ずつ読み、行サイズ、UTF-8、JSON構文、項目、型、値、ID重複を確認する
  5. 合格行とエラー情報を分けて書き、最後にsummary.jsonを作る
  6. 完了後だけ.incompleteを削除し、判定に応じた終了コードを返す

行が上限を超えた場合は、その行全体をメモリへためず、改行まで読み捨てて次の行へ進みます。入力全体もメモリへ載せません。ただし、重複検査用のrecord_id集合は保持するため、ユニークID数に応じてメモリ使用量が増えます。

通常の例外とCtrl+Cでは、この実行が作った出力ディレクトリの削除を試みます。プロセスの強制終了や電源断では、途中のディレクトリが残る場合があります。同じ出力先へ複数のプロセスを同時実行しない運用も必要です。

4.実行して終了コードを確認する

次はbash系シェルでの実行例です。

python3 validate_ai_jsonl.py \
  --input ai_results.jsonl \
  --policy result_policy.json \
  --output-dir checked
echo $?

標準出力と終了コードは次のようになります。

{"status":"rejected","total_lines":3,"valid_records":1,"rejected_records":2,"output_dir":"checked"}
1

終了コード1は、検査処理は完了したものの、少なくとも1行を拒否したことを表します。後続のCSV更新や登録処理は止めます。

終了コード 意味
0 全行が契約を通過
1 1行以上を拒否。検査結果は出力済み
2 引数、入力、ポリシー、出力先の契約違反
3 ファイル操作または予期しない実行エラー
130 Ctrl+Cによる中断

argparseが検出する必須引数の不足や未知の引数も終了コード2です。この場合はスクリプト独自のJSONではなく、使用法が標準エラー出力へ表示されます。

5.合格行とエラーを確認する

checked/valid.jsonlには1行目だけが入ります。

{"record_id":"R-001","category":"配送","needs_review":false,"reason":"到着日の確認依頼"}

checked/errors.jsonlには、元の理由文やカテゴリ値を複製せず、位置とコードだけを記録します。

{"line":2,"code":"CATEGORY_NOT_ALLOWED"}
{"line":3,"code":"NEEDS_REVIEW_TYPE"}

summary.jsonには入力とポリシーのSHA-256も入ります。ハッシュは同じ内容かを照合するための値であり、暗号化や匿名化ではありません。

主なエラーコードは次のとおりです。

  • JSON_SYNTAX:JSONとして解析できない
  • DUPLICATE_KEY:同じオブジェクト内に同名キーがある
  • NON_STANDARD_NUMBER:NaNやInfinityがある
  • FIELDS_MISMATCH:必須項目の不足または未知項目の追加
  • CATEGORY_NOT_ALLOWED:許可されていない分類
  • NEEDS_REVIEW_TYPE:真偽値の型が違う
  • DUPLICATE_RECORD_ID:検査を通過した行のIDが重複
  • LINE_TOO_LONG、INVALID_UTF8:行サイズまたは文字コードの違反

6.全行が合格する経路を確認する

次をai_results_ok.jsonlとして保存します。

{"record_id":"R-001","category":"配送","needs_review":false,"reason":"到着日の確認依頼"}
{"record_id":"R-002","category":"請求","needs_review":true,"reason":"請求内容を担当者が確認する必要あり"}

既存ディレクトリを上書きしないため、別の出力先を指定します。

python3 validate_ai_jsonl.py \
  --input ai_results_ok.jsonl \
  --policy result_policy.json \
  --output-dir checked-ok
echo $?

この入力では次の結果になります。

{"status":"ok","total_lines":2,"valid_records":2,"rejected_records":0,"output_dir":"checked-ok"}
0

自動処理へ組み込む場合は、終了コードが0で、.incompleteがなく、summary.jsonのstatusがokであることを確認してから次へ進めます。終了コード1で作られたvalid.jsonlは調査用の部分結果です。完全な結果として登録してはいけません。

このゲートが判定しないこと

  • 分類や理由が業務上正しいか
  • reasonに事実誤認、機密情報、不適切な文言がないか
  • 依頼したIDがすべて返ったか、依頼していないIDが追加されたか
  • needs_reviewの判断基準が妥当か
  • AIサービス固有の利用規約、保存条件、データ送信可否

このスクリプトが確認するのは、構文と決めたデータ契約です。意味の正しさは保証しません。反映前には依頼時と結果のID一覧を突合し、欠落と余分なIDを検出します。そのうえで、needs_reviewが真の行や業務上重要な分類を人が確認します。

導入前の確認

正常な1行、未知カテゴリの1行、型違いの1行を用意し、終了コード0と1の両方を確認します。確認できたら、「AI結果を受け取る処理」と「CSVや業務システムへ反映する処理」の間に、このゲートを置きます。

第161回 AIへ渡すCSVを必要列だけに絞る:Pythonで作る許可列エクスポートゲート

問い合わせCSVをAIで分類するとき、氏名やメールアドレスまで入力に残す必要がない場合があります。手作業で列を削除する運用では、削除漏れや担当者ごとの手順差を防げません。

この記事では、ポリシーで許可した列だけを別のCSVへ書き出すゲートを、Python標準ライブラリだけで作ります。このスクリプトはAIサービスへ通信しません。生成したCSVを外部へ渡してよいかは、組織の規則、契約、データの内容に基づいて別途確認します。

今回作る処理の契約

  • 出力列は allowed_columns に記載した列だけとし、記載順も維持する
  • 禁止列が許可リストにも含まれていたら、ポリシーの誤りとして拒否する
  • ヘッダーの重複、許可列の不足、行の列数不一致を拒否する
  • 既存の出力ファイルを上書きしない
  • ヘルプ表示を除き、標準出力には1行のJSONを出す
  • データ行のセル値は結果JSONへ出さない

forbidden_columns は、ポリシー編集時の事故を防ぐための二重チェックです。入力CSVに禁止列が存在すること自体はエラーではありません。その列を出力対象へ含めようとした場合に停止します。

終了コード 意味
0 出力完了
2 引数、入力CSV、またはポリシーを拒否
3 ファイル操作または予期しない実行エラー
130 Ctrl+Cによる中断

1. サンプルCSVを用意する

動作確認用の架空データを source.csv として保存します。文字コードはUTF-8です。

case_id,customer_name,email,category,request_text,internal_note
C-001,山田花子,hanako@example.com,配送,到着日を確認したい,返金履歴あり
C-002,佐藤太郎,taro@example.com,請求,領収書を再発行したい,本人確認済み

2. 許可列と禁止列を定義する

次の内容を export-policy.json として保存します。出力CSVの列順は allowed_columns の順番になります。

{
  "allowed_columns": [
    "case_id",
    "category",
    "request_text"
  ],
  "forbidden_columns": [
    "customer_name",
    "email",
    "phone",
    "address",
    "internal_note"
  ]
}

ここでの列名は例です。実際には、自社のCSV仕様と利用目的に合わせて明示的に決めます。自由記述欄には氏名や連絡先が入力される可能性があるため、列名だけの検査では安全性を保証できません。

3. エクスポートゲートを実装する

次のコードを export_allowed_csv.py として保存します。Python 3.10以降を想定し、外部ライブラリは使いません。

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import csv
import json
import os
import sys
from pathlib import Path
from typing import Any

EXIT_OK = 0
EXIT_REJECTED = 2
EXIT_OPERATIONAL_ERROR = 3
EXIT_INTERRUPTED = 130


class RejectedError(Exception):
    """引数、入力、またはポリシーを拒否した。"""


class OperationalError(Exception):
    """ファイル操作など、実行環境のエラー。"""


class JsonArgumentParser(argparse.ArgumentParser):
    def error(self, message: str) -> None:
        raise RejectedError(f"引数エラー: {message}")


def emit(payload: dict[str, Any]) -> None:
    print(json.dumps(payload, ensure_ascii=False, separators=(",", ":")))


def find_duplicates(names: list[str]) -> list[str]:
    seen: set[str] = set()
    duplicates: set[str] = set()
    for name in names:
        if name in seen:
            duplicates.add(name)
        seen.add(name)
    return sorted(duplicates)


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 RejectedError(
                f"ポリシーJSONに重複キーがあります: {key}"
            )
        result[key] = value
    return result


def read_name_list(
    data: dict[str, Any], key: str, *, allow_empty: bool
) -> list[str]:
    value = data[key]
    if not isinstance(value, list):
        raise RejectedError(f"{key} は配列で指定してください")
    if not allow_empty and not value:
        raise RejectedError(f"{key} を1件以上指定してください")

    names: list[str] = []
    for item in value:
        if (
            not isinstance(item, str)
            or not item
            or item != item.strip()
            or "\r" in item
            or "\n" in item
        ):
            raise RejectedError(
                f"{key} の列名に空欄、前後空白、または改行があります"
            )
        names.append(item)

    duplicates = find_duplicates(names)
    if duplicates:
        raise RejectedError(
            f"{key} に重複があります: {', '.join(duplicates)}"
        )
    return names


def load_policy(path: Path) -> list[str]:
    try:
        text = path.read_text(encoding="utf-8-sig")
    except UnicodeDecodeError as exc:
        raise RejectedError("ポリシーはUTF-8で読めません") from exc
    except OSError as exc:
        raise OperationalError(f"ポリシーを読めません: {path}") from exc

    try:
        raw = json.loads(text, object_pairs_hook=unique_object)
    except json.JSONDecodeError as exc:
        raise RejectedError(
            f"ポリシーJSONの構文エラーです(行{exc.lineno}、列{exc.colno})"
        ) from exc

    if not isinstance(raw, dict):
        raise RejectedError("ポリシーの最上位はオブジェクトにしてください")

    expected_keys = {"allowed_columns", "forbidden_columns"}
    actual_keys = set(raw)
    if actual_keys != expected_keys:
        details: list[str] = []
        missing = sorted(expected_keys - actual_keys)
        extra = sorted(actual_keys - expected_keys)
        if missing:
            details.append(f"不足: {', '.join(missing)}")
        if extra:
            details.append(f"未定義: {', '.join(extra)}")
        raise RejectedError(
            f"ポリシーのキーが不正です({'; '.join(details)})"
        )

    allowed = read_name_list(raw, "allowed_columns", allow_empty=False)
    forbidden = set(
        read_name_list(raw, "forbidden_columns", allow_empty=True)
    )

    overlap = sorted(set(allowed) & forbidden)
    if overlap:
        raise RejectedError(
            "allowed_columns と forbidden_columns が重複しています: "
            + ", ".join(overlap)
        )
    return allowed


def validate_header(
    header: list[str] | None, allowed: list[str]
) -> list[str]:
    if header is None:
        raise RejectedError("入力CSVが空です")

    invalid = [
        name
        for name in header
        if not name
        or name != name.strip()
        or "\r" in name
        or "\n" in name
    ]
    if invalid:
        raise RejectedError(
            "CSVヘッダーに空欄、前後空白、または改行があります"
        )

    duplicates = find_duplicates(header)
    if duplicates:
        raise RejectedError(
            f"CSVヘッダーに重複があります: {', '.join(duplicates)}"
        )

    missing = [name for name in allowed if name not in header]
    if missing:
        raise RejectedError(
            f"入力CSVに許可列がありません: {', '.join(missing)}"
        )
    return header


def same_path(left: Path, right: Path) -> bool:
    try:
        return left.resolve(strict=False) == right.resolve(strict=False)
    except (OSError, RuntimeError) as exc:
        raise OperationalError("入出力パスを解決できません") from exc


def cleanup_partial(path: Path) -> None:
    try:
        path.unlink()
    except FileNotFoundError:
        return
    except OSError as exc:
        raise OperationalError(
            f"不完全な出力ファイルを削除できません: {path}"
        ) from exc


def export_allowed_csv(
    source_path: Path, policy_path: Path, output_path: Path
) -> dict[str, Any]:
    if same_path(source_path, output_path):
        raise RejectedError("入力元と出力先に同じパスは指定できません")
    if output_path.exists() or output_path.is_symlink():
        raise RejectedError(f"出力先が既に存在します: {output_path}")
    if not output_path.parent.exists() or not output_path.parent.is_dir():
        raise OperationalError(
            f"出力先ディレクトリを利用できません: {output_path.parent}"
        )

    allowed = load_policy(policy_path)
    output_created = False
    completed = False
    reader = None
    header: list[str] = []
    row_count = 0

    try:
        with source_path.open(
            "r", encoding="utf-8-sig", newline=""
        ) as source:
            reader = csv.DictReader(source, strict=True)
            header = validate_header(reader.fieldnames, allowed)

            with output_path.open(
                "x", encoding="utf-8", newline=""
            ) as target:
                output_created = True
                writer = csv.DictWriter(
                    target,
                    fieldnames=allowed,
                    extrasaction="raise",
                    lineterminator="\n",
                )
                writer.writeheader()

                for row in reader:
                    if None in row:
                        raise RejectedError(
                            "列数がヘッダーより多い行があります: "
                            f"物理行{reader.line_num}"
                        )

                    missing_values = [
                        name for name in header if row.get(name) is None
                    ]
                    if missing_values:
                        raise RejectedError(
                            "列数がヘッダーより少ない行があります: "
                            f"物理行{reader.line_num}"
                        )

                    writer.writerow(
                        {name: row[name] for name in allowed}
                    )
                    row_count += 1

                target.flush()
                os.fsync(target.fileno())

        completed = True

    except FileExistsError as exc:
        raise RejectedError(
            f"出力先が既に存在します: {output_path}"
        ) from exc
    except UnicodeDecodeError as exc:
        raise RejectedError(
            "入力CSVはUTF-8(BOM可)として読めません"
        ) from exc
    except csv.Error as exc:
        line_number = reader.line_num if reader is not None else 0
        raise RejectedError(
            f"CSV構文エラーです(物理行{line_number})"
        ) from exc
    except OSError as exc:
        target_name = exc.filename or "対象不明"
        raise OperationalError(
            f"ファイル操作に失敗しました: {target_name}"
        ) from exc
    finally:
        if output_created and not completed:
            cleanup_partial(output_path)

    return {
        "rows": row_count,
        "input_columns": len(header),
        "output_columns": allowed,
        "dropped_columns": len(header) - len(allowed),
        "output": str(output_path),
    }


def build_parser() -> argparse.ArgumentParser:
    parser = JsonArgumentParser(
        description="許可した列だけをCSVへ書き出します"
    )
    parser.add_argument("source", type=Path, help="入力CSV")
    parser.add_argument("policy", type=Path, help="ポリシーJSON")
    parser.add_argument("output", type=Path, help="新規出力CSV")
    return parser


def main(argv: list[str] | None = None) -> int:
    try:
        args = build_parser().parse_args(argv)
        report = export_allowed_csv(
            args.source, args.policy, args.output
        )
    except RejectedError as exc:
        emit({"status": "rejected", "reason": str(exc)})
        return EXIT_REJECTED
    except OperationalError as exc:
        emit({"status": "error", "reason": str(exc)})
        return EXIT_OPERATIONAL_ERROR
    except KeyboardInterrupt:
        emit({"status": "error", "reason": "処理が中断されました"})
        return EXIT_INTERRUPTED
    except Exception as exc:
        emit(
            {
                "status": "error",
                "reason": f"予期しないエラー: {type(exc).__name__}",
            }
        )
        return EXIT_OPERATIONAL_ERROR

    emit({"status": "ok", **report})
    return EXIT_OK


if __name__ == "__main__":
    sys.exit(main())

4. 実行結果を確認する

出力先には、まだ存在しないファイル名を指定します。

python3 export_allowed_csv.py source.csv export-policy.json ai_input.csv
rc=$?
echo "exit=$rc"

スクリプトが標準出力へ出すJSONは次の1行です。

{"status":"ok","rows":2,"input_columns":6,"output_columns":["case_id","category","request_text"],"dropped_columns":3,"output":"ai_input.csv"}

続いて、シェルの echo が exit=0 を表示します。生成された ai_input.csv は次の内容です。

case_id,category,request_text
C-001,配送,到着日を確認したい
C-002,請求,領収書を再発行したい

元の行順は維持されます。空文字のセルは許可しますが、列自体が不足している行は拒否します。データ行が0件でも、ヘッダーが正しければ正常終了します。

5. ポリシーの誤りを止められるか確認する

email を許可列と禁止列の両方へ入れた bad-policy.json を用意します。

{
  "allowed_columns": [
    "case_id",
    "email",
    "category",
    "request_text"
  ],
  "forbidden_columns": [
    "customer_name",
    "email",
    "phone",
    "address",
    "internal_note"
  ]
}

既存ファイルと衝突しない出力名で実行します。

python3 export_allowed_csv.py source.csv bad-policy.json bad.csv
rc=$?
echo "exit=$rc"
test ! -e bad.csv
echo "output_absent=$?"

標準出力と終了コードは次のようになります。output_absent=0 は、bad.csv が存在しないことを確認できたという意味です。

{"status":"rejected","reason":"allowed_columns と forbidden_columns が重複しています: email"}
exit=2
output_absent=0

運用へ組み込む順序

  1. 処理目的に必要な列を担当者とデータ管理者が決める
  2. 許可列と禁止列をポリシーJSONでレビューする
  3. 元CSVから別名のCSVを生成する
  4. 終了コードが0であることと、結果JSONの件数・列名を確認する
  5. 生成ファイルの内容を確認してから、承認済みの処理へ渡す
  6. 生成ファイルは組織の保管・削除ルールに従って扱う

後続処理は、ファイルの存在だけではなく終了コード0を開始条件にします。処理中も出力ファイルは見えるため、同じ出力先を監視する別処理がある場合は、完了前に読み込ませない設計が必要です。

この実装で保証しないこと

  • 許可した自由記述欄に含まれる氏名、電話番号、秘密情報の検出
  • case_idなど、別データと照合できる識別子の匿名化
  • AIサービスへの送信可否や法令・契約への適合判断
  • 表計算ソフトで開く場合の数式注入対策
  • タブ区切りやセミコロン区切りなど、カンマ区切り以外の形式
  • 同じ出力先へ複数プロセスが同時に書き込む運用
  • 強制終了や電源断のあとに不完全なファイルが残らないこと

通常の検証エラーやCtrl+Cでは、作成途中と認識できた出力を削除します。ただし、プロセスの強制終了や電源断では不完全なファイルが残る可能性があります。再実行前に出力ファイルを確認し、不要と判断できた場合だけ明示的に削除してください。

まとめ

AIへ渡すデータを減らすときは、「削除する列」を毎回選ぶより、「出してよい列」を固定する方が手順を再現しやすくなります。このゲートは完全な匿名化ツールではありません。それでも、許可列、ポリシーの矛盾、CSV構造、上書き防止、終了コードを検査し、列の削除漏れを機械的に止められます。

第160回 CSVを表計算ソフトへ渡す前に止める:Pythonで作る数式注入チェック

第159回では、CSVをUTF-8へ正規化する入口を作りました。しかし、文字コードが正しくても、表計算ソフトへ渡す前に確認したいことがあります。外部から受け取った文字列が特定の記号で始まると、文字列ではなく数式として解釈される可能性があるためです。

Pythonのcsvモジュールが数式を実行するわけではありません。問題になるのは、担当者や取引先がCSVを表計算ソフトで開く段階です。今回は、外部入力が入るテキスト列を検査し、数式候補があれば終了コード3で配布を止めるCLIを作ります。入力ファイルは変更しません。

CSVの引用符だけでは防げない

CSVの引用符は、カンマや改行を含む値を一つのセルとして扱うための構文です。たとえばCSV内で"=1+1"と書かれていても、読み込み後の値は=1+1です。引用符で囲むことと、数式としての解釈を止めることは別の処理です。

このスクリプトでは、先頭の半角スペースを除いた最初の文字が、次のいずれかなら数式候補として検出します。

  • 半角の=、+、-、@
  • タブ、CR、LF
  • 全角の=、+、-、@

全角文字は、日本語環境を考慮した保守的な検査です。実際の解釈は、表計算ソフト、バージョン、地域設定、読み込み方法によって変わります。この検査だけで、すべての環境における安全性を保証することはできません。

入出力の決まり

  • 入力はヘッダー付き、カンマ区切りのUTF-8 CSVとする。UTF-8 BOMの有無はどちらも受け付ける。
  • --text-columnで指定した列だけを検査する。
  • 外部入力が入るテキスト列は漏れなく指定する。
  • 数値列は事前のスキーマ検査で数値だけに制限する。負数を許可するためだけに、未検証の列を検査対象から外さない。
  • セルの元の値はJSONへ出さず、論理レコード番号、列名、検出文字だけを記録する。
  • 入力ファイルは変更せず、変換ファイルも作らない。

数式候補を自動的に書き換えないのは、先頭へのアポストロフィやタブの追加が、すべての表計算ソフトと後続処理で同じ意味になるとは限らないためです。検出時は配布を止め、元データまたはCSVを作る処理を修正します。

検査スクリプト

次のコードをcsv_formula_guard.pyとして保存します。Python 3.8以降の標準ライブラリだけを使います。

#!/usr/bin/env python3
import argparse
import csv
import json
import sys
from pathlib import Path


DANGEROUS_PREFIXES = frozenset("=+-@\t\r\n=+-@")
LEADING_SPACES = " "
MAX_DETAILS = 100


class InputError(Exception):
    pass


def formula_trigger(value):
    candidate = value.lstrip(LEADING_SPACES)
    if not candidate:
        return None

    first = candidate[0]
    if first in DANGEROUS_PREFIXES:
        return first
    return None


def validate_header(header):
    if not header:
        raise InputError("ヘッダーがありません")
    if any(not name.strip() for name in header):
        raise InputError("空の列名があります")
    if any(name != name.strip() for name in header):
        raise InputError("列名の前後に空白があります")
    if len(header) != len(set(header)):
        raise InputError("列名が重複しています")


def unique_columns(names):
    result = []
    for name in names:
        if name not in result:
            result.append(name)
    return result


def select_columns(header, names):
    missing = [name for name in names if name not in header]
    if missing:
        raise InputError("対象列がありません: " + ", ".join(missing))
    return [(header.index(name), name) for name in names]


def scan_csv(path, columns):
    finding_count = 0
    details = []

    try:
        with path.open("r", encoding="utf-8-sig", newline="") as source:
            reader = csv.reader(source, strict=True)
            try:
                header = next(reader)
            except StopIteration as exc:
                raise InputError("CSVが空です") from exc

            validate_header(header)
            selected = select_columns(header, columns)

            for record_number, row in enumerate(reader, start=2):
                if len(row) != len(header):
                    raise InputError(
                        f"レコード{record_number}: 列数が{len(row)}です"
                        f"(期待値: {len(header)})"
                    )

                for index, column_name in selected:
                    trigger = formula_trigger(row[index])
                    if trigger is None:
                        continue

                    finding_count += 1
                    if len(details) < MAX_DETAILS:
                        details.append({
                            "record": record_number,
                            "column": column_name,
                            "trigger": trigger,
                        })

    except UnicodeDecodeError as exc:
        raise InputError(f"{path}: UTF-8として読めません") from exc
    except csv.Error as exc:
        raise InputError(f"{path}: CSV構文エラー: {exc}") from exc

    return finding_count, details


def build_parser():
    parser = argparse.ArgumentParser(
        description="CSVの指定列にある数式候補を検査します"
    )
    parser.add_argument("input", type=Path)
    parser.add_argument(
        "--text-column",
        dest="text_columns",
        action="append",
        required=True,
        help="検査する列名。複数回指定できます",
    )
    return parser


def print_report(report):
    print(json.dumps(
        report,
        ensure_ascii=False,
        separators=(",", ":"),
    ))


def main():
    args = build_parser().parse_args()
    columns = unique_columns(args.text_columns)

    try:
        finding_count, details = scan_csv(args.input, columns)
    except (InputError, OSError) as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1

    report = {
        "status": "blocked" if finding_count else "ok",
        "input": str(args.input),
        "checked_columns": columns,
        "finding_count": finding_count,
        "findings": details,
        "findings_truncated": finding_count > len(details),
    }
    print_report(report)
    return 3 if finding_count else 0


if __name__ == "__main__":
    sys.exit(main())

全レコードをリストへ保存せず、1レコードずつ検査します。詳細は最大100件に制限しますが、finding_countは最後まで数えます。

検出結果を再現する

次の内容をUTF-8のsample.csvとして保存します。amount列は、別工程で数値として検証済みという前提です。

id,customer_name,memo,amount
1,株式会社A,折り返し連絡,1200
2,=1+1,通常,900
3,株式会社C," +SUM(1,2)",-500

外部入力が入るcustomer_name列とmemo列を検査します。

python3 csv_formula_guard.py sample.csv --text-column customer_name --text-column memo

標準出力は次の1行です。

{"status":"blocked","input":"sample.csv","checked_columns":["customer_name","memo"],"finding_count":2,"findings":[{"record":3,"column":"customer_name","trigger":"="},{"record":4,"column":"memo","trigger":"+"}],"findings_truncated":false}

終了コードは3です。recordは物理的な行番号ではなく、ヘッダーを1番としたCSV上の論理レコード番号です。セル内に改行があっても、一つのレコードとして数えます。

amount列の-500は検査対象ではないため、数式候補には数えません。ただし、数値列へ文字列を混入できる状態であれば、この運用は安全ではありません。先に整数や小数として解析できることを検査してください。

終了コードを後続処理へ渡す

終了コード 意味 後続処理
0 数式候補がない 次の工程へ進める
1 ファイル、文字コード、CSV構文、ヘッダーまたは列数のエラー 入力を修正して再実行する
2 コマンド引数の誤り ファイル名と列指定を確認する
3 数式候補を検出した 配布を止め、元データを確認する

入力エラーは標準エラー出力へ書き、JSONは出力しません。後続処理ではJSONの文言だけで判断せず、終了コードを確認します。

業務フローへ入れる順番

  1. 第159回の入口で文字コードをUTF-8へそろえる。
  2. 固定ヘッダー、列数、必須項目、数値型などをスキーマ検査する。
  3. 外部入力が入るテキスト列をすべて指定して、このスクリプトを実行する。
  4. 終了コード3なら配布を止め、該当レコードの元データを修正する。
  5. 修正後に同じ条件で再実行する。
  6. 終了コード0を確認したうえで、実際の受け渡し先と同じ方法でも読み込み結果を確認する。

この検査の制約

  • 対象は指定した列だけです。列指定の漏れは検出できません。
  • ヘッダー自体は数式候補として検査しません。外部入力から列名を作らず、固定スキーマと照合してください。
  • CSVの区切り文字はカンマ、引用符はダブルクォートというcsvモジュールの既定設定です。
  • strict=Trueでも、あらゆる不正形式を判定できるわけではありません。列数と固定スキーマの検査を併用します。
  • 検出文字の集合は保守的な規則です。表計算ソフト固有の挙動や将来の変更までは判定しません。
  • 終了コード0は、この規則に一致する値がなかったことを示すだけです。受け渡し先での安全性を証明するものではありません。

最初に行うのは、顧客名、備考、問い合わせ内容など、外部入力が入る列の洗い出しです。検出後に自動変換するのではなく、終了コード3で配布を止めるところまで業務手順に組み込みます。

確認資料

第159回 CSVの文字コードを推測しない:Pythonで作るUTF-8正規化ゲート

第158回では、CSVの列構成をv1からv2へ移行する処理を扱いました。しかし、その前段で文字コードを正しく扱えなければ、移行処理には進めません。

今回は、受け取ったCSVをバイト列で読み、まずUTF-8として厳密に検査します。UTF-8でなかった場合は、明示的に許可したときだけCP932として読みます。文字コードを自動推測せず、受け入れる形式を業務ルールとして限定する設計です。

今回の入出力契約

項目 仕様
既定の入力 BOMなしUTF-8、または先頭にUTF-8 BOMが1個だけあるUTF-8
任意の入力 --allow-cp932を指定した場合のみCP932
対象外 UTF-16、UTF-32、およびUTF-8 BOMが2個以上連続する入力
出力 BOMなしUTF-8
CSV条件 1行目に少なくとも1つ値のあるヘッダーがあり、すべてのレコードがヘッダーと同じ列数であること
上書き 既存ファイルへの上書きは禁止
変換成功時 標準出力へ1行のJSON、終了コード0
スクリプト内で検出したエラー 標準エラー出力へ1行のJSON、終了コード2~5

CP932は自動判定するのではなく、UTF-8でなかった場合の代替候補です。入力元がCP932を出力すると確認できている業務だけで有効にします。

UTF-8 BOMは先頭に1個だけ許可します。複数のBOMを成功扱いにすると、utf-8-sigで除去されなかったBOMが出力へ残るため、この実装では文字コードエラーとして拒否します。

処理の順序

  1. 入力を文字列ではなくバイト列として読む
  2. UTF-16またはUTF-32のBOMがあれば対象外として停止する
  3. UTF-8 BOMが2個以上連続していれば停止する
  4. 単一のUTF-8 BOM、BOMなしUTF-8、許可時のCP932の順で厳密にデコードする
  5. CSVとして解析し、ヘッダーと列数を確認する
  6. BOMなしUTF-8へ変換し、新しいファイルとして保存する
  7. 入力文字コード、列数、データ行数をJSONで報告する

CSVの検査にはcsv.readerを使いますが、出力をcsv.writerでは再生成しません。引用符や改行コードを組み直さず、デコードした文字列をUTF-8へ変換します。単一のBOMが付いたUTF-8入力では、そのBOMだけを除いて内容を維持します。

標準ライブラリだけで実装する

次のコードをnormalize_csv_encoding.pyとして保存します。型注釈にはPython 3.9以降の記法を使用しています。

import argparse
import codecs
import csv
import io
import json
import sys
from pathlib import Path

EXIT_OK = 0
EXIT_INPUT = 2
EXIT_ENCODING = 3
EXIT_CSV = 4
EXIT_OUTPUT = 5


class EncodingRejected(Exception):
    pass


class CsvRejected(Exception):
    pass


class OutputRejected(Exception):
    pass


def decode_input(data: bytes, allow_cp932: bool) -> tuple[str, str]:
    unsupported_boms = (
        codecs.BOM_UTF32_LE,
        codecs.BOM_UTF32_BE,
        codecs.BOM_UTF16_LE,
        codecs.BOM_UTF16_BE,
    )
    if data.startswith(unsupported_boms):
        raise EncodingRejected(
            'UTF-16/32は対象外です。送信元でUTF-8またはCP932として書き出してください'
        )

    if data.startswith(codecs.BOM_UTF8 * 2):
        raise EncodingRejected('UTF-8 BOMが複数連続しているため受け入れません')

    if data.startswith(codecs.BOM_UTF8):
        try:
            return data.decode('utf-8-sig', errors='strict'), 'utf-8-sig'
        except UnicodeDecodeError as exc:
            raise EncodingRejected(
                'UTF-8 BOMがありますが、本文をUTF-8としてデコードできません'
            ) from exc

    try:
        return data.decode('utf-8', errors='strict'), 'utf-8'
    except UnicodeDecodeError as utf8_error:
        if not allow_cp932:
            raise EncodingRejected(
                '入力はUTF-8ではありません。CP932を受け入れる場合だけ '
                '--allow-cp932 を指定してください'
            ) from utf8_error

    try:
        return data.decode('cp932', errors='strict'), 'cp932'
    except UnicodeDecodeError as cp932_error:
        raise EncodingRejected(
            '入力をUTF-8またはCP932として厳密にデコードできません'
        ) from cp932_error


def inspect_csv(text: str) -> tuple[int, int]:
    if '\x00' in text:
        raise CsvRejected('NUL文字を含むためCSVとして受け入れません')

    reader = csv.reader(io.StringIO(text, newline=''), strict=True)

    try:
        header = next(reader)
    except StopIteration as exc:
        raise CsvRejected('CSVが空です') from exc
    except csv.Error as exc:
        raise CsvRejected('ヘッダーをCSVとして解析できません') from exc

    if not header or all(cell == '' for cell in header):
        raise CsvRejected('有効なヘッダーがありません')

    expected_columns = len(header)
    data_rows = 0

    try:
        for record_number, row in enumerate(reader, start=2):
            if len(row) != expected_columns:
                raise CsvRejected(
                    f'レコード{record_number}の列数は{len(row)}です。'
                    f'ヘッダーの{expected_columns}列と一致しません'
                )
            data_rows += 1
    except csv.Error as exc:
        raise CsvRejected(
            f'CSV構文を解析できません(物理行{reader.line_num}付近)'
        ) from exc

    return expected_columns, data_rows


def write_new_file(path: Path, payload: bytes) -> None:
    created = False

    try:
        handle = path.open('xb')
        created = True
        with handle:
            written = handle.write(payload)
            if written != len(payload):
                raise OSError('short write')
    except FileExistsError as exc:
        raise OutputRejected(f'出力先が既に存在します: {path}') from exc
    except OSError as exc:
        if created:
            try:
                path.unlink()
            except OSError:
                pass
        raise OutputRejected(f'出力先へ書き込めません: {path}') from exc


def emit_error(code: int, message: str) -> int:
    report = {'status': 'error', 'message': message}
    print(
        json.dumps(report, ensure_ascii=False, separators=(',', ':')),
        file=sys.stderr,
    )
    return code


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description='CSVを検査し、BOMなしUTF-8へ変換します'
    )
    parser.add_argument('input', type=Path, help='入力CSV')
    parser.add_argument('output', type=Path, help='新規作成する出力CSV')
    parser.add_argument(
        '--allow-cp932',
        action='store_true',
        help='UTF-8でない入力をCP932として試します',
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()

    try:
        data = args.input.read_bytes()
    except OSError:
        return emit_error(EXIT_INPUT, f'入力を読み込めません: {args.input}')

    try:
        text, source_encoding = decode_input(data, args.allow_cp932)
    except EncodingRejected as exc:
        return emit_error(EXIT_ENCODING, str(exc))

    try:
        columns, data_rows = inspect_csv(text)
    except CsvRejected as exc:
        return emit_error(EXIT_CSV, str(exc))

    try:
        write_new_file(args.output, text.encode('utf-8'))
    except OutputRejected as exc:
        return emit_error(EXIT_OUTPUT, str(exc))

    report = {
        'status': 'ok',
        'source_encoding': source_encoding,
        'output_encoding': 'utf-8',
        'columns': columns,
        'data_rows': data_rows,
        'input': str(args.input),
        'output': str(args.output),
    }
    print(json.dumps(report, ensure_ascii=False, separators=(',', ':')))
    return EXIT_OK


if __name__ == '__main__':
    raise SystemExit(main())

再現手順

1.CP932の入力例を作る

customers_utf8.csvが存在しない作業ディレクトリで、次のコマンドを実行します。

python -c "from pathlib import Path; Path('customers_cp932.csv').write_bytes('顧客ID,氏名\r\nC001,山田太郎\r\n'.encode('cp932'))"

2.許可せずに実行する

python normalize_csv_encoding.py customers_cp932.csv customers_utf8.csv
echo $?

CP932を許可していないため、標準エラー出力と終了コードは次のようになります。出力CSVは作られません。

{"status":"error","message":"入力はUTF-8ではありません。CP932を受け入れる場合だけ --allow-cp932 を指定してください"}
3

3.入力元を確認したうえでCP932を許可する

python normalize_csv_encoding.py customers_cp932.csv customers_utf8.csv --allow-cp932
echo $?

正常時は標準出力へ1行のJSONが出ます。

{"status":"ok","source_encoding":"cp932","output_encoding":"utf-8","columns":2,"data_rows":1,"input":"customers_cp932.csv","output":"customers_utf8.csv"}
0

出力にBOMがなく、UTF-8として読めることも確認します。

python -c "from pathlib import Path; b=Path('customers_utf8.csv').read_bytes(); print(b.startswith(bytes((239, 187, 191))), b.decode('utf-8').splitlines()[0])"
False 顧客ID,氏名

4.UTF-8 BOMが複数連続する入力を拒否する

codecs.BOM_UTF8を2個付けたCSVを作り、未使用の出力先を指定します。

python -c "from pathlib import Path; import codecs; Path('double_bom.csv').write_bytes(codecs.BOM_UTF8 * 2 + b'a,b\n1,2\n')"
python normalize_csv_encoding.py double_bom.csv double_bom_output.csv
echo $?

この入力は文字コードエラーになり、出力CSVは作られません。

{"status":"error","message":"UTF-8 BOMが複数連続しているため受け入れません"}
3

5.列数が異なるCSVを拒否する

python -c "from pathlib import Path; Path('broken.csv').write_bytes(b'id,name\n1\n')"
python normalize_csv_encoding.py broken.csv unused.csv
echo $?
{"status":"error","message":"レコード2の列数は1です。ヘッダーの2列と一致しません"}
4

検査に失敗したため、unused.csvは作られません。

6.既存の出力ファイルを上書きしないことを確認する

手順3で作成したcustomers_utf8.csvを出力先にして、同じ処理をもう一度実行します。

python normalize_csv_encoding.py customers_cp932.csv customers_utf8.csv --allow-cp932
echo $?
{"status":"error","message":"出力先が既に存在します: customers_utf8.csv"}
5

xbモードで新規作成しているため、既存のcustomers_utf8.csvは変更されません。

終了コードを後続処理の条件にする

終了コード 意味 対応
0 正規化成功 出力CSVを次工程へ渡す
2 入力を読めない、または引数エラー パス、権限、引数を確認する
3 文字コードを受け入れられない 送信元の出力設定とBOMを確認する
4 CSV構文または列数が不正 ファイルを隔離して内容を確認する
5 出力できない 既存ファイル、親ディレクトリ、権限を確認する

入力ファイルの読み込みに失敗した場合は、終了コード2と1行のJSONを返します。一方、引数不足や未知のオプションをargparseが処理開始前に検出した場合は、同じ終了コード2でもJSONではなく使用法が標準エラー出力へ表示されます。--helpは使用法を表示し、終了コード0で終了します。

この実装で意図的に行わないこと

  • EUC-JPなど、契約にない文字コードの推測
  • UTF-16またはUTF-32からの自動変換
  • 複数のUTF-8 BOMを除去して処理を継続すること
  • 改行コード、区切り文字、引用符の正規化
  • 日付、金額、必須項目など業務スキーマの検証
  • 既存ファイルの上書き
  • 大容量CSVのストリーミング処理

このコードは入力全体をメモリへ読み込むため、大容量CSVには向きません。まず小さな業務ファイルの入口処理として使い、容量が大きい場合は上限を決めたうえで別の実装を検討します。

CP932としてデコードできることは、入力元が本当にCP932を採用している証明にはなりません。別の文字コードのバイト列でも偶然デコードできる可能性があるため、--allow-cp932は入力元単位で設定します。ASCII文字だけの入力はUTF-8と同じバイト表現なので、UTF-8として扱われます。

書き込み中に通常のエラーが発生した場合、コードは作成途中のファイルの削除を試みます。ただし、削除自体に失敗する場合や、プロセスの強制終了、OS障害まで含む原子的な保存は保証しません。終了コード0を確認したファイルだけを次工程へ渡します。

実務へ組み込むときの確認項目

  • 入力元ごとに、許可する文字コードを記録する
  • 下流処理には原本ではなく正規化後のパスを渡す
  • 終了コード0のときだけスキーマ移行や集計を開始する
  • 標準出力と標準エラー出力を分けて保存する
  • 原本を保管する場合は、保存期間とアクセス権を業務ルールで決める

文字コードの問題を入口で分離すれば、後段のスキーマ移行や品質検査はUTF-8だけを前提にできます。対応形式をむやみに増やさず、受け入れ条件を狭く明示することが、再現可能なCSV処理につながります。