JSONL Store
このスキルは、スケジュールジョブが以前に行ったこと（発行済みアラート、処理済みID、キャプチャ済みイベントなど）を記憶する必要があるが、データベースを使うほどではない場合に使用します。JSON Lines（1行に1つのJSONオブジェクト）は追記に適しており、人間が読めてgrep可能で、ファイル全体をロードせずにストリーミングできます。このヘルパーは、アトミックな追記、壊れた行をスキップするストリーム読み取り、append_unique(record, key)による重複排除を提供するため、同じIDが複数回書き込まれることはありません。
呼び出すタイミング
- ユーザーが「JSONLファイルに結果を記録」「イベントをディスクに追記」「実行間でIDによる重複排除」「DBなしでシンプルな履歴を保持」と言った場合。
- cronやスケジュールスクリプトが、フラットファイルに実行間の状態を永続化する必要がある場合。
呼び出さないタイミング
- クエリ、インデックス、同時書き込みが必要な場合 → SQLite（
sqlite-state）や本格的なDBを使用してください。 - レコードが巨大またはバイナリの場合 → JSONLは小さな構造化行を対象としています。
具体的な例
ユーザー入力：
各実行後、新しいアラートをalerts.jsonlに追記するが、同じアラートIDが二度と追加されないようにしてください。
出力：
# assets/store.py をプロジェクトにコピーしてから：
from store import JsonlStore
store = JsonlStore("data/alerts.jsonl")
for alert in new_alerts:
if store.append_unique(alert, key="id"):
send(alert) # 以前の実行で見られなかったIDに対してのみ実行
print("履歴サイズ:", store.count())
読み取りは1行ずつストリーミングされるため、1年分の履歴でもメモリに完全にロードされることはなく、クラッシュによる中途半端な最終行はスキップされ、次の実行が破綻することはありません。
適用パターン
- 1行に1つのJSONオブジェクト。
ensure_ascii=Falseを指定して追記することで、テキストが読みやすくなります。 - ジェネレータによるストリーム読み取り。空白行や壊れた行は例外を発生させずにスキップします。
seen_keys(key)/append_unique(record, key)による重複排除で、実行を冪等にします。- スケジューラと組み合わせることで、ファイルが呼び出し間の永続メモリとなります。
参照: assets/store.py。
ソース
作者のスケジュールドボットの状態管理から抽出されました。v1.0.0。 関連項目: [[sqlite-state]], [[cron-dispatch]], [[pipeline-orchestrator]]。
→ Trawlkitで完全な実行可能なボットを構築してください。