最初は全件取っても問題ありません。1000件なら数秒で終わります。ところが件数が増えると、取得に何時間もかかるようになり、途中で失敗する確率も上がります。の提供側から見ても、毎日全件を要求してくる相手は歓迎されません。
全件取得をやめる判断は、件数そのものより所要時間で決めるのが実際的です。数分で終わるうちは全件のほうが単純で安全です。1時間を超えたあたりから、失敗したときのやり直しが重くなり、差分に切り替える理由が出てきます。
そこで前回から変わったぶんだけ取ることになります。素直に「前回実行した時刻より後に更新されたもの」を要求すればよさそうですが、この素直な実装は取りこぼします。この回では、どこで取りこぼすのかと、どう境界を決めるかを扱います。
素直な実装が取りこぼす理由
前回の実行時刻を覚えておいて、次はそれ以降の更新分を取る。この方式には、データが記録される時刻と、こちらが取りに行く時刻がずれるという前提が抜けています。
たとえば 10:00 に取得を開始し、10:00:30 に完了したとします。次回は「10:00 以降」を要求します。ところが 10:00:15 に登録されたデータが、処理の都合で 10:00:40 になってから検索対象に入った場合、そのデータは今回も次回も取れません。今回の取得時点ではまだ見えず、次回は 10:00 以降なので範囲には入るはずですが、開始時刻を 10:00:30 に更新していると漏れます。
取得にかかった時間のぶんだけ、取りこぼしの窓が開きます。次回の起点にするなら開始時刻であり、さらに安全側に少し戻すのが基本です。
境界を安全側に倒す
確実にするには、取る範囲を少し重ねます。前回の開始時刻からさらに数分〜数時間戻したところを起点にすれば、遅れて見えるようになったデータも拾えます。この「どれだけ遅れを許容するか」の線が です。
重ねて取ると同じデータを2回取ることになりますが、保存側がを持っていれば問題になりません。鍵を決めて上書きする形にしておけば、何度取っても結果は同じです。
from datetime import datetime, timedelta, timezone
# 遅れて見えるようになるデータをどれだけ許容するか。# 配信元の性質で決める。分からなければ広めに取る。LATENESS = timedelta(hours=6)
def next_window(last_started_at: datetime | None) -> tuple[datetime, datetime]: """(取得開始, 取得終了) を返す。終了は「今」ではなく開始時刻で固定する。""" now = datetime.now(timezone.utc) if last_started_at is None: start = now - timedelta(days=365) # 初回は広く else: start = last_started_at - LATENESS # 前回の開始から巻き戻す return start, now更新日時が信用できないことがある
そもそも配信元の更新日時が当てにならない場合があります。内容を直しても更新日時を変えない実装や、一括で全件の更新日時が書き換わる実装が実在します。前者だと差分取得が変更を見逃し、後者だと差分のはずが全件になります。
更新日時が使えないときは、内容そのものを比べるしかありません。項目を並べて要約値を作り、前回のものと違えば変更あり、と判定します。全件を取る必要は残りますが、書き込みは変わったぶんだけに減らせます。下流の処理が「変わった行だけ」を対象にできるので、全体としては十分に効きます。
要約値を作るときは、比較に使わない項目を必ず除いてください。取得時刻をそのまま含めると、中身が同じでも毎回違う値になり、全件が「変更あり」と判定されます。この取り違えは実装した本人でも見落としやすく、しかも「差分取得が効いていない」という形でしか表に出ません。
配信元が を返すなら、要約値を自分で作らずに済みます。前回の版の印を添えて要求し、304 が返れば内容は変わっていないと判断できます。相手の実装に依存しますが、使えるときは最も安くつきます。
import hashlibimport json
def content_hash(record: dict, ignore: set[str] = frozenset()) -> str: """比較に使わない項目（取得時刻など）を除いてから要約値を作る。 ここを除かないと、毎回「変わった」と判定されてしまう。""" target = {k: v for k, v in sorted(record.items()) if k not in ignore} blob = json.dumps(target, ensure_ascii=False, sort_keys=True) return hashlib.sha256(blob.encode("utf-8")).hexdigest()
def changed_records(new: list[dict], known: dict[str, str], key: str = "id"): """要約値が前回と違うものだけを返す。""" for rec in new: h = content_hash(rec, ignore={"fetched_at"}) if known.get(rec[key]) != h: yield rec, h削除は差分では取れない
差分取得には構造的な穴があります。消えたデータは差分に出てきません。更新日時で絞る方式は「変わったもの」を返すだけで、「無くなったもの」は返しようがないからです。
配信元が削除フラグを持っていればそれを使えますが、無い場合は定期的に全件を取って突き合わせるしかありません。毎日は差分、月に一度は全件、という組み合わせが現実的な落としどころです。この全件取得はの仕組みをそのまま使えます。
| 方式 | 必要なもの | 取りこぼす条件 |
|---|---|---|
| 更新日時で絞る | 信頼できる更新日時 | 遅れて見える／日時を更新しない実装 |
| カーソルの続きから | 追記のみのデータ | 過去分が書き換わる場合 |
| 内容の要約値で比較 | 全件の取得 | 取りこぼさないが、毎回全件を読む |
| 削除フラグ | 配信元の対応 | 対応していなければ使えない |
どの方式でも、定期的な全件との突き合わせは必要だと考えておいてください。差分だけで長期間回し続けると、ずれが静かに蓄積します。次回は、取得が途中で落ちたときに続きから再開する方法を扱います。
この記事の感想を教えてください
あなたの 1 クリックで、本当にこの記事は更新されます。「もっと詳しく」「続編希望」が一定数集まった記事は、 ふくふくが 実際に内容を拡充したり続編記事を公開 します。 送信したリアクションはお使いのブラウザに記録され、再カウントされません。