Hi,
Sharing some best practices I developed from successfully building Tulip connectors for Airbyte and FiveTran we solved a couple of use cases
- syncing 10Million(s) of records from Tulip Tables to Snowflake for archiving
- Syncing 10 tables across 40 global instances to Snowflake for production tracking, insights and benchmarking
Airbyte Connector - feat(source-tulip): 🎉 New Source: Initial Release of Tulip Source Connector by eddya101 · Pull Request #73360 · airbytehq/airbyte · GitHub
FiveTran Connector - community_connectors/tulip_interfaces at main · fivetran/community_connectors · GitHub
Best Practices for Syncing Tulip Tables to External Data Sources
Overview
This guide is for integration developers and technical operators building custom connectors to sync Tulip Table data from the Tulip API to external destinations such as data warehouses, databases, or analytics platforms. It covers patterns validated through production connector implementations tested with our product and engineering teams.
Following these practices will help you build reliable, efficient syncs that stay within API limits, avoid data loss, and recover cleanly from failures.
Note: Code snippets use Python and assume Lambda-compatible patterns. Adapt as needed for your runtime environment. All API parameter names and filter syntax should be verified against the Tulip Tables API Reference.
Core Concepts
Before diving into best practices, there are a few Tulip Tables concepts that drive most of the patterns below.
The Sequence Number
Every record in a Tulip Table has a sequence number — a monotonically increasing integer that is incremented each time a record is created or updated. This makes it an ideal cursor for syncing:
-
It is stable and deterministic
-
It uniquely identifies each record mutation
-
It can be used to resume a sync from exactly where it left off
Linked Records
Tulip Tables support linked records (tableLink fields) — fields in one table that reference records in another. Including these fields in a records query causes the Tulip API to resolve the references server-side, creating cascading database joins. See Section 3.5 for how to exclude them using the fields query parameter.
Pagination
The Tulip Tables API returns records in pages. Your connector must handle pagination correctly to ensure all records are captured. Using the sequence number as your page cursor (rather than offset-based pagination) is the recommended approach.
Avoid offset-based pagination for large tables. When you paginate using OFFSET N, the database cannot skip directly to row N — it must scan and discard all preceding rows first. At small offsets this is negligible, but as N grows (e.g. page 500 of a 100-record page size means OFFSET 50000), each successive request becomes progressively slower and places increasing read load on the database. On large tables this can visibly degrade performance for other users of the workspace.
Cursor-based pagination using the sequence number (sequenceNumber > last_seen_seq) avoids this entirely — the database resolves the query via an index lookup regardless of how deep into the table you are.
Best Practices
Summary
3.1 Use the Sequence Number as a Cursor vs. offset parameter ---- Stable, index-backed pagination that won’t skip or duplicate records
3.2 Efficient record syncing with Bootstrap and Incremental modes ---- Full historical load first, then targeted incremental syncs using updated_at + seq
3.3 Respect Rate Limits and Use Exponential Backoff ---- Handle 429 errors gracefully without stalling your sync
3.4 Match Sync Frequency to Your Consumption Cadence ---- Avoid syncing more often than you actually use the data
3.5 Exclude Linked Record Fields Using the Fields Filter ---- tableLink fields cause server-side cascading queries — exclude them at the API level using the fields parameter
3.6 - Implement a Checkpoint-Based Error Recovery to handle sync failures ---- Save progress periodically so failures don’t require a full resync
3.7 – Handle Record Deletion ----The API has no delete event stream — use an archived boolean field in Tulip instead of hard-deleting
End-to-End Sync Flow
The diagram below shows how these practices fit together for a typical connector run.
┌─────────────────────────────────────────────────┐
│ Connector Startup │
│ Load checkpoint from store (3.6) │
│ Checkpoint exists? → Incremental Sync (3.2) │
│ No checkpoint? → Bootstrap Sync (3.2) │
└─────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────┐
│ Fetch Page using Seq Cursor (3.1) │
│ GET /records?lastUpdated > last_sync_date, sort=seq ASC │
│ → Backoff on 429 (3.3) │
│ → Preserve linked record IDs as-is (3.5) │
└─────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────┐
│ Upsert Records to Destination │
│ Save checkpoint every ~500 records (3.6) │
│ No deletes in destination (3.7) │
└─────────────────────────────────────────────────┘
│
No more records returned
│
▼
Save final checkpoint
Sync Complete
3.1 Use the Sequence Number as a Cursor
When paginating through records, use the sequence number as your cursor rather than offset-based pagination.
Why offset pagination breaks: If records are inserted or updated while your sync is running, offset-based pagination can skip or duplicate records. Sequence number-based cursors are stable — sequenceNumber > last_seen_seq always gives you the correct next page regardless of concurrent writes. See Core Concepts: Pagination for why large offsets also place significant load on the database.
def fetch_page(
base_url: str,
table_id: str,
headers: dict,
after_seq: int,
page_size: int = 100
) -> list[dict]:
"""Fetch one page of records, ordered by sequence number, after the given cursor."""
params = {
"sortBy": "sequenceNumber",
"sortDir": "asc",
"filters": [{"field": "sequenceNumber", "operator": "greater than", "value": after_seq}],
"limit": page_size,
}
data = fetch_with_backoff(f"{base_url}/tables/{table_id}/records", headers, params)
return data.get("records", [])
Tip: Always sort ascending by sequence number. This guarantees pages are ordered consistently and your cursor advances forward without gaps.
3.2 Syncing Records: Bootstrap and Incremental
Syncing a table happens in two phases: a one-time bootstrap to load all historical records, followed by ongoing incremental syncs to capture changes. Both use the sequence number as a cursor.
Bootstrap (First Run)
Before running incremental syncs, perform a bootstrap sync — a full load of all historical records. This gives your destination a complete baseline.
Pattern:
-
Start with last_seq = 0 (or last saved checkpoint if resuming an interrupted bootstrap)
-
Fetch pages sorted by sequence number ascending
-
Write each page to the destination
-
Save a checkpoint every ~500 records
-
Stop when a page returns no records
def bootstrap_sync(
base_url: str,
table_id: str,
headers: dict,
checkpoint_store,
page_size: int = 100
) -> None:
"""Full historical load using sequence number as cursor."""
last_seq = checkpoint_store.load(table_id, default=0)
records_since_checkpoint = 0
print(f"Starting bootstrap for table={table_id} from seq={last_seq}")
while True:
records = fetch_page(base_url, table_id, headers, after_seq=last_seq, page_size=page_size)
if not records:
print("Bootstrap complete — no more records.")
break
write_to_destination(records)
last_seq = records[-1]["sequenceNumber"]
records_since_checkpoint += len(records)
if records_since_checkpoint >= 500:
checkpoint_store.save(table_id, last_seq)
print(f"Checkpoint saved at seq={last_seq}")
records_since_checkpoint = 0
checkpoint_store.save(table_id, last_seq)
print(f"Bootstrap finished. Final checkpoint: seq={last_seq}")
Resuming after failure: If the bootstrap is interrupted, load the saved checkpoint and resume from seq > checkpoint_seq. Do not restart from seq = 0 — this avoids re-syncing large volumes of data unnecessarily.
Incremental Sync (Ongoing)
After the bootstrap, run incremental syncs to capture only records that have changed since the last run. Use updated_at to scope the query to a recent time window, and the sequence number to paginate reliably within it.
Why both?
-
updated_at limits the query to recently modified records — efficient for large tables where most records are unchanged
-
Sequence number provides a stable pagination cursor and a precise checkpoint within the result set
def incremental_sync(
base_url: str,
table_id: str,
headers: dict,
checkpoint_store,
last_sync_time: str,
page_size: int = 100
) -> None:
"""
Sync records modified since last_sync_time.
Args:
last_sync_time: ISO 8601 timestamp, e.g. "2024-01-15T10:00:00Z".
Subtract a small buffer (60-120s) to account for clock skew.
"""
last_seq = checkpoint_store.load(table_id, default=0)
records_since_checkpoint = 0
print(f"Incremental sync: table={table_id}, updated_at >= {last_sync_time}, seq > {last_seq}")
while True:
params = {
"sortBy": "sequenceNumber",
"sortDir": "asc",
"filters": [
{"field": "updatedAt", "operator": ">=", "value": last_sync_time},
{"field": "sequenceNumber", "operator": "greater than", "value": last_seq},
],
"limit": page_size,
}
data = fetch_with_backoff(
f"{base_url}/tables/{table_id}/records", headers, params
)
records = data.get("records", [])
if not records:
break
write_to_destination(records) # use upsert logic in the destination
last_seq = records[-1]["sequenceNumber"]
records_since_checkpoint += len(records)
if records_since_checkpoint >= 500:
checkpoint_store.save(table_id, last_seq)
print(f"Checkpoint saved at seq={last_seq}")
records_since_checkpoint = 0
checkpoint_store.save(table_id, last_seq)
print("Incremental sync complete.")
Clock skew buffer: Subtract 60–120 seconds from last_sync_time when constructing the filter. This ensures records written just before the previous sync window closes are not missed due to small timing differences between systems. Your destination’s upsert logic will handle any resulting duplicates safely.
3.3 Respect Rate Limits and Use Exponential Backoff
The Tulip Tables API enforces rate limits on all requests. There are two layers to handling this correctly:
-
Proactive throttling — space your requests to stay within the limit from the start, rather than waiting to be rejected
-
Reactive backoff — when a 429 is received despite throttling (e.g. due to other concurrent processes sharing the same workspace API quota), back off and retry
See the Tulip API Rate Limits guide for the current limits. Do not hard-code limit values in your connector — use a named constant so they are easy to update if limits change.
Guidelines:
-
Instantiate a single RateLimiter per sync run and pass it through — do not create one per request
-
When you receive a 429, pause and retry with exponential backoff
-
Add random jitter to backoff to prevent multiple concurrent syncs all retrying at the same moment
-
Check for a Retry-After header on 429 responses — use it if present, as it overrides the calculated wait
import time
import random
import requests
# Set to match the limit in https://support.tulip.co/docs/api-rate-limits
REQUESTS_PER_MINUTE = 50
class RateLimiter:
"""Proactive throttler that enforces a minimum interval between requests."""
def __init__(self, requests_per_minute: int = REQUESTS_PER_MINUTE):
self._min_interval = 60.0 / requests_per_minute
self._last_request_at = 0.0
def wait(self) -> None:
"""Block until it is safe to make the next request."""
elapsed = time.monotonic() - self._last_request_at
remaining = self._min_interval - elapsed
if remaining > 0:
time.sleep(remaining)
self._last_request_at = time.monotonic()
def fetch_with_backoff(
url: str,
headers: dict,
params: dict,
rate_limiter: RateLimiter,
max_retries: int = 5,
) -> dict:
"""Make a GET request with proactive rate limiting and exponential backoff on 429s."""
for attempt in range(max_retries):
rate_limiter.wait() # enforce minimum interval before every request
response = requests.get(url, headers=headers, params=params)
if response.status_code == 200:
return response.json()
if response.status_code == 429:
retry_after = response.headers.get("Retry-After")
wait = float(retry_after) if retry_after else (2 ** attempt) + random.uniform(0, 1)
print(f"Rate limited. Retrying in {wait:.1f}s (attempt {attempt + 1}/{max_retries})")
time.sleep(wait)
continue
response.raise_for_status()
raise RuntimeError(f"Max retries exceeded for {url}")
Usage — instantiate once per sync run and pass through to all fetch calls:
rate_limiter = RateLimiter(requests_per_minute=REQUESTS_PER_MINUTE)
bootstrap_sync(base_url, table_id, headers, checkpoint_store, rate_limiter)
# or
incremental_sync(base_url, table_id, headers, checkpoint_store, last_sync_time, rate_limiter)
Why not just rely on backoff? Backoff is a recovery mechanism — it kicks in after you have already exceeded the limit and received an error. On a long sync with many pages, repeatedly hitting the limit and waiting adds significant latency and puts unnecessary load on the API. Proactive throttling keeps your connector within the allowed rate continuously, without needing to back off under normal conditions.
3.4 Match Sync Frequency to Your Consumption Cadence
As long as your connector stays within the Tulip API rate limits, there is no hard rule on how often you sync. However, syncing more frequently than you actually consume the data creates unnecessary work — both for the API and for your destination.
The core issue: The incremental sync pattern captures every intermediate state of a record between syncs. If a record is updated 20 times between two syncs, you will write 20 rows to your destination. If your downstream query only ever reads the latest state (e.g. a daily report), then 19 of those writes are wasted — extra API calls, extra destination writes, and extra data to filter through.
Rule of thumb: Set your sync interval to match how fresh the data actually needs to be for your use case.
Consumption pattern
Recommended sync interval
- Daily batch reports or exports - Once daily, shortly before the report window
- Hourly dashboards or summaries - Once per hour
- Operational monitoring, near-real-time - Every 15 minutes
- True real-time requirements - Consider other methods vs polling the Table API
Consider per-table frequency. Not all tables change at the same rate. A table tracking active job status may update hundreds of times per hour, while a reference table of machine configurations may change once a week. Syncing all tables on the same schedule is simple to implement but wasteful. Where possible, configure higher-volatility tables to sync more frequently and low-volatility tables less often.
# Example: per-table sync interval configuration
SYNC_INTERVALS = {
"active_jobs": timedelta(minutes=5), # high volatility
"machine_output": timedelta(minutes=15), # medium volatility
"machine_config": timedelta(hours=6), # low volatility
"user_assignments": timedelta(hours=24), # rarely changes
}
def should_sync(table_id: str, last_sync_time: datetime) -> bool:
interval = SYNC_INTERVALS.get(table_id, timedelta(hours=1))
return datetime.utcnow() - last_sync_time >= interval
Sanity-check your API budget. Before setting your schedule, estimate the total number of API calls your connector will make per hour:
calls per hour ≈ (average pages per table × number of tables) × syncs per hour
Ensure this stays comfortably within the rate limit documented in the Tulip API Rate Limits guide. Leave headroom — other processes may also be calling the API on behalf of your workspace.
3.5 Exclude Linked Record Fields Using the Fields Filter
Tulip Tables support linked record fields (tableLink type) — fields in one table that reference records in another. When a tableLink field is included in a records response, the Tulip API has to resolve those references server-side, which creates cascading database queries. On tables with many linked fields or large record counts, this places significant load on the database and affects all users of the workspace.
The fix: use the fields query parameter to explicitly list the fields you want returned, and exclude tableLink fields entirely. This requires a small amount of upfront work — you need to fetch the table schema first to know which fields to include — but it is the correct approach and results in faster, lighter queries.
Pattern:
-
Call GET /tables/{tableId} to retrieve the table schema and its column definitions
-
Inspect each field’s dataType.type — skip any with type “tableLink”
-
Always include the system fields required for sync (id, _createdAt, _updatedAt, _sequenceNumber)
-
Pass the resulting list as the fields parameter on every records request
import json
SYSTEM_FIELDS = ["id", "_createdAt", "_updatedAt", "_sequenceNumber"]
def get_allowed_fields(
base_url: str, table_id: str, headers: dict, rate_limiter: RateLimiter
) -> list[str]:
"""
Fetch table schema and return field IDs, excluding tableLink fields.
Always includes system fields required for sync operations.
"""
data = fetch_with_backoff(
f"{base_url}/tables/{table_id}", headers, params={}, rate_limiter=rate_limiter
)
allowed = SYSTEM_FIELDS.copy()
excluded = []
for field in data.get("columns", []):
field_id = field["name"]
field_type = field.get("dataType", {}).get("type")
if field_id in SYSTEM_FIELDS:
continue # already included
if field_type == "tableLink":
excluded.append(field_id)
continue
allowed.append(field_id)
if excluded:
print(f"Excluding {len(excluded)} tableLink field(s): {excluded}")
return allowed
Then pass the result as the fields parameter on every records request:
# At the start of your sync, fetch the schema once
allowed_fields = get_allowed_fields(base_url, table_id, headers, rate_limiter)
fields_param = json.dumps(allowed_fields)
# Include fields on every page fetch
params = {
"sortBy": "sequenceNumber",
"sortDir": "asc",
"filters": [...],
"limit": page_size,
"fields": fields_param, # only return non-tableLink fields
}
Why not just ignore linked record values after fetching them? The database load happens at query time on the Tulip side — the joins are performed before the response is sent. Filtering the response after the fact does nothing to reduce that load. The fields parameter prevents the joins from being executed at all.
3.6 Checkpoint-Based Error Recovery
Network failures, timeouts, and unexpected errors will happen. Design your connector to recover gracefully without restarting from scratch.
Pattern:
-
Persist the last successfully processed sequence number to a durable store after every ~500 records
-
On restart, load the checkpoint and resume from seq > checkpoint_seq
-
Write records using upsert logic (not inserts) in the destination so re-processing records from the checkpoint is safe
Lambda-compatible checkpoint store using S3:
i
mport json
import boto3
class S3CheckpointStore:
"""Persistent checkpoint store backed by S3. Safe for Lambda use."""
def __init__(self, bucket: str, key: str):
self.bucket = bucket
self.key = key
self.s3 = boto3.client("s3")
self._data = self._load()
def _load(self) -> dict:
try:
obj = self.s3.get_object(Bucket=self.bucket, Key=self.key)
return json.loads(obj["Body"].read())
except self.s3.exceptions.NoSuchKey:
return {}
def save(self, table_id: str, seq: int) -> None:
self._data[table_id] = seq
self.s3.put_object(
Bucket=self.bucket,
Key=self.key,
Body=json.dumps(self._data).encode()
)
def load(self, table_id: str, default: int = 0) -> int:
return self._data.get(table_id, default)
Why ~500 records? This balances checkpoint frequency against overhead. Checkpointing every single record adds unnecessary write cost; checkpointing too infrequently means a failure could force you to re-process a large batch. Adjust the interval based on your record size and destination write latency.
3.7 Handling Record Deletion
The Tulip Tables API does not provide a stream of delete events. If a record is hard-deleted in Tulip, the connector has no way to detect it — the record simply disappears from the API with no notification. How you handle this depends on what you want the external destination to do.
Scenario A: Keep the destination in sync with Tulip (mirror deletions)
If your goal is for the destination to reflect the current state of Tulip — including deletions — do not hard-delete records in Tulip. The connector cannot propagate a deletion it never sees.
Instead, add a boolean column to the table — archived or deleted — and set it to true when a record should be treated as inactive. Because this is a field update, the next incremental sync will pick it up and write it to the destination automatically, with no special connector logic required. You can then filter them out at the destination.
In Tulip: add an archived boolean column and update it via your app or an automation rather than deleting the record.
In the destination: filter on archived = false in your queries and reports rather than assuming all synced records are active.
Scenario B: Clean up Tulip for performance, keep records in the destination
Tulip Tables perform better when kept lean. If a table grows very large, hard-deleting old or completed records is a valid way to manage its size — for example, archiving completed work orders after a retention period.
In this scenario, hard-deleting in Tulip is fine. Because the connector never receives a delete event, records removed from Tulip are simply never updated again in the destination — they are retained as-is. The destination becomes a long-term archive while Tulip stays performant.
No connector changes are needed for this pattern. Just ensure downstream consumers understand that the destination may contain records that no longer exist in Tulip, and design their queries accordingly (e.g. join on a completed_at timestamp rather than assuming the record is still active in Tulip).