Data Sources & Extraction

VirtuousAI uses dlt (data load tool) under the hood for reliable data extraction. This guide covers how the dlt_extract action works and the available data sources.

How dlt_extract Works

The dlt_extract action:

1. Resolves credentials from the specified connection
2. Builds a dlt pipeline targeting your bronze data layer
3. Extracts resources one at a time with checkpointing between each
4. Writes Parquet files to S3 with Hive-style partitioning
5. Tracks state for incremental extraction on subsequent runs

Per-Resource Checkpointing

Resources are extracted sequentially with state committed after each:

This means if extraction is interrupted (worker crash, cancellation), it resumes from the last completed resource — not from the beginning.

Available Sources

Shopify

Extract e-commerce data from Shopify stores.

Resources:

Default Resources: orders, customers, products, locations

Configuration:

{
  "kind": "dlt_extract",
  "connectionRef": { "slug": "shopify" },
  "definition": {
    "source": "shopify",
    "resources": ["orders", "products", "locations"],
    "start_date": "2026-01-01",
    "source_config": {
      "items_per_page": 250
    }
  }
}

Source Config Options:

Incremental Extraction:

• Orders: Uses updated_at cursor, sliced into 7-day windows for large stores
• Products: Uses updated_at cursor
• Customers: Uses updated_at cursor, sliced into 7-day windows for large stores
• Locations: Full sync (reference data, typically under 100 records)

REST API (Generic)

Extract data from any REST API endpoint.

Configuration:

{
  "kind": "dlt_extract",
  "connectionRef": { "id": "conn_rest_api" },
  "definition": {
    "source": "rest_api",
    "resources": ["customers"],
    "source_config": {
      "base_url": "https://api.example.com/v1",
      "endpoints": {
        "customers": {
          "path": "/customers",
          "paginate": {
            "type": "cursor",
            "cursor_param": "after"
          }
        }
      }
    }
  }
}

Klaviyo

Extract email and SMS marketing data from Klaviyo.

Resources:

Default Resources: profiles, events, lists

Configuration:

{
  "kind": "dlt_extract",
  "connectionRef": { "slug": "klaviyo" },
  "definition": {
    "source": "klaviyo",
    "resources": ["profiles", "events", "lists"],
    "start_date": "2026-01-01"
  }
}

Credentials Required:

• api_key — Klaviyo Private API Key (starts with pk_)

Incremental Extraction:

• Profiles: Uses updated cursor for changes
• Events: Uses datetime cursor for new events
• Lists: Uses updated cursor for changes
• Other resources: Full sync (reference data)

Amazon Seller (SP-API)

Extract e-commerce data from Amazon Seller Central accounts using the Selling Partner API.

Resources:

Default Resources: orders

Configuration:

{
  "kind": "dlt_extract",
  "connectionRef": { "slug": "amazon-seller" },
  "definition": {
    "source": "amazon_sp",
    "resources": ["orders", "inventory"],
    "start_date": "2026-01-01"
  }
}

Credentials Required:

Marketplace IDs:

Incremental Extraction:

• Orders: Uses last-updated-date cursor for changes
• Inventory: Full sync (returns current state snapshot)

Adding New Sources

VirtuousAI's source registry is extensible. Contact support for custom source integrations or check the API for programmatic source registration.

Resource Ordering

When extracting multiple resources, VirtuousAI uses size-optimized ordering by default — extracting smaller, faster resources first before heavier ones.

Size Classes

Each resource is classified by expected data volume:

Ordering Behavior

Example: For Klaviyo with default resources, the extraction order is:

1. lists (S) — fast reference data
2. profiles (XL) — large but typically faster than events
3. events (XL) — highest volume, runs last

This ensures you get partial results quickly and heavy resources don't block lighter ones.

Preserving Custom Order

If you specify resources explicitly, your order is preserved:

{
  "definition": {
    "source": "klaviyo",
    "resources": ["events", "profiles", "lists"]
  }
}

This extracts in exactly that order, regardless of size class.

Rate Limiting

VirtuousAI implements pre-emptive rate limiting to avoid hitting vendor API limits entirely — rather than reacting to 429 errors after they occur.

How It Works

Each source has a dedicated rate limiter that enforces minimum intervals between API calls:

Amazon SP-API Rate Limits

Amazon's Reports API has strict sustained rate limits:

Runtime Impact: For historical syncs with many time slices, expect ~65 seconds per slice for report creation/retrieval. A 118-slice sync takes approximately 4-5 hours.

Klaviyo Rate Limits

Klaviyo has per-endpoint rate limits with burst capacity:

Why pre-emptive? Klaviyo's rate limits are aggressive. By staying well under the limit, we avoid 429 responses and the exponential backoff delays they cause.

Reactive Fallback

If a 429 response does occur (e.g., shared API key with other systems), the connector:

1. Reads the Retry-After header
2. Waits the specified duration (up to 5 minutes for rate limits)
3. Retries with a time budget (10 minutes total retry window)

{
  "error": {
    "code": "RATE_LIMITED",
    "message": "Rate limited by vendor API",
    "details": {
      "retry_after_seconds": 120,
      "budget_remaining_seconds": 480
    },
    "retryable": true
  }
}

Extraction Modes

Incremental (Default)

Only fetches new or modified records since last extraction:

{
  "definition": {
    "source": "shopify",
    "resources": ["orders"],
    "incremental": true
  }
}

How it works:

1. dlt maintains state with last cursor position
2. Each run queries only records after the cursor
3. Dramatically reduces API calls and processing time

Full Resync

Drops existing state and extracts all data:

{
  "definition": {
    "source": "shopify",
    "resources": ["orders"],
    "full_resync": true
  }
}

When to use:

• Schema changes in source system
• Data corruption in bronze layer
• Initial backfill with new date range

Output Format

Bronze Layer Structure

Extracted data is written to S3 as Parquet files:

s3://vai-bronze-{env}/
└── {tenant_id}/
    └── {connection_id}/
        └── {source}/
            └── {resource}/
                └── year=2026/
                    └── month=01/
                        └── day=22/
                            └── data.parquet

Metrics

Each extraction returns detailed metrics:

{
  "pipeline_name": "bronze__tenant123__conn456__shopify",
  "rows_extracted": 1250,
  "files_written": 3,
  "bytes_written": 2456789,
  "resources_loaded": ["orders", "order_line_items", "products"],
  "per_resource_metrics": {
    "orders": { "rows_extracted": 500, "files_written": 1 },
    "order_line_items": { "rows_extracted": 1500, "files_written": 1 },
    "products": { "rows_extracted": 250, "files_written": 1 }
  },
  "duration_seconds": 45.2,
  "state_fingerprint": "abc123..."
}

Error Handling

Retryable Errors

Non-Retryable Errors

Resume After Failure

If extraction fails mid-way, the next run resumes from where it left off:

{
  "result": {
    "resumed_from": ["orders", "order_line_items"],
    "resources_loaded": ["orders", "order_line_items", "products"]
  }
}

The progress is tracked via:

{
  "progress": {
    "completed_resources": ["orders", "order_line_items"],
    "total_resources": 3
  }
}

Heartbeat & Lease

Long-running extractions use the lease system:

The ExtractionHeartbeat context manager extends the lease during pipeline execution:

with ExtractionHeartbeat(run_id, pipeline_name):
    # dlt pipeline runs here
    # Heartbeat extends lease automatically

Scheduling Extractions

For regular data syncs, create an automation:

vai automations create \
  --name "Daily Shopify Sync" \
  --trigger-type schedule \
  --config '{
    "schedule": "0 6 * * *",
    "timezone": "UTC",
    "action": {
      "kind": "dlt_extract",
      "connectionRef": {"slug": "shopify"},
      "definition": {
        "source": "shopify",
        "resources": ["orders", "products"],
        "incremental": true
      }
    }
  }'

Best Practices

1. Use incremental mode — Only use full_resync when necessary
2. Limit resources — Extract only the resources you need
3. Set date ranges — Use start_date to limit historical data
4. Monitor metrics — Track rows_extracted to detect anomalies
5. Handle rate limits — The executor handles this, but schedule during off-peak hours

Next Steps