Skip to content
Octane AIOctane AI

A filtered people file for your warehouse

Goal: a gzipped CSV of the people who took a quiz last month and answered a given way, loaded into BigQuery, Snowflake or a spreadsheet, with the exact rule that built it stored beside it.

Uses: Exports with a search filter, and the export.completed webhook to know when the file is ready. Scopes: profiles:read (the file carries emails and phone numbers); webhooks:manage for the webhook variant.

1. Request the file

Terminal window
curl -X POST https://api.octaneai.com/v1/exports \
  -H "Authorization: Bearer $OCTANE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "dataset": "profiles",
    "quiz_id": "quiz_7c9e6679742540de944be07fc1f90ae7",
    "from": "2026-08-01",
    "to": "2026-08-31",
    "search": {
      "groups": [{ "all": [
        { "quiz_id": "quiz_7c9e6679742540de944be07fc1f90ae7", "kind": "choice", "op": "in", "page_key": "gifting", "component_key": "choice-gift1", "option_labels": ["Corporate gifting"] }
      ] }]
    }
  }'

The answer is 202 with the export in pending status and a Location header pointing at it. quiz_id, from and to stay on the export (from/to are the person’s last visit); inside search they answer 422.

2a. Be told when it is ready: the export.completed webhook

Subscribe an endpoint to export.completed once (webhooks:manage); the store’s other topics are unaffected:

Terminal window
curl -X POST https://api.octaneai.com/v1/webhooks \
  -H "Authorization: Bearer $OCTANE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"url": "https://hooks.example.com/octane", "topics": ["export.completed"], "description": "Warehouse loader"}'

When the file is built, the endpoint receives the export object (no download link; see the payload):

{
  "id": "evt_6ba7b8109dad11d180b400c04fd430c8",
  "type": "export.completed",
  "created": "2026-09-01T12:00:09Z",
  "api_version": "v1",
  "data": {
    "object": {
      "id": "exp_3f2504e04f8911d39a0c0305e82c3301",
      "dataset": "profiles",
      "status": "completed",
      "filters": { "quiz_id": "quiz_7c9e6679742540de944be07fc1f90ae7", "from": "2026-08-01", "to": "2026-08-31", "tz": null },
      "row_count": 1240,
      "bytes": 58311,
      "error": null,
      "source": "api",
      "requested_at": "2026-09-01T12:00:00Z",
      "started_at": "2026-09-01T12:00:02Z",
      "completed_at": "2026-09-01T12:00:09Z",
      "expires_at": "2026-10-01T12:00:00Z",
      "download_url": null,
      "download_url_expires_at": null,
      "urls": { "self": "https://api.octaneai.com/v1/exports/exp_3f2504e04f8911d39a0c0305e82c3301" }
    }
  }
}

The receiver (Flask and httpx, pip install flask httpx; Express; Go’s net/http) verifies the signature exactly as Verify a webhook signature does, checks that data.object.id is the export it asked for (the id you stored in step 1), then fetches urls.self with your key for a fresh 15-minute download_url and loads the file:

import base64
import csv
import gzip
import hashlib
import hmac
import io
import os
import time
import httpx
from flask import Flask, request


app = Flask(__name__)
SECRET = os.environ["OCTANEAI_WEBHOOK_SECRET"]  # whsec_...
HEADERS = {"Authorization": f"Bearer {os.environ['OCTANE_API_KEY']}"}
WANTED = "exp_3f2504e04f8911d39a0c0305e82c3301"  # the id step 1 printed; keep it where the receiver can read it


def verify(secret: str, headers, body: bytes) -> bool:
    try:
        key = base64.b64decode(secret[len("whsec_"):], validate=True)
        msg_id, ts, sigs = headers["webhook-id"], headers["webhook-timestamp"], headers["webhook-signature"]
        if abs(time.time() - int(ts)) > 300:
            return False
        expected = hmac.new(key, f"{msg_id}.{ts}.".encode() + body, hashlib.sha256).digest()
        for sig in sigs.split(" "):
            version, _, value = sig.partition(",")
            if version == "v1" and hmac.compare_digest(base64.b64decode(value, validate=True), expected):
                return True
    except (KeyError, ValueError):
        pass
    return False


def load(export_id: str) -> None:
    r = httpx.get(f"https://api.octaneai.com/v1/exports/{export_id}", headers=HEADERS, timeout=30)
    r.raise_for_status()
    export = r.json()  # status completed, download_url good for 15 minutes
    raw = httpx.get(export["download_url"], timeout=300).content  # signed URL: no Authorization header
    text = gzip.decompress(raw).decode("utf-8-sig")
    rows = list(csv.DictReader(io.StringIO(text)))
    with open(f"people-{export_id}.csv", "w", newline="") as f:
        f.write(text)  # load this file; keep the search body next to it as the rule that built it
    print(export["row_count"], len(rows))  # 1240 1240


@app.post("/octane")
def octane():
    if not verify(SECRET, request.headers, request.get_data()):
        return "bad signature", 401
    event = request.get_json()
    if event["type"] == "export.completed" and event["data"]["object"]["id"] == WANTED:
        load(WANTED)  # in production: enqueue, then answer; see the signature recipe
    return "", 200


if __name__ == "__main__":
    app.run(port=8080)

A failed export sends no webhook: keep the polling variant below as the fallback for that case, and dedupe on webhook-id as the signature recipe shows, since a retry can deliver the same event twice.

2b. Or poll, then download

For a setup without a public endpoint, poll GET /v1/exports/{export_id} until status is completed:

import csv
import gzip
import io
import os
import time
import httpx


HEADERS = {"Authorization": f"Bearer {os.environ['OCTANE_API_KEY']}"}
EXPORT = "https://api.octaneai.com/v1/exports/exp_3f2504e04f8911d39a0c0305e82c3301"  # the id step 1 printed


while True:
    r = httpx.get(EXPORT, headers=HEADERS, timeout=30)
    r.raise_for_status()
    export = r.json()
    if export["status"] not in ("pending", "running"):
        break
    time.sleep(10)
if export["status"] != "completed":
    raise SystemExit(export["error"])


raw = httpx.get(export["download_url"], timeout=300).content   # signed URL: no Authorization header
text = gzip.decompress(raw).decode("utf-8-sig")
rows = list(csv.DictReader(io.StringIO(text)))
print(export["row_count"], "rows,", export["bytes"], "bytes")  # 1240 rows, 58311 bytes


with open(f"people-{export['id']}.csv", "w", newline="") as f:
    f.write(text)  # load this file; keep the search body next to it as the rule that built it

Load people-<id>.csv with your warehouse’s CSV loader (the file starts with a UTF-8 byte-order mark). The three JSON cells (answers, points, formulas) load as strings; parse them in SQL (JSON_EXTRACT in BigQuery, PARSE_JSON in Snowflake) or in the job.

What is in the file

One row per person (one per person and quiz without quiz_id), with their latest completed result for the quiz: person_id, removed, removed_at, email, phone, first_seen_at, last_seen_at, sessions, completed, returning, orders, spent, currency, quiz, quiz_id, version, completed_at, answers, points, formulas, top_match, total_points, result_page. The header never changes, so a loader can be written once. A person in the range with no completed result for the quiz still gets a row, with the result cells blank.

import json


# one row of the file, as csv.DictReader yields it
row = {"answers": '{"number_input-re853": {"label": "Budget", "kind": "number", "value": 120, "values": ["120"]}}'}
answers = json.loads(row["answers"]) if row["answers"] else {}
print(answers.get("number_input-re853", {}).get("value"))  # 120

Things to know

  • One file of a kind at a time. A second profiles export while one is building answers 409 conflict; poll the running one. Requesting costs 2 units, polling 1.
  • The link is short-lived. download_url is good for 15 minutes; GET /v1/exports/{export_id} mints a fresh one on every call while the export is completed, which is why the webhook payload carries none. The file itself is kept up to 30 days (expires_at), then the export answers 404.
  • The count is the file’s. A paged search caps total at 5,000 and bounds each of its statements at 10 seconds; the export has neither limit, holds every matching person, and row_count says how many.
  • Nightly. Run it with yesterday as both from and to to get the people last seen that day, one file per day.
  • No email for API requests. Only a file requested from the dashboard emails the admin; your job listens for export.completed or polls.