Sync Unif into your warehouse

Unif data is most valuable joined to data you already have — your orders, your margins, your creator payouts. This builds the pipeline that gets it there.

The shape of it

Resolve once, store the IDs

Backfill Unif IDs onto your existing rows. Do this once; everything after is keyed on IDs.

Export nightly with a job

One export job per entity and market, collected by webhook.

Load as an append-only fact table

Key on entity, period and currency. Never overwrite a historical row.

1. Backfill IDs

Resolve rather than enrich for a backfill — you only need identity at this stage, and resolution is cheaper.

import itertools, requests

AUTH = {"Authorization": f"Bearer {UNIF_API_KEY}"}

def chunks(rows, size=100):
    it = iter(rows)
    while batch := list(itertools.islice(it, size)):
        yield batch

for batch in chunks(shop_rows):
    resp = requests.post(
        "https://api.unif.dev/v1/resolve",
        headers=AUTH,
        json={"inputs": [
            {"type": "shop", "url": r.url, "ref": str(r.id)} for r in batch
        ]},
        timeout=60,
    ).json()

    for result in resp["data"]:
        if result["status"] == "resolved":
            db.update_shop(id=result["ref"], unif_id=result["id"])
        else:
            db.flag_unresolved(id=result["ref"], reason=result["status"])

Keep the unresolved rows visible. A growing not_found count means your source list is decaying, and that is worth knowing before it shows up as missing rows downstream.

2. Export nightly

An export job shards past the 1,000-row search depth and hands you a file.

job = requests.post(
    "https://api.unif.dev/v1/jobs",
    headers={**AUTH, "Idempotency-Key": f"export-shops-us-{run_date}"},
    json={
        "type": "export",
        "input": {
            "entity": "shop",
            "format": "jsonl",
            "search": {
                "market": "US",
                "period": {"preset": "previous_month"},
                "currency": "USD",
                "filters": {"revenue": {"gte": 10000}},
            },
        },
        "webhook_id": WEBHOOK_ID,
    },
    timeout=30,
).json()

The Idempotency-Key is doing real work here. If your scheduler retries the task — and eventually it will — the same key returns the original job instead of running and charging for a second export.

Prefer previous_month or an explicit date range over last_30d for warehouse loads. A preset means a different window on every run, so the same job run twice produces two incomparable files.

3. Collect on the webhook

@app.post("/webhooks/unif")
def handle():
    if not verify(request.get_data(), request.headers["Unif-Signature"], SECRET):
        return "", 401

    event = request.get_json()
    if event["type"] == "job.completed":
        load_export.delay(event["data"]["job_id"])
    elif event["type"] == "job.failed":
        alert(f"Unif export failed: {event['data']['job_id']}")

    return "", 200

def load_export(job_id):
    job = requests.get(f"https://api.unif.dev/v1/jobs/{job_id}", headers=AUTH).json()

    if job["status"] == "partially_completed":
        log.warning("export %s: %s rows failed", job_id, job["progress"]["failed"])

    urllib.request.urlretrieve(job["download_url"], f"/tmp/{job_id}.jsonl")
    copy_into_warehouse(f"/tmp/{job_id}.jsonl")

The download_url is valid for 24 hours. Pull the file into your own storage rather than pointing a downstream job at it.

4. Model it as append-only

The schema mistake to avoid is a table with one row per shop that gets overwritten. Metrics are period-scoped, so an overwrite destroys the ability to compare.

CREATE TABLE unif_shop_metrics (
  unif_shop_id   TEXT        NOT NULL,
  channel        TEXT        NOT NULL,
  market         CHAR(2)     NOT NULL,
  period_start   DATE        NOT NULL,
  period_end     DATE        NOT NULL,
  granularity    TEXT        NOT NULL,
  currency       CHAR(3)     NOT NULL,
  revenue        NUMERIC,
  units_sold     BIGINT,
  revenue_growth_rate NUMERIC,
  refreshed_at   TIMESTAMPTZ NOT NULL,
  loaded_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
  PRIMARY KEY (unif_shop_id, period_start, period_end, currency)
);

CREATE TABLE unif_shop (            -- attributes: overwrite freely
  unif_shop_id  TEXT PRIMARY KEY,
  channel       TEXT NOT NULL,
  market        CHAR(2) NOT NULL,
  name          TEXT,
  handle        TEXT,
  seller_type   TEXT,
  updated_at    TIMESTAMPTZ NOT NULL
);

Two tables, following the attributes-versus-metrics split. A shop’s name can be overwritten; its August revenue cannot.

Take period_start, period_end, granularity and currency from the response, not from your request. If period.adjusted was true, the window you measured is not the one you asked for — and keying on the requested dates corrupts the series silently.

Joining to your own data

SELECT
  o.sku,
  o.units          AS our_units,
  m.units_sold     AS market_units,
  o.units::numeric / NULLIF(m.units_sold, 0) AS share_of_category_volume
FROM our_orders o
JOIN our_products p     ON p.sku = o.sku
JOIN unif_shop_metrics m
  ON m.unif_shop_id = p.unif_shop_id
 AND m.period_start = date_trunc('month', o.ordered_at)::date
WHERE m.currency = 'USD';

Unif revenue counts orders placed, before returns and fees. Your own revenue is almost certainly net. They are different measures — use Unif for market context and relative share, not to reconcile your books. See the metrics reference.

Operating the pipeline

Watch completeness

Track the average meta.completeness per load. A sustained drop is an upstream change worth catching before dashboards look wrong.

Alert on job.failed

A silent pipeline and a healthy one look identical until someone asks why the numbers stopped.

Reconcile row counts

Compare job.result_count against rows landed. A gap means a load problem, not a data problem.

Keep credits in view

Export jobs are the largest line in most workspaces. GET /usage/events attributes cost by endpoint.

Getting started

Core concepts

Workflows

Guides

Platform

Release notes

Sync Unif into your warehouse

The shape of it

1. Backfill IDs

2. Export nightly

3. Collect on the webhook

4. Model it as append-only

Joining to your own data

Operating the pipeline

Watch completeness

Alert on job.failed

Reconcile row counts

Keep credits in view

​The shape of it

​1. Backfill IDs

​2. Export nightly

​3. Collect on the webhook

​4. Model it as append-only

​Joining to your own data

​Operating the pipeline

Watch completeness

Alert on job.failed

Reconcile row counts

Keep credits in view

The shape of it

1. Backfill IDs

2. Export nightly

3. Collect on the webhook

4. Model it as append-only

Joining to your own data

Operating the pipeline