> ## Documentation Index > Fetch the complete documentation index at: https://help.vori.com/llms.txt > Use this file to discover all available pages before exploring further. # Download transaction data > Pull completed checkout transactions from the Vori API on a schedule, with cursor pagination, a resumable watermark, and rate-limit backoff. This guide walks through building a small client that pulls completed checkout transactions from the Vori API on a recurring schedule — the pattern you can use to feed your own reporting, BI, or accounting tools. ## Authenticate with an API key See the [authentication docs](https://help.vori.com/api/introduction#authentication) for how to create an API key. Make sure the role you assign it can read transactions. ## List transactions ```bash theme={null} curl -G "https://api.vori.com/v1/transactions" \ -H "Authorization: Bearer sk_live_" \ --data-urlencode 'completed_at[gte]=2026-08-01T00:00:00Z' \ --data-urlencode 'limit=100' ``` The list endpoint supports the following filters, declared as typed query parameters rather than one opaque filter blob: | Filter | Type | Notes | | -------------- | ---------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `store_id` | identifier | Store where the transaction was completed. Use this if you want to filter for a multi-store banner; otherwise, the endpoint returns transactions for all stores by default. | | `completed_at` | timestamp | Supports `[gte]`/`[lte]` range operators | | `type` | enum | Financial transaction type. Values: `sale`, `refund`, `void` | | `status` | enum | Lifecycle status of the transaction. Values: `completed`, `suspended`, `voided`, `expired` | | `lane_id` | identifier | Lane the transaction was completed on | | `employee_id` | identifier | Cashier who completed the transaction | | `shopper_id` | identifier | Identified loyalty shopper, if any | Every list response uses one envelope: ```json theme={null} { "data": [ ... ], "has_more": true } ``` ## Page through results Pagination is cursor-based, not offset-based — pass `limit` and `starting_after`/`ending_before` with the `id` of the last record you saw, and keep going while `has_more` is `true`: ```python theme={null} import requests BASE = "https://api.vori.com/v1/transactions" HEADERS = {"Authorization": "Bearer sk_live_"} 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 ``` Results are always ordered by `id` descending — this order is not client-selectable. ## Run it on a schedule for periodic downloads For a recurring reporting job, don't re-download the full history each run. Track a watermark and filter on `completed_at[gte]` from where the last run left off: ```python theme={null} 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() ``` One run covers every store your API key can read, so keep a single watermark rather than one per store. Every transaction carries `store_id`, so you can split the results by store in your own reporting store after the fact. Polling store by store multiplies your request count and leaves you several watermarks to keep in step, for no benefit. Two things make that watermark safe. It advances to the time the sync *started* rather than the largest `completed_at` observed, and it rewinds by a short buffer before it is stored. 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 — so a window beginning exactly where the last one ended would miss whatever landed in between. Fifteen minutes covers the ordinary gap; raise it if your lanes can be offline for longer. Both mean a run re-reads a little of what the previous run already wrote, which is why the write has to key on `id` and upsert. Re-reading a transaction is harmless. Missing one is not. ## Handle rate limits Requests are rate-limited per API key. Every response carries your current standing: * `X-RateLimit-Limit` — max requests allowed in the current window * `X-RateLimit-Remaining` — requests left in the window A `429` response includes `Retry-After` (seconds) — back off and retry rather than hammering the endpoint. ```python theme={null} 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 ``` ## Putting it together Everything above in one script. Point `VORI_API_KEY` at your key, replace `write_to_your_reporting_store`, and run it on whatever schedule suits you — cron, Airflow, a worker loop. It writes each page as it arrives rather than collecting everything first, so memory stays flat however long the window is, and a run that dies partway keeps the pages it already wrote. The checkpoint records how far through the current window it got and where the next window will start. The checkpoint is saved after every page and carries both values, but only the cursor changes as you go — the watermark keeps the previous run's value until the whole window has been written. Transactions come back newest first, so a run partway through has written the recent ones and has not yet reached the older ones. Advancing the watermark there and then failing would leave the next run starting after those older transactions, skipping them permanently. The watermark describes the window, not the page, so it can only move once the window is exhausted. ```python theme={null} """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() ``` Each row in `data` is a complete transaction, with its own line items and payments, so this script never needs a second call per transaction. The `refunds` array is the exception: it holds a reference to each refund — its ID, timestamps and amounts — not the refund's own line items and payments. You do not need to fetch those separately either, because a refund is itself a transaction and arrives as its own row in the same feed, where you can match it back by ID.
## Related topics - [Track Movement and Inventory Reports](/reporting/inventory-and-purchase-reports/track-movement-and-inventory-reports.md) - [List transactions](/api-reference/transactions/list-transactions.md) - [How to access and download CardConnect monthly statements](/payments-and-financials/cardconnect/how-to-access-and-download-cardconnect-monthly-statements.md)