# Data Sync Patterns

> Use these patterns to move deals, CRM records, and related objects into your own warehouse or application safely.

Source: https://www.lev.com/docs/build/data-sync-patterns

Last updated: May 2026

---
## Sync Modes

**Choose the sync mode deliberately**

- **Backfill:** Use cursor pagination to walk the full dataset deterministically when you are loading historical records.
- **Incremental:** Use filters like updated_at together with stored checkpoints to pull only what changed since the last successful run.
- **On-demand refresh:** Use resource-specific reads when a user opens a detail view and needs the freshest known state.

## Backfill Pattern

Start with the smallest set of fields that proves the pipeline works.

Use cursor pagination wherever the endpoint supports it.

Persist both the downstream write result and the last successful cursor checkpoint.

Only after the happy path is stable, widen the field set.

**Backfill pseudocode**

### TypeScript

```typescript
let cursor: string | null = null

while (true) {
  const url = new URL("https://api.lev.com/api/external/v2/deals")
  url.searchParams.set("limit", "100")
  if (cursor) url.searchParams.set("cursor", cursor)

  const response = await fetch(url, {
    headers: {
      Authorization: "Bearer YOUR_API_KEY",
      "X-Origin-App": "warehouse-sync",
    },
  })

  const payload = await response.json()
  await writeBatch(payload.data)

  if (!payload.pagination?.has_more) break
  cursor = payload.pagination.next_cursor
}
```

### Python

```python
cursor = None

while True:
    params = {"limit": 100}
    if cursor:
        params["cursor"] = cursor

    response = requests.get(
        "https://api.lev.com/api/external/v2/deals",
        params=params,
        headers={
            "Authorization": "Bearer YOUR_API_KEY",
            "X-Origin-App": "warehouse-sync",
        },
    )

    payload = response.json()
    write_batch(payload["data"])

    if not payload.get("pagination", {}).get("has_more"):
        break

    cursor = payload["pagination"]["next_cursor"]
```

## Incremental Pattern

Use an `updated_at` or similar time-based filter together with a persisted checkpoint:

- Store the timestamp of the last fully successful run.
- Re-read a small overlap window to tolerate clock skew and delayed writes.
- Deduplicate in your destination using resource IDs.

  Offset pagination is appropriate for sorted browsing, not for durable bulk syncs. For production data movement, prefer cursor pagination whenever the endpoint supports it.

## Operational Checklist

- Monitor `request_id` values for failed batches.
- Alert on repeated `401`, `403`, and `429` responses.
- Keep writes idempotent in your destination so replaying a batch is safe.
- Version your destination schema deliberately as Lev fields expand.