File upload connector

Resumable CSV/Parquet uploads into a file_upload source — the fastest path from "I have an extract" to canonical events, and what the quickstart uses. Files land in Tessera's data bucket (never a customer system), are fingerprinted on completion, and sync through the same landing → profiling → normalization pipeline as every warehouse connector.

Prerequisites

An API key with the admin scope (source + upload management).
A CSV or Parquet file ≤ 50 GiB. CSV dialects are sniffed on completion; override via the source's config.csv if your file is exotic.

Create the credential

None needed — uploads use your API key alone. There is no customer-side system to grant access to.

Create the source

You can POST /sources with kind: file_upload explicitly, or skip it: completing an upload without a source_id creates the source for you (the sample below relies on that). Attach later files to the same source by passing its source_id to POST /uploads — each completion becomes a new snapshot of that source.

{
  "name": "media-events",
  "kind": "file_upload",
  "config": {}
}

Map columns, test, and run the first sync

The end-to-end flow — session, bytes, complete, mapping, sync — using the sample dataset's media-streaming CSV:

01-upload-and-sync.sh
#!/usr/bin/env bash
# File upload, end to end: session -> bytes -> complete -> mapping -> sync.
# Uses the media-streaming CSV of the sample dataset.
set -euo pipefail

: "${TESSERA_API_BASE:?e.g. https://api.staging.tessera.zone/v1}"
: "${TESSERA_API_KEY:?your tsk_test_… key}"
: "${TESSERA_STATE_DIR:=.}"
SAMPLE_URL="${TESSERA_SAMPLE_URL:-https://docs.staging.tessera.zone/downloads/tessera-sample-v1.zip}"
AUTH=(-H "Authorization: Bearer $TESSERA_API_KEY")

curl -sSL "$SAMPLE_URL" -o "$TESSERA_STATE_DIR/sample.zip"
python3 -c 'import sys, zipfile; zipfile.ZipFile(sys.argv[1]).extract("media_streaming.csv", sys.argv[2])' \
  "$TESSERA_STATE_DIR/sample.zip" "$TESSERA_STATE_DIR"

# 1. Create the upload session (a resumable URL, valid for 24 h).
SESSION="$(curl -sS -X POST "$TESSERA_API_BASE/uploads" "${AUTH[@]}" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: docs-file-upload-media" \
  -d '{"filename": "media_streaming.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"])')"

# 2. Send the bytes, then complete (verifies size, fingerprints the object,
#    attaches it to a `file_upload` source, and lands + profiles it).
curl -sS -X PUT "$UPLOAD_URL" --data-binary "@$TESSERA_STATE_DIR/media_streaming.csv" > /dev/null
curl -sS -X POST "$TESSERA_API_BASE/uploads/$UPLOAD_ID:complete" "${AUTH[@]}" > /dev/null
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: $(printf '%s' "$UPLOAD" | python3 -c 'import json,sys; print(json.load(sys.stdin)["status"])')"
for _ in $(seq 1 300); do
  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" ] || [ "$LANDING" = "failed" ]; then break; fi
  sleep 1
done
echo "landing sync: $LANDING"

# 3. Map columns to canonical events, then re-sync to normalize.
curl -sS -X PUT "$TESSERA_API_BASE/sources/$SOURCE_ID/mapping" "${AUTH[@]}" \
  -H "Content-Type: application/json" --data-binary @mapping-media.json > /dev/null
SYNC="$(curl -sS -X POST "$TESSERA_API_BASE/sources/$SOURCE_ID/syncs" "${AUTH[@]}" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: docs-file-upload-media-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)
print("sync: %s rows_landed=%s events_normalized=%s rejected=%s"
      % (sync["status"], sync["rows_landed"], sync["events_normalized"], sync["rows_rejected"]))
'

Download connect/file-upload/01-upload-and-sync.sh — this exact file runs in CI on every change (hosts shown are for this environment).

The mapping (see the quickstart's walkthrough for field-by-field commentary):

mapping-media.json
{
  "version": 1,
  "entity": { "type": { "constant": "viewer" }, "id_column": "viewer_id" },
  "event_type": { "from_column": "event_type" },
  "timestamp": { "column": "occurred_at" },
  "event_id_column": "event_id",
  "attributes": [
    { "column": "genre" },
    { "column": "device" },
    { "column": "minutes_watched" },
    { "column": "plan" }
  ]
}

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

Verification is the sync report: rows_landed = your row count, rows_rejected = 0, events_normalized > 0, plus a per-event-type profile. GET /syncs/{sync_id}/diagnostics accounts for every row exactly once — rows_read = rows_accepted + rows_rejected + rows_skipped + rows_duplicate (a row repeating an earlier row's event_id is a duplicate and is committed once) — and names the request_id of the call that started the sync. GET /syncs/{sync_id}/rejections lists each rejected row with its typed reason, offending column and landed row number; the console shows the same at /sources/{source_id}?sync={sync_id}. GET /uploads/{id} also serves a schema preview inferred from the landed snapshot — useful before you write the mapping.

Which sync ingests an upload

Poll the upload, not only the sync complete named: GET /uploads/{id} returns sync_id, the sync that now ingests the file. That is the sync enqueued at completion, unless it failed before reading the file: then a follow-up sync takes the file over, and the failed sync's message says so. When you start a sync of the source, or its schedule does, before that follow-up has started, your sync takes the file instead and the follow-up is deleted (GET /syncs/{id} answers 404 for it).

Uploading the identical file again under the same name (the same size and checksums) depends on the sync that ingests the earlier upload:

It is still queued or running (any sync): the upload attaches nothing, complete names that sync, and the content lands once.
It took the file over and succeeded (a follow-up sync, or your sync or a scheduled one that took the file from a follow-up): the same, so re-uploading after seeing the first sync fail never ingests the file twice. If a project purge or an entity erasure in the project completed after that sync finished, the upload is a new file instead.
Any other sync that has ended (the one enqueued at completion, or one that failed): the upload is a new file and is ingested again.

An erasure never brings erased rows back: its tombstones keep them out of any later ingest. To re-ingest a source's files on purpose, start a mode=full sync of the source.

A changed file under the same name replaces an earlier upload of it that no sync has read yet.

Automatic interpretation or a manual mapping

A source in automatic or guided mode interprets landed rows without a mapping. Each successful sync hands its landed snapshot to automatic interpretation, and that handoff is recorded durably: a failed attempt is retried with a backoff, and a handoff that keeps failing stops with a typed reason instead of waiting forever.

GET /sources/{source_id}/onboarding-diagnostics explains where a source stands:

governed_by: what actually governs the source, by the same rule the interpretation worker applies: project_fenced, source_unavailable (disabled, errored or deleted), manual_mapping, manual_mode, or automatic;
intake: the newest snapshot handoff. Its state is pending, retrying, complete, failed, stopped, superseded or not_recorded, with attempts, last_error_code, last_error_message and next_attempt_at. A sync that succeeded moments ago reads pending (with recorded_at null) while its handoff is still on its way; only after that window is an unrecorded upload reported as not_recorded;
preparation: the automatic model preparation, with its state, reason and next_attempt_at;
settled: true once nothing changes without new data or an action;
resume_available: true only when POST /sources/{source_id}/onboarding:resume would retry a failed or stopped handoff or record a lost one. Check it before offering a resume: some stops can never be retried (a saved mapping, a disabled or deleted source, a project being deleted, or an unreadable upload from before the durable handoff existed).

The diagnostics are a pure read: they never change the source or its onboarding.

Saving a mapping takes the source under manual control. When you store a column or batch-catalog mapping with PUT /sources/{source_id}/mapping, the source's onboarding mode becomes manual and pending snapshot handoffs stop with reason manual_mapping. Syncs then normalize rows through your mapping, and automatic interpretation stops for that source. Mappings are immutable versions, so a mapped source stays manual: switching it back to automatic or guided, through PATCH /sources/{source_id}/onboarding or PATCH /sources/{source_id}, returns failed_precondition with reason manual_mapping_governs. Raw JSON sent to a mapped source with POST /sources/{source_id}/raw-events returns failed_precondition with reason manual_source; send its data through syncs instead. To use automatic interpretation for the same data, upload it to a new source and don't save a mapping there.

Sources that saved a mapping before this behaviour existed are brought under manual control once, by an audited operator backfill. Until then their diagnostics already report governed_by: manual_mapping.

Troubleshooting

Symptom	Cause	Fix
`failed_precondition: upload_url expired`	sessions live 24 h	create a new session; re-PUT
`complete` fails with size mismatch	truncated PUT	re-PUT the full file to the same URL
`mapping_rejection_rate_exceeded` on sync	> 5% of rows failed the mapping	`GET /syncs/{sync_id}/rejections` names each rejected row, its reason and column; fix those rows or the mapping (`mapping:preview` shows per-row errors), then re-sync
rows rejected as `invalid_timestamp`	naive/local timestamps	export RFC 3339 UTC or set `timestamp.format`
sync `succeeded` but `events_normalized: 0`	no mapping stored	`PUT /sources/{id}/mapping`, then re-sync
automatic interpretation shows `failed` or `stopped` in `onboarding-diagnostics`	the snapshot handoff hit a typed failure, or manual control took over	read `intake.last_error_code` and `intake.stop_reason`; after fixing the cause, resume with `POST /sources/{id}/onboarding:resume` only when `resume_available` is `true`
`failed_precondition: manual_mapping_governs` when switching to automatic or guided	the source has a saved mapping	upload to a new source for automatic interpretation
`failed_precondition: manual_source` from `raw-events`	the source is under manual control, for example because it has a saved mapping	send the data through a sync, or use a source without a mapping

Prerequisites​

Create the credential​

Create the source​

Map columns, test, and run the first sync​

Which sync ingests an upload​

Automatic interpretation or a manual mapping​

Troubleshooting​