Implement incremental sync strategies for Navan BOOKING and TRANSACTION data with ETL pipeline patterns.
Use when setting up production data pipelines, debugging sync drift, or adding real-time event processing.
Trigger with "navan data sync", "navan incremental sync", "navan ETL pipeline".
Instalar com Codex ou Claude Copie este prompt, cole no Codex, Claude ou outro assistente e deixe que ele revise a página da skill e instale para você.
Um comando direto ignora o prompt de revisão. Verifique a origem antes de executá-lo.
Implement incremental sync strategies for Navan BOOKING and TRANSACTION data with ETL pipeline patterns.
Use when setting up production data pipelines, debugging sync drift, or adding real-time event processing.
Trigger with "navan data sync", "navan incremental sync", "navan ETL pipeline".
This skill provides production-grade sync strategies for Navan data. The two primary tables have fundamentally different sync models: BOOKING requires weekly full-refresh with merge-upsert logic (every record is re-imported, keyed by UUID), while TRANSACTION is incremental and append-only. Real-time use cases require webhook callbacks for event-driven processing. This skill covers all three tiers — scheduled full-refresh, incremental watermark-based sync, and real-time webhooks — along with Airbyte connector configuration and idempotent SQL upsert patterns.
Prerequisites
Navan account with OAuth 2.0 API credentials (see navan-install-auth)
Destination warehouse (Snowflake, BigQuery, PostgreSQL, or Redshift)
For managed sync: Airbyte instance (Cloud or OSS) with source-navan v0.0.42+
For webhooks: publicly accessible HTTPS endpoint for callbacks
The BOOKING table is re-imported weekly by Navan. Every record is refreshed, so your sync must use merge-upsert logic to avoid duplicates while capturing updates.
// Ensure loads are idempotent — safe to re-run without side effectsasyncfunctionidempotentLoad(records: any[], tableName: string) {
const batchId = `${tableName}-${newDate().toISOString()}`;
// 1. Write to staging with batch IDconsole.log(`Loading ${records.length} records to ${tableName}_staging (batch: ${batchId})`);
// 2. Merge-upsert from staging to target// Uses UUID as natural key — same record always resolves to same rowconsole.log(`Merging ${tableName}_staging -> ${tableName}`);
// 3. Record batch metadata for auditconsole.log(`Batch ${batchId} complete: ${records.length} records processed`);
return { batchId, recordCount: records.length, status: 'complete' };
}
Output
Successful execution produces:
Full-refresh BOOKING sync with merge-upsert deduplication
Incremental TRANSACTION sync with watermark state tracking
Webhook endpoint for real-time event processing
Configured Airbyte connector with production-ready sync schedule
Sync health monitoring with staleness alerting
Error Handling
Error
HTTP Code
Cause
Solution
Unauthorized
401
Expired or invalid bearer token
Re-authenticate via POST /ta-auth/oauth/token
Rate Limited
429
Too many API requests
Use exponential backoff; increase sync interval
Timeout
504
Full refresh too large
Chunk by date range (30-day windows)
Webhook Sig Invalid
401
Tampered or replayed event
Verify NAVAN_WEBHOOK_SECRET; check clock skew
Duplicate Records
N/A
Missing UUID dedup in BOOKING sync
Apply merge-upsert with ON CONFLICT (uuid)
Sync Drift
N/A
Missed incremental window
Fall back to full refresh; reset watermark
Examples
Python — Incremental TRANSACTION sync with watermark:
import requests
import json
import os
from datetime import datetime
base_url = os.environ.get('NAVAN_BASE_URL', 'https://api.navan.com')
auth = requests.post(f'{base_url}/ta-auth/oauth/token', data={
'grant_type': 'client_credentials',
'client_id': os.environ['NAVAN_CLIENT_ID'],
'client_secret': os.environ['NAVAN_CLIENT_SECRET'],
})
headers = {'Authorization': f'Bearer {auth.json()["access_token"]}'}
# Load watermarktry:
withopen('.navan-sync-state.json') as f:
state = json.load(f)
except FileNotFoundError:
state = {'last_sync_date': '2025-01-01'}
today = datetime.now().strftime('%Y-%m-%d')
resp = requests.get(
f'{base_url}/v1/bookings',
params={'createdFrom': state['last_sync_date'], 'createdTo': today, 'page': 0, 'size': 50},
headers=headers,
).json()
txns = resp['data']
print(f'Fetched {len(txns)} records since {state["last_sync_date"]}')
# Save updated watermarkwithopen('.navan-sync-state.json', 'w') as f:
json.dump({'last_sync_date': today}, f)
