Add historical snapshot generation runner

This commit is contained in:
devRaGonSa
2026-03-21 13:16:33 +01:00
parent d466b51d68
commit 97331a76d4
6 changed files with 342 additions and 11 deletions

View File

@@ -123,6 +123,31 @@ def persist_historical_snapshot(
)
def persist_historical_snapshot_batch(
snapshots: list[dict[str, object]],
*,
db_path: Path | None = None,
) -> list[HistoricalSnapshotRecord]:
"""Persist a batch of snapshots generated in one runner cycle."""
records: list[HistoricalSnapshotRecord] = []
for snapshot in snapshots:
records.append(
persist_historical_snapshot(
server_key=str(snapshot["server_key"]),
snapshot_type=str(snapshot["snapshot_type"]),
payload=snapshot["payload"],
metric=snapshot.get("metric"),
window=snapshot.get("window"),
generated_at=snapshot.get("generated_at"),
source_range_start=snapshot.get("source_range_start"),
source_range_end=snapshot.get("source_range_end"),
is_stale=bool(snapshot.get("is_stale", False)),
db_path=db_path,
)
)
return records
def get_historical_snapshot(
*,
server_key: str,