> ## Documentation Index
> Fetch the complete documentation index at: https://docs.sequenzy.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Data Pipeline Ingestion

> Ingest contacts and events from your data warehouse or pipeline (Airflow, Dagster, dbt, cron) at scale, with built-in deduplication

# Ingesting Data from Your Pipeline

You don't need a native warehouse connector to keep Sequenzy in sync with
Snowflake, BigQuery, Redshift, or any other source of truth. The API is
designed so an existing DAG - an Airflow task, a Dagster job, a dbt
post-hook, or a plain cron script - can push contacts and events in bulk,
re-run safely after failures, and never create duplicates.

This guide covers the endpoints a pipeline needs, how deduplication works on
each of them, and how to structure a task that moves large volumes of data
within the API rate limits.

## Endpoints at a Glance

| Purpose                           | Endpoint                                                                                | Batch size    |
| --------------------------------- | --------------------------------------------------------------------------------------- | ------------- |
| Bulk import/update contacts       | [POST /api/v1/subscribers/imports](/api-reference/subscribers/import-create)            | 5,000 records |
| Check import progress             | [GET /api/v1/subscribers/imports/\{importId}](/api-reference/subscribers/import-status) | -             |
| Upsert a single contact           | [POST /api/v1/subscribers](/api-reference/subscribers/create)                           | 1 record      |
| Bulk record events, many contacts | [POST /api/v1/subscribers/events/imports](/api-reference/subscribers/events/import)     | 25 events     |
| Record an event                   | [POST /api/v1/subscribers/events](/api-reference/subscribers/events/trigger)            | 1 event       |
| Record many events (one contact)  | [POST /api/v1/subscribers/events/bulk](/api-reference/subscribers/events/trigger-bulk)  | 500 events    |
| Bulk tag changes                  | [POST /api/v1/subscribers/bulk/tags/add](/api-reference/subscribers/tags/bulk-add)      | -             |
| Export metrics to your warehouse  | [GET /api/v1/metrics](/api-reference/analytics/metrics)                                 | -             |

All endpoints use [API key authentication](/authentication) and share the
[rate limits](/api-reference/introduction#rate-limiting) of 100 requests per
minute and 20 requests per second per key.

## Why Re-Runs Are Safe

Every ingestion path deduplicates, so your retry strategy can simply be
"re-run the whole task":

* **Contacts** match on `email`, `externalId`, or `phone`. You choose what
  happens on a match with `duplicateStrategy`: `skip` (default), `merge`, or
  `overwrite`. Running the same import twice never creates a second contact.
* **Events** accept your own `eventId` as an idempotency key. A retried event
  with the same ID for the same contact and event name is acknowledged with
  `duplicate: true` and records nothing.
* **Historical contacts** are those whose rows in a request are all more than
  an hour old. Their events appear in analytics and segments without firing
  automations or double-opt-in mail. If any row for a contact is recent, that
  contact's whole group is live; split live and historical extracts when you
  need the silent path.

## Syncing Contacts at Scale

Use the [bulk import endpoint](/api-reference/subscribers/import-create) with
5,000-record batches. At the standard rate limit that sustains roughly 500,000
contact rows per minute, so contact volume is effectively unconstrained for a
scheduled sync.

A typical incremental task:

1. Query your warehouse for rows changed since the last successful run (a
   simple `updated_at > :watermark` works; you own the watermark).
2. Chunk the rows into batches of up to 5,000.
3. POST each batch with `duplicateStrategy: "merge"` so existing contacts are
   updated in place.
4. Poll each returned import ID until `status` is `completed`, then advance
   your watermark.

```python theme={null}
import time
import requests

API = "https://api.sequenzy.com/api/v1"
HEADERS = {"Authorization": "Bearer YOUR_API_KEY"}
BATCH_SIZE = 5000

def sync_contacts(rows, run_id):
    import_ids = []
    for i in range(0, len(rows), BATCH_SIZE):
        batch = rows[i : i + BATCH_SIZE]
        resp = requests.post(
            f"{API}/subscribers/imports",
            headers=HEADERS,
            json={
                "duplicateStrategy": "merge",
                "optInMode": "confirmed",
                # The key is scoped to the request content: retrying the same
                # batch after a timeout replays the original import, while a
                # corrected batch under the same key imports as a new one.
                "idempotencyKey": f"{run_id}-batch-{i // BATCH_SIZE}",
                "subscribers": [
                    {
                        "email": row["email"],
                        "externalId": row["customer_id"],
                        "firstName": row["first_name"],
                        "customAttributes": {
                            "plan": row["plan"],
                            "mrr": row["mrr"],
                        },
                        "createdAt": row["signed_up_at"],
                    }
                    for row in batch
                ],
            },
            timeout=60,
        )
        if resp.status_code in (429, 503):
            time.sleep(int(resp.headers.get("Retry-After", "5")))
            continue  # rate limit or overlapping enqueue; the keyed retry is safe
        resp.raise_for_status()
        import_ids.append(resp.json()["import"]["id"])
    return import_ids
```

Instead of polling, you can subscribe an
[outbound webhook](/integrations/outbound-webhooks) to
`subscriber_import.completed`. It has one stable logical event ID per import,
and transient enqueue failures are repaired automatically. HTTP delivery is
retried, so consumers should deduplicate by event ID as usual.
If you prefer polling, confirm every import finished cleanly before advancing
your watermark:

```python theme={null}
def wait_for_imports(import_ids, timeout_seconds=1800):
    for import_id in import_ids:
        deadline = time.monotonic() + timeout_seconds
        while True:
            data = requests.get(
                f"{API}/subscribers/imports/{import_id}", headers=HEADERS, timeout=30
            ).json()["import"]
            # Imports report status "running", "completed", or "blocked" -
            # there is no "failed" status; row failures appear in failedCount
            # on a completed import.
            if data["status"] in ("completed", "blocked"):
                break
            if time.monotonic() > deadline:
                raise TimeoutError(f"{import_id} still {data['status']}")
            time.sleep(5)
        if data["status"] == "blocked":
            raise RuntimeError(f"{import_id} was blocked pending review")
        if data.get("failedCount"):
            raise RuntimeError(f"{import_id}: {data['failedReasons']}")
```

<Note>
  Counts on the status endpoint reconcile exactly: `addedCount + updatedCount +
      skippedCount + failedCount == totalRows`. Inspect `skippedReasons` and
  `failedReasons` before treating a run as successful - a `duplicate_csv_row`
  skip means two rows in your own extract carried the same email, phone, or
  external ID.
</Note>

### Preserve Signup Dates

Pass each contact's original signup date in `createdAt`. Date-relative
segments and analytics then reflect real history instead of the day your
pipeline first ran. An existing contact's date only ever moves earlier, so
repeated syncs cannot corrupt it.

### Do Not Enroll Backfills in Automations

Bulk imports do not fire signup automations by default. Leave
`enrollInSequences` off for warehouse syncs unless you explicitly want
imported contacts to enter matching sequences.

## Recording Events at Scale

For behavioral data (purchases, product usage, lifecycle events), use the
[events import endpoint](/api-reference/subscribers/events/import): bounded
batches of up to 25 events per request, each carrying its own contact identity.
Every event must have a deterministic `eventId` derived from your source data - an order ID, a
warehouse row key, or a hash of the row - so re-running the task records
nothing twice:

```python theme={null}
EVENT_BATCH_SIZE = 25

def sync_events(rows):
    for i in range(0, len(rows), EVENT_BATCH_SIZE):
        batch = rows[i : i + EVENT_BATCH_SIZE]
        resp = requests.post(
            f"{API}/subscribers/events/imports",
            headers=HEADERS,
            json={
                "events": [
                    {
                        # null identifiers are accepted as absent. An
                        # externalId-only row must already exist from contact sync.
                        "externalId": row["customer_id"],
                        "email": row["email"],
                        "name": "warehouse.purchase",
                        "eventId": row["order_id"],
                        "occurredAt": row["ordered_at"],
                        # Row data must live inside properties - keys outside
                        # the documented fields are ignored.
                        "properties": {"amount": row["amount"], "sku": row["sku"]},
                    }
                    for row in batch
                ],
            },
            timeout=120,
        )
        if resp.status_code == 429:
            time.sleep(int(resp.headers.get("Retry-After", "5")))
            continue
        resp.raise_for_status()
        summary = resp.json()
        if summary["failed"]:
            raise RuntimeError(f"{summary['failed']} events failed: {summary['failures'][:5]}")
        if summary["sideEffectFailed"]:
            raise RuntimeError(
                f"{summary['sideEffectFailed']} event receipts need downstream recovery: "
                f"{summary['sideEffectFailures'][:5]}"
            )
```

At 25 events per request within the standard request limit, that sustains up to
2,500 events per minute while keeping each synchronous request bounded. Only a
contact whose rows in the request are all more than an hour old uses the silent
historical path; any recent row makes that contact's whole group live. Separate
live and historical rows when a backfilled purchase must not trigger a
"thanks for your order" sequence. For single fresh events (a checkout
happening right now) where you do want automations, use the
[single-event endpoint](/api-reference/subscribers/events/trigger).

## Handling Rate Limits

When you exceed the limit the API returns `429 Too Many Requests` with a
`Retry-After` header. An overlapping subscriber-import request using the same
idempotency key can similarly return a retryable `503` while the first request
still owns its short enqueue lease. For pipeline work:

* Honor `Retry-After` on `429` and retryable `503` responses, then resend the same request. Idempotency makes the
  resend safe even if the original was actually processed.
* Keep the task single-threaded or coarsely parallel. One worker sending
  5,000-row import batches already saturates the useful throughput; extra
  concurrency only spends the budget faster.
* Fail the task on anything other than `429` and re-run it from the top. You
  do not need checkpoint bookkeeping for correctness - only for speed.

## Syncing Data Out

For the reverse direction - warehouse tables fed from Sequenzy - schedule a
task that pulls [aggregated metrics](/api-reference/analytics/metrics), or
subscribe to [outbound webhooks](/integrations/outbound-webhooks) and land the
payloads in your ingestion bucket.

## Checklist

1. Store the API key in your pipeline's secret manager, never in DAG code.
2. Sync contacts with 5,000-record batches and `duplicateStrategy: "merge"`.
3. Pass an `idempotencyKey` per batch so timed-out requests can be resent.
4. Pass `externalId` on every record so identity survives email changes.
5. Pass `createdAt` (contacts) and `occurredAt` (events) from source data.
6. Derive `eventId` deterministically from your source rows.
7. Subscribe a webhook to `subscriber_import.completed`, or poll import
   status, and reconcile counts before advancing your watermark.
8. On failure, re-run with the same IDs: receipts deduplicate and downstream
   recovery is re-attempted idempotently.