A daily loader

A minimal integration: create an export job, poll it, download and unpack the archives, and replace the exported days in your table. curl shows each raw call; the Python tab builds the same steps into one loader.py with requests.

1. Authenticate

Every request carries your key in Authorization, without a scheme prefix. Requests without a key get 401; a key outside its entitlements gets 403.

export ATTEX_API="https://global-api.attentionpanel.com"
export ATTEX_API_KEY="…"

import os, time, pathlib, urllib.parse, requests

API = os.environ.get("ATTEX_API", "https://global-api.attentionpanel.com")
api = requests.Session()
api.headers["Authorization"] = os.environ["ATTEX_API_KEY"]

2. Create an export job

One job covers one platform, one market and an inclusive range of UTC dates. On mobile, "version": "v2" selects the current export, the one these docs describe; without the field a mobile job returns the legacy export until the cutover. Desktop has no versions, so leave the field out. v2 data starts on 13 October 2026, and earlier days are being added: a from_date before the first day available is a 400 V2_FROM_DATE_NOT_AVAILABLE with that day in available_from.

All fields and errors: Create an export →

curl -sS -X POST "$ATTEX_API/impressions/export" \
  -H "Authorization: $ATTEX_API_KEY" -H "Content-Type: application/json" \
  -d '{"device":"mobile","market":"semarket","from_date":"2026-10-13","to_date":"2026-10-19","version":"v2"}'

def create_export(device, market, from_date, to_date) -> str:
    body = {"device": device, "market": market, "from_date": from_date, "to_date": to_date}
    if device == "mobile":
        body["version"] = "v2"
    r = api.post(f"{API}/impressions/export", json=body)
    r.raise_for_status()
    return r.json()["export_id"]

Response

{ "export_id": "acme_3f6c1d2e-8b4a-4c55-9a77-2d1f0e9b6a10", "status": "RUNNING", "version": "v2" }

3. Poll until terminal

Poll with backoff and a deadline. RUNNING is the only non-terminal state.

States, failure reasons and errors: Get an export →

curl -sS "$ATTEX_API/exports/$EXPORT_ID" -H "Authorization: $ATTEX_API_KEY"

def wait_for(export_id, timeout_s=3600) -> dict:
    delay, deadline = 10, time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        r = api.get(f"{API}/exports/{urllib.parse.quote(export_id)}")
        r.raise_for_status()
        job = r.json()
        if job["status"] != "RUNNING":
            return job
        time.sleep(delay)
        delay = min(delay * 2, 120)
    raise TimeoutError(export_id)

Response

{
  "status": "SUCCEEDED",
  "version": "v2",
  "download_urls": [
    "https://…/acme_3f6c1d2e-8b4a-4c55-9a77-2d1f0e9b6a10_2026_10.tar.gz?X-Amz-Signature=…"
  ]
}

4. Download the archives

One link per calendar month in the range: <export_id>_<year>_<month>.tar.gz, holding a day=DD/ folder per day with gzipped JSON Lines files. URLs are presigned and valid for 24 hours after completion; do not send your API key to them, the signature is in the query string. Unpack each archive under year=<year>/month=<month>/ so the day folders line up as year=/month=/day=, which the load in step 5 reads as the row's date. File names change on every export, so clear the platform's directory first; the load then sees exactly the days this job exported.

Download links and their expiry: Get an export →

rm -rf exports/mobile
curl -sS "$ATTEX_API/exports/$EXPORT_ID" -H "Authorization: $ATTEX_API_KEY" \
  | jq -r '.download_urls[]' \
  | while read -r url; do
      name=$(basename "${url%%\?*}")                 # <export_id>_<year>_<month>.tar.gz
      ym=${name%.tar.gz}; month=${ym##*_}; year=${ym%_*}; year=${year##*_}
      dir="exports/mobile/year=$year/month=$month"
      mkdir -p "$dir" && curl -sS "$url" | tar -xz -C "$dir" --strip-components=1
    done

import re, shutil, tarfile, tempfile

def download(job, dest) -> None:
    shutil.rmtree(dest, ignore_errors=True)
    for url in job["download_urls"]:
        name = urllib.parse.urlparse(url).path.rsplit("/", 1)[-1]  # <export_id>_<year>_<month>.tar.gz
        year, month = re.search(r"_(\d{4})_(\d{2})\.tar\.gz$", name).groups()
        with requests.get(url, stream=True) as r, tempfile.TemporaryFile() as tmp:  # plain requests: no Authorization header
            r.raise_for_status()
            for chunk in r.iter_content(1 << 20):
                tmp.write(chunk)
            tmp.seek(0)
            with tarfile.open(fileobj=tmp) as tar:
                for member in tar.getmembers():
                    if not member.isfile():
                        continue
                    out = pathlib.Path(dest, f"year={year}", f"month={month}", *pathlib.Path(member.name).parts[1:])
                    out.parent.mkdir(parents=True, exist_ok=True)
                    with tar.extractfile(member) as src, open(out, "wb") as f:
                        shutil.copyfileobj(src, f)

5. Load and replace by day

Keep one table per platform; the two schemas differ. Read the platform's files as JSON Lines with schema union (a missing key is a null), delete every day present in the batch from your table, and insert the batch. Do not merge on impression_id: when a day is rebuilt its rows can get new ids, and a merge by id keeps the old rows next to the new ones. DuckDB below, for mobile; the desktop statement is the same with desktop in place of mobile in the table name and the path. The same delete-insert works in BigQuery, Snowflake or Postgres.

BEGIN;
CREATE TEMP TABLE batch AS
  SELECT * FROM read_json_auto('exports/mobile/**/*.json.gz',
                               union_by_name = true, hive_partitioning = true);
CREATE TABLE IF NOT EXISTS impressions_mobile AS SELECT * FROM batch LIMIT 0;
DELETE FROM impressions_mobile USING (SELECT DISTINCT year, month, day FROM batch) AS d
  WHERE impressions_mobile.year = d.year AND impressions_mobile.month = d.month AND impressions_mobile.day = d.day;
INSERT INTO impressions_mobile BY NAME SELECT * FROM batch;
DROP TABLE batch;
COMMIT;

6. Schedule it

A day settles about two days after the day: late uploads from devices arrive first, then the brand reading. Run daily over the last three days. Each run replaces the platform's files and the load replaces those days, so reruns and overlapping windows are safe. Treat NO_DATA as an empty day, anything else as a failure.

Every error code: API reference →

# 06:00 UTC daily; load.sql holds the mobile and desktop statements from step 5
0 6 * * *  python loader.py && duckdb warehouse.db < load.sql

from datetime import date, timedelta

if __name__ == "__main__":
    end = date.today() - timedelta(days=1)
    start = end - timedelta(days=2)
    for device, market in [("mobile", "semarket"), ("desktop", "semarket")]:
        job = wait_for(create_export(device, market, start.isoformat(), end.isoformat()))
        if job["status"] == "SUCCEEDED":
            download(job, f"exports/{device}")
        elif job.get("failure_reason") != "NO_DATA":
            raise RuntimeError(job)