klaviyo-poller

A snapshot poller that turns Klaviyo into something it isn't: an audited system. Klaviyo has no native change-history API, so the only way to know who changed which flow step on which day is to snapshot every entity on a schedule and diff successive snapshots offline. This script handles the snapshot half of that pattern.

What it does

1. Pulls every campaign, flow, list, and segment from one or more Klaviyo accounts.
2. Strips volatile fields (timestamps that move every poll without anyone touching the entity), then SHA-256 hashes the canonicalized JSON.
3. Bulk-loads the result into Postgres via COPY into a temp staging table, then upserts with ON CONFLICT DO NOTHING on (account, entity_type, entity_id, content_hash).

The third step is the key trick: identical snapshots collapse to a single row, so the table grows with changes, not with poll frequency. Daily polling against an account that never edits anything writes the same row count as a single poll.

What it deliberately doesn't do

Diff emission. Once you have two distinct hashes for the same entity, you can compute field-level diffs in plain SQL or a downstream job. Keeping the diff layer separate means:

• The snapshot table is a faithful raw archive. Replay diffs from scratch when your rules change.
• Snapshot ingestion stays cheap and idempotent. Rerun any day's poll any number of times without side effects.
• Tooling can plug in: write SQL diffs, ship snapshots to a warehouse, build a UI on top, all from the same table.

A reference SQL diff query is below.

Quickstart

git clone https://github.com/koreydillon/klaviyo-poller.git
cd klaviyo-poller
python3 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
cp .env.example .env   # fill in Klaviyo keys + Postgres conn
psql "$DATABASE_URL" -f schema.sql
python klaviyo_poller.py --dry-run   # fetch only, no DB writes
python klaviyo_poller.py             # full snapshot

CLI flags:

--account NAME        Filter to a single account (repeatable)
--entity-type TYPE    Filter to campaign | flow | list | segment (repeatable)
--dry-run             Fetch and count, no DB writes

Run it on whatever scheduler you have. Cron, GitHub Actions, AWS Glue, Airflow, doesn't matter. Each run is idempotent.

Configuration

Multiple accounts are first-class. Set KLAVIYO_ACCOUNTS to a comma-separated list of names, then provide a key per name:

KLAVIYO_ACCOUNTS=brand_a,brand_b
KLAVIYO_API_KEY_BRAND_A=pk_xxx
KLAVIYO_API_KEY_BRAND_B=pk_yyy

Postgres uses standard libpq env vars (PGHOST, PGDATABASE, PGUSER, PGPASSWORD, PGPORT). If those are already in your shell environment, you don't need to repeat them in .env. Set KLAVIYO_SNAPSHOT_TABLE to override the default klaviyo_snapshot table name (e.g., public.klaviyo_snapshot or a schema-qualified name in your warehouse).

Schema

schema.sql creates a single table:

klaviyo_snapshot (
    id, account, entity_type, entity_id, entity_name, status,
    klaviyo_updated_at, snapshot_data JSONB, content_hash,
    captured_at TIMESTAMPTZ DEFAULT NOW()
)

A unique index on (account, entity_type, entity_id, content_hash) is the dedup mechanism. A composite index on (account, entity_type, entity_id, captured_at DESC) makes "latest snapshot for this entity" and "all snapshots over time" both fast.

Diff query (reference)

The poller stops at storing snapshots. Here's the matching diff query if you want to know what changed for a given entity between any two captured times:

WITH paired AS (
  SELECT
    entity_type,
    entity_id,
    entity_name,
    snapshot_data,
    captured_at,
    LAG(snapshot_data) OVER w AS prev_data,
    LAG(captured_at)   OVER w AS prev_at
  FROM klaviyo_snapshot
  WHERE account = 'brand_a'
  WINDOW w AS (PARTITION BY account, entity_type, entity_id ORDER BY captured_at)
)
SELECT entity_type, entity_id, entity_name, prev_at, captured_at,
       jsonb_diff_val(prev_data, snapshot_data) AS changed_fields
FROM paired
WHERE prev_data IS DISTINCT FROM snapshot_data
ORDER BY captured_at DESC;

(jsonb_diff_val is a stock Postgres helper. There are several public implementations. Pick one or write your own field-level diff in the language of your downstream pipeline.)

Why content-hash dedup, not ON CONFLICT (entity_id) DO UPDATE

Two reasons:

1. History is the product. An update-in-place table only ever holds the current state. The whole reason this script exists is to keep a time-series of snapshots so you can answer "what did this flow look like on March 4th?"
2. Volatile-field noise. Klaviyo bumps updated_at and recipient counts on entities you didn't touch. Hashing after stripping those fields lets you write only on real edits, even though the API response changed.

Klaviyo API notes

• The poller uses the official klaviyo-api SDK, which handles 429 retries and the token-bucket rate limit (burst 75/s, steady 700/min on most read endpoints) internally.
• Campaigns require a channel filter (email or sms); the fetcher requests both and concatenates.
• All endpoints use JSON:API cursor pagination. The cursor parser handles both URL-encoded (page%5Bcursor%5D=) and unencoded (page[cursor]=) link headers because Klaviyo has been inconsistent about which it returns.

License

MIT. See LICENSE.