# Load the export

## Change feed

Every pull returns everything that is new since your last pull: impressions that were added, impressions that changed, and impressions that were removed. With it comes a cursor. Pass the cursor with your next pull and you get whatever happened after it; upsert the rows on `impression_id`, and your table holds the whole export.

The change feed covers mobile rows (`"version": "v2"`). Desktop rows come through the day export: [Day export →](/docs/integration?method=day)

It does not matter how often you fetch. Pull every hour or once a week and the result is the same; a pull that failed, or a week you skipped, is covered by the next one. There are no dates to keep track of and no days to retry. Data that arrives late, and updates to earlier days, come with the next pull like everything else.

### 1. Authenticate

The steps show each raw call in curl; the Python tab builds them into one `loader.py` with `requests` and `duckdb`.

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

```bash tab=curl
export ATTEX_API="https://global-api.attentionpanel.com"
export ATTEX_API_KEY="…"
```

```python tab=Python
import os, time, pathlib, urllib.parse, re, shutil, tarfile, tempfile, requests, duckdb

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

### 2. Ask for what is new

A pull is one request. The first time, send `"feed": true`: you get everything your key may see, from its first allowed day. After that, send `"since"` with the cursor from your last pull: you get everything added, changed or removed after it. One feed covers one market.

```bash tab=curl
# The first pull
curl -sS -X POST "$ATTEX_API/impressions/export" \
  -H "Authorization: $ATTEX_API_KEY" -H "Content-Type: application/json" \
  -d '{"device":"mobile","market":"semarket","version":"v2","feed":true}'

# Every pull after that
curl -sS -X POST "$ATTEX_API/impressions/export" \
  -H "Authorization: $ATTEX_API_KEY" -H "Content-Type: application/json" \
  -d "{\"device\":\"mobile\",\"market\":\"semarket\",\"version\":\"v2\",\"since\":\"$CURSOR\"}"
```

```python tab=Python
def create_pull(market, cursor=None) -> requests.Response:
    body = {"device": "mobile", "market": market, "version": "v2"}
    body.update({"since": cursor} if cursor else {"feed": True})
    return api.post(f"{API}/impressions/export", json=body)
```

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

[All fields and errors: Create an export →](/docs/api-reference#create-export)

### 3. Wait for it to be ready

A pull takes a few minutes, also when nothing is new. Check its status now and then, waiting a little longer each time, until it is no longer `RUNNING`. When it says `SUCCEEDED`, it carries the download links and `next_cursor`: the cursor for your next pull.

```bash tab=curl
curl -sS "$ATTEX_API/exports/$EXPORT_ID" -H "Authorization: $ATTEX_API_KEY"
```

```python tab=Python
def wait_for(export_id, timeout_s=3 * 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)
```

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

[States, failure reasons and errors: Get an export →](/docs/api-reference#get-export)

### 4. Download the files

One link per calendar month with rows in the pull: `<export_id>_<year>_<month>.tar.gz`, holding a `day=DD/` folder per day with gzipped JSON Lines files. A pull with nothing new has no links. 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=`, and clear the folder first so the load sees exactly this pull.

```bash tab=curl
rm -rf feed/semarket
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="feed/semarket/year=$year/month=$month"
      mkdir -p "$dir" && curl -sS "$url" | tar -xz -C "$dir" --strip-components=1
    done
```

```python tab=Python
def download(job, dest) -> None:
    shutil.rmtree(dest, ignore_errors=True)
    for url in job.get("download_urls") or []:
        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)
```

[Download links and their expiry: Get an export →](/docs/api-reference#get-export)

### 5. Upsert, and keep the cursor

A pull holds each impression it sends once: as it is now, or, when it is gone, as `{"impression_id": "…", "date": "2026-10-21", "removed": true}` under its day. Upsert on `impression_id`: delete every id in the pull from your table, then insert the rows that are not removed. A removed id you never had is a no-op. A first pull replaces the table. Read the files as JSON Lines with schema union, so a missing key is a null and a field that appears later becomes a new column.

DuckDB below; the same delete and insert works in BigQuery, Snowflake or Postgres. Read the pull into `batch` and `gone` first, then apply it and save the cursor in one transaction, so a run that fails part-way changes nothing and the next run repeats the same pull safely.

```sql
CREATE OR REPLACE TEMP TABLE batch AS
  SELECT * FROM read_json_auto('feed/semarket/**/*.json.gz', union_by_name = true, hive_partitioning = true);
CREATE OR REPLACE TEMP TABLE gone AS
  SELECT json ->> 'impression_id' AS impression_id
  FROM read_ndjson_objects('feed/semarket/**/*.json.gz') WHERE json ->> 'removed' = 'true';
CREATE TABLE IF NOT EXISTS impressions_mobile AS
  SELECT COLUMNS(lambda c: c NOT IN ('removed', 'date')) FROM batch LIMIT 0;

BEGIN;
DELETE FROM impressions_mobile WHERE impression_id IN (SELECT impression_id FROM batch);
INSERT INTO impressions_mobile BY NAME
  SELECT COLUMNS(lambda c: c NOT IN ('removed', 'date')) FROM batch
  WHERE impression_id NOT IN (SELECT impression_id FROM gone);
INSERT OR REPLACE INTO feed_cursor VALUES ('semarket', '<next_cursor>');
COMMIT;
```

```python tab=Python
db = duckdb.connect("warehouse.db")
db.execute("CREATE TABLE IF NOT EXISTS feed_cursor (market VARCHAR PRIMARY KEY, cursor VARCHAR)")

def read_pull(folder) -> None:
    files = f"{folder}/**/*.json.gz"
    db.execute(f"""CREATE OR REPLACE TEMP TABLE batch AS
        SELECT * FROM read_json_auto('{files}', union_by_name = true, hive_partitioning = true)""")
    db.execute(f"""CREATE OR REPLACE TEMP TABLE gone AS
        SELECT json ->> 'impression_id' AS impression_id
        FROM read_ndjson_objects('{files}') WHERE json ->> 'removed' = 'true'""")
    db.execute("""CREATE TABLE IF NOT EXISTS impressions_mobile AS
        SELECT COLUMNS(lambda c: c NOT IN ('removed', 'date')) FROM batch LIMIT 0""")
    have = {c for (c,) in db.execute(
        "SELECT column_name FROM information_schema.columns WHERE table_name = 'impressions_mobile'").fetchall()}
    for name, kind in db.execute(
            "SELECT column_name, data_type FROM information_schema.columns WHERE table_name = 'batch'").fetchall():
        if name not in have and name not in ("removed", "date"):
            db.execute(f'ALTER TABLE impressions_mobile ADD COLUMN "{name}" {kind}')

def apply_pull(market, job, first) -> None:
    db.begin()
    if first:
        db.execute("DELETE FROM impressions_mobile")
    if job.get("download_urls"):
        db.execute("DELETE FROM impressions_mobile WHERE impression_id IN (SELECT impression_id FROM batch)")
        db.execute("""INSERT INTO impressions_mobile BY NAME
            SELECT COLUMNS(lambda c: c NOT IN ('removed', 'date')) FROM batch
            WHERE impression_id NOT IN (SELECT impression_id FROM gone)""")
    db.execute("INSERT OR REPLACE INTO feed_cursor VALUES (?, ?)", [market, job["next_cursor"]])
    db.commit()
```

### 6. Repeat whenever you like

Run the same steps again on any schedule; each pull starts from the cursor of the one before. We recommend a daily pull. Any schedule gives the same table, and a cursor does not expire, so after a failed run or a long pause the next pull simply catches up. Fetching every few minutes gains nothing, since a pull takes a few minutes itself.

```bash tab=curl
# 06:00 UTC daily; any schedule works
0 6 * * *  python loader.py
```

```python tab=Python
def sync(market) -> None:
    saved = db.execute("SELECT cursor FROM feed_cursor WHERE market = ?", [market]).fetchone()
    r = create_pull(market, saved[0] if saved else None)
    if r.status_code == 400 and r.json().get("message") == "CURSOR_GAP":
        r = create_pull(market)  # start over with a first pull
        saved = None
    r.raise_for_status()
    job = wait_for(r.json()["export_id"])
    if job["status"] != "SUCCEEDED":
        raise RuntimeError(job)
    download(job, f"feed/{market}")
    if job.get("download_urls"):
        read_pull(f"feed/{market}")
    apply_pull(market, job, first=saved is None)

if __name__ == "__main__":
    for market in ["semarket"]:
        sync(market)
```

### Starting over

Occasionally a pull answers 400 `CURSOR_GAP`, with an `action`: changes after your cursor can no longer be sent, so start again with a first pull, which replaces your table (the loader above does this). Do the same when your access changes, for example a new market or an earlier first day, since a feed only sends changes to rows. A 400 `CURSOR_INVALID` means the cursor is not one the API issued: pass `next_cursor` back exactly as you received it.

[Every error code: API reference →](/docs/api-reference)

## Day export

A day export returns the rows of chosen UTC days, mobile or desktop, as they are at that moment. It does not tell you what happened afterwards: rows that arrive later, or updates to those days, are picked up only by exporting the days again. Use it for a one-off pull of chosen days, and to keep desktop rows current, which have no change feed: export the last three days every day and replace those days in your table. To keep mobile rows current, use the [change feed](/docs/integration?method=feed).

### 1. Authenticate

The steps show each raw call in curl; the Python tab builds them into one `loader.py` with `requests`.

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

```bash tab=curl
export ATTEX_API="https://global-api.attentionpanel.com"
export ATTEX_API_KEY="…"
```

```python tab=Python
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

One export covers one platform, one market and an inclusive range of UTC dates. On mobile, send `"version": "v2"`, the current export, on every request; desktop has no versions, so leave the field out. v2 mobile data starts on 13 March 2026 in Sweden and Denmark, 26 March in Norway and 19 May in Finland: a `from_date` before 13 March is a 400 `V2_FROM_DATE_NOT_AVAILABLE` with that day in `available_from`. [Versions →](/docs/api-reference#versions)

```bash tab=curl
curl -sS -X POST "$ATTEX_API/impressions/export" \
  -H "Authorization: $ATTEX_API_KEY" -H "Content-Type: application/json" \
  -d '{"device":"desktop","market":"semarket","from_date":"2026-10-13","to_date":"2026-10-15"}'
```

```python tab=Python
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"]
```

```json response
{ "export_id": "acme_3f6c1d2e-8b4a-4c55-9a77-2d1f0e9b6a10", "status": "RUNNING" }
```

[All fields and errors: Create an export →](/docs/api-reference#create-export)

### 3. Wait for it to be ready

An export takes a few minutes. Check its status now and then, waiting a little longer each time, until it is no longer `RUNNING`; a succeeded one carries the download links.

```bash tab=curl
curl -sS "$ATTEX_API/exports/$EXPORT_ID" -H "Authorization: $ATTEX_API_KEY"
```

```python tab=Python
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)
```

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

[States, failure reasons and errors: Get an export →](/docs/api-reference#get-export)

### 4. Download the files

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 day. File names change on every export, so clear the platform's directory first; the load then sees exactly the days this job exported.

```bash tab=curl
rm -rf exports/desktop
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/desktop/year=$year/month=$month"
      mkdir -p "$dir" && curl -sS "$url" | tar -xz -C "$dir" --strip-components=1
    done
```

```python tab=Python
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)
```

[Download links and their expiry: Get an export →](/docs/api-reference#get-export)

### 5. Replace those days

Keep one table per platform; the two schemas differ. Read the 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`: a rebuilt day can give its rows new ids, and a merge by id keeps the old rows next to the new ones. DuckDB below, for desktop; for mobile, put `mobile` in place of `desktop` in the table name and the path. The same delete and insert works in BigQuery, Snowflake or Postgres.

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

### 6. Schedule it

To keep desktop rows current, run daily over the last three days: a day settles about two days after the day, as late uploads arrive. For mobile rows, the change feed does this without dates. Each run replaces the downloaded 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.

```bash tab=curl
# 06:00 UTC daily; load.sql holds the statement from step 5
0 6 * * *  python loader.py && duckdb warehouse.db < load.sql
```

```python tab=Python
from datetime import date, timedelta

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

[Every error code: API reference →](/docs/api-reference)