Skip to main content

Upload historical data

Models need history. Download the sample dataset, upload its e-commerce CSV (~1,700 users, ~21,000 events: order.placed, order.completed, order.returned, support.ticket_opened), and tell Tessera how its columns become canonical events.

The upload flow is three calls — POST /uploads returns a resumable URL, you PUT the bytes to it, POST /uploads/{id}:complete fingerprints the object, attaches it to a file_upload source, and enqueues the sync that lands + profiles it (the example polls the upload's sync to the end: if that sync fails before ingesting the file, a follow-up sync takes it over and the upload names it) — then a column mapping and a full sync normalize every row:

02-upload-events.sh
#!/usr/bin/env bash
# Quickstart step 2 — download the sample dataset, upload the e-commerce CSV,
# map its columns to canonical events, and run the first full sync.
set -euo pipefail

: "${TESSERA_API_BASE:?e.g. https://api.staging.tessera.zone/v1}"
: "${TESSERA_API_KEY:?your tsk_test_… key}"
: "${TESSERA_STATE_DIR:=.}"
# The published dataset lives on the docs site; CI points this at its stack.
SAMPLE_URL="${TESSERA_SAMPLE_URL:-https://docs.staging.tessera.zone/downloads/tessera-sample-v1.zip}"
AUTH=(-H "Authorization: Bearer $TESSERA_API_KEY")

# --- 1. download + verify the sample dataset --------------------------------
SAMPLE_DIR="$TESSERA_STATE_DIR/sample"
mkdir -p "$SAMPLE_DIR"
curl -sSL "$SAMPLE_URL" -o "$SAMPLE_DIR/tessera-sample-v1.zip"
(cd "$SAMPLE_DIR" && python3 -c '
import hashlib, json, zipfile
archive = zipfile.ZipFile("tessera-sample-v1.zip")
archive.extractall()
manifest = json.load(open("manifest.json"))
for name, meta in manifest["files"].items():
    digest = hashlib.sha256(open(name, "rb").read()).hexdigest()
    assert digest == meta["sha256"], f"checksum mismatch: {name}"
print("sample dataset verified:", ", ".join(sorted(manifest["files"])))
print("dataset fingerprint:", manifest["fingerprint"])
')

# --- 2. create an upload session and send the bytes --------------------------
SESSION="$(curl -sS -X POST "$TESSERA_API_BASE/uploads" "${AUTH[@]}" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: quickstart-upload-orders" \
  -d '{"filename": "ecommerce_orders.csv", "format": "csv"}')"
UPLOAD_ID="$(printf '%s' "$SESSION" | python3 -c 'import json,sys; print(json.load(sys.stdin)["upload_id"])')"
UPLOAD_URL="$(printf '%s' "$SESSION" | python3 -c 'import json,sys; print(json.load(sys.stdin)["upload_url"])')"

curl -sS -X PUT "$UPLOAD_URL" --data-binary "@$SAMPLE_DIR/ecommerce_orders.csv" > /dev/null
curl -sS -X POST "$TESSERA_API_BASE/uploads/$UPLOAD_ID:complete" "${AUTH[@]}" > /dev/null

# Completing attaches the file to a `file_upload` source (status `completed`)
# and enqueues a landing sync; wait for it before mapping. Poll the upload, not
# only the first sync: if that sync fails before ingesting the file, Tessera
# hands the file to one follow-up sync, and the upload names it.
UPLOAD="$(curl -sS "$TESSERA_API_BASE/uploads/$UPLOAD_ID" "${AUTH[@]}")"
SOURCE_ID="$(printf '%s' "$UPLOAD" | python3 -c 'import json,sys; print(json.load(sys.stdin)["source_id"])')"
LANDING_SYNC_ID="$(printf '%s' "$UPLOAD" | python3 -c 'import json,sys; print(json.load(sys.stdin)["sync_id"])')"
echo "upload status: $(printf '%s' "$UPLOAD" | python3 -c 'import json,sys; print(json.load(sys.stdin)["status"])')"
for _ in $(seq 1 300); do
  LANDING_SYNC_ID="$(curl -sS "$TESSERA_API_BASE/uploads/$UPLOAD_ID" "${AUTH[@]}" \
    | python3 -c 'import json,sys; print(json.load(sys.stdin)["sync_id"])')"
  LANDING="$(curl -sS "$TESSERA_API_BASE/syncs/$LANDING_SYNC_ID" "${AUTH[@]}" \
    | python3 -c 'import json,sys; print(json.load(sys.stdin)["status"])')"
  if [ "$LANDING" = "succeeded" ]; then break; fi
  if [ "$LANDING" = "failed" ]; then
    # The read that saw `failed` may itself have handed the file on: re-read
    # the upload once, and keep polling if it now names another sync.
    OWNER="$(curl -sS "$TESSERA_API_BASE/uploads/$UPLOAD_ID" "${AUTH[@]}" \
      | python3 -c 'import json,sys; print(json.load(sys.stdin)["sync_id"])')"
    if [ "$OWNER" = "$LANDING_SYNC_ID" ]; then break; fi
  fi
  sleep 1
done
echo "landing sync: $LANDING"

# --- 3. map columns to canonical events --------------------------------------
curl -sS -X PUT "$TESSERA_API_BASE/sources/$SOURCE_ID/mapping" "${AUTH[@]}" \
  -H "Content-Type: application/json" \
  --data-binary @mapping-orders.json > /dev/null
echo "mapping stored"

# --- 4. re-sync: normalize every row into canonical events -------------------
SYNC="$(curl -sS -X POST "$TESSERA_API_BASE/sources/$SOURCE_ID/syncs" "${AUTH[@]}" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: quickstart-orders-full-sync" \
  -d '{"mode": "full"}')"
SYNC_ID="$(printf '%s' "$SYNC" | python3 -c 'import json,sys; print(json.load(sys.stdin)["sync_id"])')"
for _ in $(seq 1 300); do
  SYNC="$(curl -sS "$TESSERA_API_BASE/syncs/$SYNC_ID" "${AUTH[@]}")"
  STATUS="$(printf '%s' "$SYNC" | python3 -c 'import json,sys; print(json.load(sys.stdin)["status"])')"
  if [ "$STATUS" = "succeeded" ] || [ "$STATUS" = "failed" ]; then break; fi
  sleep 1
done

printf '%s' "$SYNC" | python3 -c '
import json, sys
sync = json.load(sys.stdin)
keys = ("status", "rows_landed", "rows_rejected", "events_normalized")
print(json.dumps({key: sync[key] for key in keys}, indent=2))
'
printf '%s' "$SOURCE_ID" > "$TESSERA_STATE_DIR/source_id"

Download quickstart/02-upload-events.sh — this exact file runs in CI on every change (hosts shown are for this environment).

The mapping​

The mapping is the contract between your table shape and the canonical event: which column is the entity id, which is the event type, which is the timestamp, and which columns become attributes.

mapping-orders.json
{
  "version": 1,
  "entity": { "type": { "constant": "user" }, "id_column": "user_id" },
  "event_type": { "from_column": "event_type" },
  "timestamp": { "column": "occurred_at" },
  "event_id_column": "event_id",
  "attributes": [
    { "column": "order_value" },
    { "column": "item_count" },
    { "column": "category" },
    { "column": "payment_method" },
    { "column": "channel" },
    { "column": "ticket_topic" },
    { "column": "return_reason" }
  ]
}

Download quickstart/mapping-orders.json — this exact file runs in CI on every change (hosts shown are for this environment).

Worth noting:

  • event_type: {from_column: …} — one CSV carries many event types; a value_map can rename messy source values, and unmapped_policy decides whether unknown types reject or skip the row.
  • Empty cells behave well: a blank ticket_topic on an order row resolves to a presence state, never a silent zero.
  • event_id_column makes re-syncs no-ops — the same rows can never land twice.

The sync report is your verification: rows_landed, rows_rejected (should be 0), events_normalized, and a per-event-type profile that feeds the next step.

Next: approve the schema contract.