> ## Documentation Index
> Fetch the complete documentation index at: https://unif.dev/llms.txt
> Use this file to discover all available pages before exploring further.
# Sync Unif into your warehouse
> A nightly pipeline that lands normalized commerce data next to your own.
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
Backfill Unif IDs onto your existing rows. Do this once; everything after is keyed on IDs.
One [export job](/docs/workflows/jobs) per entity and market, collected by webhook.
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.
```python theme={null}
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.
```python theme={null}
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
```python theme={null}
@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
```
```python theme={null}
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.
```sql theme={null}
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](/docs/concepts/unified-model). 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
```sql theme={null}
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](/docs/concepts/metrics).
## Operating the pipeline
Track the average `meta.completeness` per load. A sustained drop is an upstream change worth
catching before dashboards look wrong.
A silent pipeline and a healthy one look identical until someone asks why the numbers stopped.
Compare `job.result_count` against rows landed. A gap means a load problem, not a data problem.
Export jobs are the largest line in most workspaces. `GET /usage/events` attributes cost by
endpoint.