再送を「もう一件の申請」にしない
モバイルで配送申請を送った直後に通信が切れると、利用者には登録できたか分かりません。ボタンを無効にするだけでは、再読み込みや通信の再試行を止められません。サーバー側で「同じ操作の再送」を区別する必要があります。
筆者は韓国向け配送代行サービスKimfastの関係者です。この記事は配送申請という題材を使った独立した学習用コードです。Kimfastの本番実装、障害事例、性能測定を公開するものではありません。顧客データも使いません。
先に決める3つの動作
| 入力 | 返す結果 |
|---|---|
| まだ保存されていないキー | 新しい申請ID |
| 同じアカウント・同じキー・同じ内容 | 前回と同じ申請ID |
| 同じアカウント・同じキー・違う内容 | 競合エラー |
キーは送信を押すたびではなく、一つの申請操作について生成し、結果が確定するまで再利用します。新しい申請には新しいキーを使います。アカウントIDは認証済みセッションからサーバーが決めます。クライアントに自由に指定させません。
最小構成：SQLiteに申請とキーを同時に保存する
次を idempotency_demo.py として保存します。標準ライブラリだけで動きます。例では保存された行そのものを申請とみなし、外部APIや決済は呼びません。
"""Educational example only; no connection to Kimfast production."""
import hashlib
import json
import sqlite3
import uuid
from contextlib import closing
def initialize(path):
with closing(sqlite3.connect(path)) as db:
db.execute("""CREATE TABLE IF NOT EXISTS requests (
account TEXT NOT NULL, request_key TEXT NOT NULL,
payload_hash TEXT NOT NULL, application_id TEXT NOT NULL,
PRIMARY KEY (account, request_key))""")
db.commit()
def submit(path, account, request_key, payload):
if not account or not request_key:
raise ValueError("account and request_key are required")
canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"),
ensure_ascii=False, allow_nan=False)
digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
with closing(sqlite3.connect(path, isolation_level=None, timeout=5)) as db:
db.execute("BEGIN IMMEDIATE")
try:
existing = db.execute(
"SELECT payload_hash, application_id FROM requests "
"WHERE account=? AND request_key=?",
(account, request_key)).fetchone()
if existing:
if existing[0] != digest:
raise ValueError("same key used with a different payload")
application_id, created = existing[1], False
else:
application_id, created = str(uuid.uuid4()), True
db.execute("INSERT INTO requests VALUES (?, ?, ?, ?)",
(account, request_key, digest, application_id))
db.commit()
return application_id, created
except Exception:
db.rollback()
raise
PRIMARY KEY (account, request_key) が一意性の境界です。BEGIN IMMEDIATE で書き込みトランザクションを開始し、確認と作成を同じトランザクションに入れます。この短い処理の中でネットワーク通信はしません。
JSONのキー順はそろえますが、これは業務上の同値性を完全に判定する仕組みではありません。たとえば 1 と 1.0、配列の順序は別扱いです。実サービスでは、入力の型・単位・必須項目を検証し、正規化した申請データを渡します。
まず再送と競合を試す
from idempotency_demo import initialize, submit
initialize("demo.db")
payload = {"origin": "AU", "destination": "KR", "parcels": 1}
first = submit("demo.db", "demo-account", "action-001", payload)
retry = submit("demo.db", "demo-account", "action-001", payload)
assert first[0] == retry[0]
assert first[1] is True and retry[1] is False
try:
submit("demo.db", "demo-account", "action-001", {**payload, "parcels": 2})
except ValueError as error:
print(error) # same key used with a different payload
同じデータでも別のキーなら別の申請を作ります。内容が似ているだけで勝手に統合すると、利用者が意図した2回目の注文を消してしまうためです。
同時再送も確認する
テスト用の新しいDBで、別々の接続から同じ操作を8回呼びます。
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from tempfile import TemporaryDirectory
import sqlite3
with TemporaryDirectory() as tmp:
path = str(Path(tmp) / "concurrency.db")
initialize(path)
with ThreadPoolExecutor(max_workers=4) as pool:
results = list(pool.map(
lambda _: submit(path, "demo-account", "shared", payload),
range(8),
))
assert len({result[0] for result in results}) == 1
assert sum(result[1] for result in results) == 1
with sqlite3.connect(path) as db:
assert db.execute("SELECT COUNT(*) FROM requests").fetchone()[0] == 1
2026年9月28日のローカル実行では、再送、JSONキー順、内容変更、アカウント分離、新規キー、空キー・NaN、同時再送の7つのunittestが通過しました。これは性能や本番可用性の保証ではありません。
本番に持ち込む前に残ること
- 認証・認可、CSRF対策、入力スキーマ、キー長とリクエストサイズの上限。
- キーの保存期間と再利用ルール。期限後に再送された場合の扱い。
- DBロックやタイムアウトを、成功・削除・未登録と即断しないこと。結果照会の導線を用意します。
- 決済や送り状APIなど外部処理との整合性。DBの冪等性だけでは外部処理の二重実行を防げません。必要に応じてoutbox、外部サービス側の冪等キー、照合処理を別に設計します。
- 申請IDの保存と実際の申請作成が別テーブルなら、両方を同じトランザクションで扱います。
UIでも「申請を受け付けた」と「発送が完了した」を分けて表示することが大切です。配送代行には申請、倉庫での確認、料金確認など異なる段階があります。題材の背景はKimfastのオーストラリア配送案内で確認できます。これは韓国向けサービスの案内であり、日本向け配送を提供するという意味ではありません。
参考資料・制作について
- Python公式 sqlite3 ドキュメント：接続、トランザクション、パラメータを使ったSQL。
- 図は本記事の処理と実行結果を説明するために作成したものです。実際の倉庫や顧客の写真ではありません。
- 本文と例の作成にAIを使用しました。コードは上記の範囲でローカル実行しました。