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".
Installer avec Codex ou Claude Copiez ce prompt, collez-le dans Codex, Claude ou un autre assistant, puis laissez-le vérifier la page du skill et l'installer pour vous.
Une commande directe contourne le prompt de vérification. Examinez la source avant de l'exécuter.
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)
