CSVのSQLite取込が途中で失敗したら戻す方法

CSVの取り込みが途中で失敗すると、それまでの行だけがDBへ残る構成もあります。利用者が同じCSVを再投入したときに、何件をやり直すべきか分からなくなる原因になります。

全件が正しい場合だけ登録したいなら、対象の書き込みを一つのトランザクションへまとめます。トランザクションは複数の変更をまとめて確定したり、失敗時に戻したりする単位です。

二件目で失敗させて件数を確認

次は架空の金額を登録するメモリーDBです。例の条件として金額は0以上にします。二件目の-1が拒否されたら、一件目の100も含めて戻し、ゼロ件になることを確認します。

import sqlite3

db = sqlite3.connect(":memory:", autocommit=False)
try:
    db.execute(
        "CREATE TABLE amounts("
        "id TEXT PRIMARY KEY NOT NULL,"
        "amount INTEGER NOT NULL CHECK(amount >= 0))"
    )
    db.commit()
    try:
        with db:
            db.executemany(
                "INSERT INTO amounts VALUES (?,?)",
                [("A", 100), ("B", -1)],
            )
    except sqlite3.IntegrityError:
        print("途中の不正値で全件を巻戻し")

    failed_count = db.execute("SELECT count(*) FROM amounts").fetchone()[0]
    print("失敗後の件数:", failed_count)
    assert failed_count == 0
    with db:
        db.executemany(
            "INSERT INTO amounts VALUES (?,?)",
            [("A", 100), ("B", 50)],
        )
    count = db.execute("SELECT count(*) FROM amounts").fetchone()[0]
    print("再取込後の件数:", count)
    assert count == 2
finally:
    db.close()

失敗後は0件、正常な二件を入れ直した後は2件です。Windows・Python 3.12.10・SQLite 3.49.1で、この全件の巻き戻しを確認しました。autocommit引数はPython 3.12以降の機能です。

負数を拒否するのは失敗を起こすための架空の条件です。返金や訂正を負数で表す業務へ、この制約をそのまま適用する意味ではありません。

例外はトランザクションの外で受ける

上のtry/exceptはwith dbの外側にあります。例外をwithの中で受けて通常終了させると、失敗が外へ伝わらず、その前の変更を確定させる構成になる場合があります。

一行ごとにcommitを呼ぶ構成も、全件を一単位で戻す例とは違います。途中まで成功させる要件なら、成功分と失敗分、再実行の対象を記録する設計が必要です。どちらの方針にするかを入力件数だけで決めず、業務上の処理単位から考えます。

DBの外の処理は戻らない

トランザクションで戻せるのは、そのDBの変更です。取り込みの途中で送ったメールや、別のファイルへ保存した内容までは、このrollbackで元に戻りません。

外部への通知は確定後に分ける、送信する対象を記録して再送を管理するなど、DBの外へ出る処理も合わせて設計します。まずは保存だけの小さな例で、途中失敗後の件数が期待どおりかを確認してください。

実ファイルを入力にする場合は、読込形式や必須項目を先に検査し、長時間の書き込みを始める前に不備を見つける形にもできます。検査と保存時の制約を分けると、利用者へ直す理由を返しつつ、別経路からの不備も確認できます。

DBの基本はSQLite入門、例外の扱いはPythonの例外処理で補えます。

参考資料

Python sqlite3とSQLite CREATE TABLEの公式資料を2026年11月1日に確認しました。