SQLite 状態管理
このスキルは、定期的に実行されるジョブが過去の処理内容を記憶する必要がある場合に使用します。つまり、どのアイテムについて既にアラートを送信したか、最後に処理したカーソルは何か、などを記憶し、次の実行時に同じ通知を重複して送信したり、同じスクレイピングをやり直したりするのを防ぎます。標準のSQLiteファイルを使用するため、サーバーを立てたり、追加の依存関係をインストールする必要はありません。
使用すべき場合
- 「同じものを2回通知しないで」「実行間で重複を排除して」「最後のIDを覚えて」「中断したところから再開して」といったユーザーの要望がある場合。
- 会話内のコードが、毎回すべてを再処理するcronジョブや定期実行スクリプトである場合。
使用すべきでない場合
- 状態が小さく、1回の実行内でのみ一時的に必要な場合。この場合、単純なsetやdictで十分です。
- 複数のマシンが同時に状態を共有する必要がある場合。この場合は、単一ファイルではなく、実際のサーバーデータベースを使用してください。
具体例
ユーザー入力：
私のニューススクレイパーは毎時間同じ見出しを再送信します。新しいものだけを送信するようにしてください。
出力：
from state import filter_new, mark_seen
db = "news.db"
fresh = filter_new(db, [a["url"] for a in articles]) # 未処理のURLのみ
for url in fresh:
send_alert(url)
mark_seen(db, *fresh) # 次回の実行のために記憶
適用するパターン
- 各アイテムを、配列のインデックスではなく、一意で安定したキー（URL、ID、コンテンツのハッシュなど）で特定します。
filter_newを使用して処理対象を決定し、アクションが成功した後にのみmark_seenを呼び出します。- 名前付きカーソル（
set_cursor/get_cursor）を使用して進捗状況を保存し、再開可能なスクレイパーが途中から再開できるようにします。 - 永続的な書き込みと同時読み取りのためにWALモードを有効にします。書き込みは1つに制限します。
- 1つの
.dbファイルをバックアップまたはバージョン管理します。これがジョブの全記憶です。
参考: assets/state.py。
ソース
作者の自動化プロジェクトにおける本番環境での使用経験から抽出されました。v1.0.0。 関連項目: [[pipeline-orchestrator]], [[cron-dispatch]], [[webhook-receiver]]。
→ Trawlkitで完全に実行可能なボットを構築します。