Download transaction data

curl -G "https://api.vori.com/v1/transactions" \
  -H "Authorization: Bearer sk_live_<secret>" \
  --data-urlencode 'completed_at[gte]=2026-08-01T00:00:00Z' \
  --data-urlencode 'limit=100'

{
  "data": [ ... ],
  "has_more": true
}

import requests

BASE = "https://api.vori.com/v1/transactions"
HEADERS = {"Authorization": "Bearer sk_live_<secret>"}

def fetch_all_transactions(completed_at_gte):
    params = {
        "completed_at[gte]": completed_at_gte,
        "limit": 100,
    }
    transactions = []

    while True:
        resp = requests.get(BASE, headers=HEADERS, params=params)
        resp.raise_for_status()
        body = resp.json()
        transactions.extend(body["data"])

        if not body["has_more"]:
            break
        params["starting_after"] = body["data"][-1]["id"]

    return transactions

from datetime import datetime, timedelta, timezone

def run_periodic_download():
    # Load from wherever you persist it — a database row, a config
    # store, a file.
    last_watermark = load_watermark()

    # Where the next run starts. Taken before fetching, not from the max
    # completed_at seen — a transaction that completes mid-run can land behind
    # that value and would be missed on the next poll. The 15 minutes covers
    # the gap between a transaction completing and reaching Vori.
    next_watermark = (
        datetime.now(timezone.utc) - timedelta(minutes=15)
    ).isoformat()

    transactions = fetch_all_transactions(last_watermark)
    write_to_your_reporting_store(transactions)

    # Persist only after the write succeeds. If write_to_your_reporting_store
    # raises, this line is never reached, so the stored watermark stays put
    # and the next run retries the same window instead of skipping it.
    save_watermark(next_watermark)

    print(f"Pulled {len(transactions)} transactions since {last_watermark}")

if __name__ == "__main__":
    run_periodic_download()

import time

def get_with_backoff(url, **kwargs):
    resp = requests.get(url, **kwargs)
    if resp.status_code == 429:
        time.sleep(int(resp.headers.get("Retry-After", "5")))
        return get_with_backoff(url, **kwargs)
    resp.raise_for_status()
    return resp

"""Incrementally download Vori transactions into your own reporting store.

Each run pulls every transaction completed since the last successful run,
writing each page as it arrives so nothing accumulates in memory and an
interrupted run resumes where it stopped.
"""

import json
import os
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path

import requests

BASE = "https://api.vori.com/v1/transactions"
HEADERS = {"Authorization": f"Bearer {os.environ['VORI_API_KEY']}"}
PAGE_SIZE = 100
MAX_RETRIES = 5

# Every transaction reaches Vori some time after it completes — usually
# seconds, but a lane that has lost its connection sends its batch when it
# reconnects. Each run therefore starts a little before the last one did, so
# anything that landed after the previous window closed is still picked up.
# Widen this if your lanes can be offline for longer.
WATERMARK_BUFFER = timedelta(minutes=15)

# One checkpoint for the whole key. A run covers every store the key can read,
# and each transaction carries store_id if you want to split them later.
CHECKPOINT_FILE = Path(
    os.environ.get("VORI_CHECKPOINT_FILE", "vori-transactions-checkpoint.json")
)

# Used on the first run, before a checkpoint exists. Set it to the point in
# history you want to start from.
DEFAULT_START = "2026-01-01T00:00:00Z"


def load_checkpoint():
    if CHECKPOINT_FILE.exists():
        return json.loads(CHECKPOINT_FILE.read_text())
    return {"watermark": DEFAULT_START}


def save_checkpoint(checkpoint):
    # Write then rename, so a crash mid-write cannot leave a half-written file
    # that the next run fails to parse.
    tmp = CHECKPOINT_FILE.with_suffix(".tmp")
    tmp.write_text(json.dumps(checkpoint))
    tmp.replace(CHECKPOINT_FILE)


def get_with_backoff(params):
    for _ in range(MAX_RETRIES):
        resp = requests.get(BASE, headers=HEADERS, params=params, timeout=30)
        if resp.status_code != 429:
            resp.raise_for_status()
            return resp
        # Wait exactly as long as the response asks rather than guessing.
        time.sleep(int(resp.headers.get("Retry-After", "5")))

    raise RuntimeError(f"still rate limited after {MAX_RETRIES} attempts")


def iter_pages(completed_at_gte, starting_after=None):
    """Yield each page of transactions completed at or after `completed_at_gte`.

    The bound is inclusive, which is why the watermark can be re-used verbatim
    as the next run's lower bound without losing anything on the boundary.
    """
    params = {"completed_at[gte]": completed_at_gte, "limit": PAGE_SIZE}
    if starting_after:
        params["starting_after"] = starting_after

    while True:
        body = get_with_backoff(params).json()
        page = body["data"]

        if page:
            yield page

        if not body["has_more"]:
            return

        # A cursor, not an offset: the id of the last row on the page just read.
        params["starting_after"] = page[-1]["id"]


def write_to_your_reporting_store(transactions):
    """Replace this with your own write.

    Key on `id` and upsert. The watermark advances to the time the run started
    less the buffer, so consecutive runs overlap and the same transaction can
    arrive more than once — that overlap is what guarantees nothing is skipped.
    """
    for txn in transactions:
        print(txn)


def run_periodic_download():
    checkpoint = load_checkpoint()
    watermark = checkpoint["watermark"]

    # Where the next run will start: this run's start time, less the buffer.
    # Taken before fetching, not after — a transaction that completes mid-run
    # can land behind it and would be missed next time. A resumed run keeps the
    # value the interrupted one chose, for the same reason.
    next_watermark = checkpoint.get("next_watermark") or (
        (datetime.now(timezone.utc) - WATERMARK_BUFFER)
        .isoformat()
        .replace("+00:00", "Z")
    )

    count = 0

    for page in iter_pages(watermark, checkpoint.get("cursor")):
        write_to_your_reporting_store(page)
        count += len(page)

        # Saved only after the page is written, and recording the cursor rather
        # than the clock: rows arrive newest first, so a run that has read three
        # pages has the newest transactions and none of the older ones. Moving
        # the watermark here would skip everything still to come.
        save_checkpoint(
            {
                "watermark": watermark,
                "next_watermark": next_watermark,
                "cursor": page[-1]["id"],
            }
        )

    # The whole window has been written, so the watermark can move and the
    # cursor is no longer needed.
    save_checkpoint({"watermark": next_watermark})

    print(f"Pulled {count} transactions since {watermark}")


if __name__ == "__main__":
    run_periodic_download()