結論：関連する更新をまとめ、成功ならcommit、失敗ならrollbackする
複数の更新が一組になっている場合、途中までだけ保存されるとデータの意味が壊れる。全て成功したらcommitで確定し、一つでも失敗したらrollbackでその組の変更を取り消す。各UPDATEの直後にcommitすると、その後の失敗では前の確定を戻せない。どの操作を同じトランザクションへ入れるかを、実際の業務上のまとまりに合わせて決めよう。
例は試料の保管場所AからBへ数量を移す小さな記録である。Aを減らしたのにBを増やせなければ合計が変わってしまうため、二つの更新を一まとまりにする。実際の在庫や本番データは操作せず、一時データベースで成功と二種類の失敗を確認する。トランザクションが守る範囲を、目に見える少数の値で確かめるのが目的である。
そのまま動かせる例
標準ライブラリだけで実行できるが、sqlite3.connect(..., autocommit=False)を使うためPython 3.12以降が必要である。最初にAが10、Bが0の表を作り、初期状態をcommitしてから移動を始める。成功する3個の移動の後、Bを負数へする制約違反と、主キーを重複させる制約違反を意図的に起こす。
前半のmoveはtry・exceptで明示的なcommitとrollbackを使い、後半はwith conによる自動的な確定・取消しを示す。どちらも失敗が例外として外へ出ることを確認する。外側のclosingで接続を閉じ、さらに再接続して確定済みの状態が残ることを調べる。一時フォルダーのファイルは終了時に削除される。
from contextlib import closing
from pathlib import Path
from tempfile import TemporaryDirectory
import sqlite3
def state(con):
return con.execute("SELECT name, quantity FROM stock ORDER BY name").fetchall()
def move(con, amount, invalid=False):
try:
con.execute("UPDATE stock SET quantity = quantity - ? WHERE name = ?", (amount, "A"))
if invalid:
con.execute("UPDATE stock SET quantity = ? WHERE name = ?", (-1, "B"))
else:
con.execute("UPDATE stock SET quantity = quantity + ? WHERE name = ?", (amount, "B"))
con.commit()
except Exception:
con.rollback()
raise
with TemporaryDirectory() as folder:
path = Path(folder) / "stock.sqlite3"
with closing(sqlite3.connect(path, autocommit=False)) as con:
con.execute("CREATE TABLE stock (name TEXT PRIMARY KEY, quantity INTEGER NOT NULL CHECK(quantity >= 0))")
con.executemany("INSERT INTO stock VALUES (?, ?)", [("A", 10), ("B", 0)])
con.commit()
move(con, 3)
committed = [("A", 7), ("B", 3)]
assert state(con) == committed
print("committed:", state(con))
try:
move(con, 2, invalid=True)
except sqlite3.IntegrityError:
print("manual rollback: IntegrityError")
else:
raise AssertionError("expected constraint failure")
assert state(con) == committed
try:
with con:
con.execute("UPDATE stock SET quantity = quantity - 1 WHERE name = ?", ("A",))
con.execute("INSERT INTO stock VALUES (?, ?)", ("B", 99))
except sqlite3.IntegrityError:
print("context rollback: IntegrityError")
else:
raise AssertionError("expected duplicate-key failure")
assert state(con) == committed
print("after both failures:", state(con))
with closing(sqlite3.connect(path, autocommit=False)) as con:
assert state(con) == committed
print("reconnect:", state(con))
実行結果
committed: [('A', 7), ('B', 3)]
manual rollback: IntegrityError
context rollback: IntegrityError
after both failures: [('A', 7), ('B', 3)]
reconnect: [('A', 7), ('B', 3)]
失敗した文より前の変更も取り消す
成功後の状態はAが7、Bが3である。次のmoveではまずAを2減らし、その後Bへ−1を入れようとしてCHECK制約に違反する。rollbackにより最初のAの減少も取り消され、7と3に戻る。失敗したSQLだけを修正すればよいのではなく、関連する変更のまとまりを元へ戻せることが重要である。
commit自体もtryの内側に置く。遅延制約などで確定時に失敗する場合も、rollbackの対象にするためである。exceptではrollbackした後にraiseで元の例外を伝える。取消しが済んだからと例外を消すと、呼び出し側が成功と誤認することがある。ここでは外側がIntegrityErrorを捕まえ、期待した失敗として表示する。予測していない問題を全て「在庫不足」などへ置き換えず、何が失敗したかを調査できるようにしておきたい。
with conは接続を閉じるものではない
接続のコンテキストマネージャーは、ブロックを正常に抜けたときのcommit、例外時のrollbackを扱う。ファイルを開くwithと似て見えるが、接続を閉じる操作とは別である。例ではclosingを外側に置き、二つの役割を分離している。また、with conを入れ子に書いたら自動的に独立したネスト済みトランザクションになる、という意味でもない。
autocommit=Falseでは、commitやrollbackの後も新しいトランザクションが開始される方式となる。autocommit=TrueではPython側のcommit・rollbackによる同じ制御は期待できない。従来の既定動作であるLEGACY_TRANSACTION_CONTROLとisolation_levelの説明も、今回の明示した設定とは区別しよう。実行するPython版と接続時の設定を記録することが大切である。
外部の操作まで戻せるわけではない
データベースのrollbackは、同じトランザクション内のDB変更を取り消す仕組みである。途中で送ったメール、外部APIの更新、作成済みのファイルまで自動的に元へ戻すことはできない。そうした処理を混ぜるときは、DB確定後に行う、再実行しても重複しないよう設計するなど、別の整合性管理が必要になる。
長いトランザクションは他の処理を待たせる原因になるため、通信や利用者の入力待ちを不用意に抱え込まないようにする。今回のmoveは既知のAとBを前提としており、実システムでは対象行が存在するか、更新件数が期待通りか、数量が妥当かなども検証が必要である。トランザクションだけで業務ルールが全て保証されるわけではないので、入力検証とDB制約を合わせて使おう。
確認環境と参考資料
例はLinux・CPython 3.12.14で実行した。掲載した出力はこの環境での結果である。公式資料のstable版や最新版は更新されるため、手元のバージョンと対応する仕様も確認してほしい。
- Python公式：トランザクション制御（2026年10月2日参照）
- Python公式：接続のコンテキストマネージャー（2026年10月2日参照）
関連項目：SQLiteにデータを保存・検索する：SQLはプレースホルダーで渡す ／ APIをむやみに再試行しない：回数制限・待ち時間・冪等性